CLI improvements, mostly.

- Improve cli commands, in particular, `list-stored` and `get-eds`.
- Add more parsing and typing to `Machine.list_runs_in_storage` and `Machine.list_files` (breaking change if relying on atime/mtime/ctime being floats instead of datetime objects).
- Use setuptools_scm to write version number.

.gitignore

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: EUPL-1.2
 
-src/qslib/_version.py
+src/qslib/version.py
 
 # Temporary and binary files
 *~

CHANGELOG.md

@@ -6,6 +6,12 @@ SPDX-License-Identifier: EUPL-1.2
 
 # Changelog
 
+## Version 0.14.0
+
+- Improve cli commands, in particular, `list-stored` and `get-eds`.
+- Add more parsing and typing to `Machine.list_runs_in_storage` and `Machine.list_files` (breaking change if relying on atime/mtime/ctime being floats instead of datetime objects).
+- Use setuptools_scm to write version number.
+
 ## Version 0.13.0
 
 - Add `PlateSetup.from_picklist` method to generate a PlateSetup from a Kithairon PickList, if Kithairon is installed.

pyproject.toml

@@ -8,6 +8,8 @@ build-backend = "setuptools.build_meta"
 
 [tool.setuptools_scm]
 version_scheme = "no-guess-dev"
+version_file = "src/qslib/version.py"
+
 
 [tool.setuptools.packages.find]
 where = ["src"]
@@ -50,6 +52,7 @@ dependencies = [
     "typeguard >= 2",
     "nest_asyncio >= 1.5",
     "click >=8.0,<9.0",
+    "click-aliases",
     "typing-extensions >= 4",
     "toml",
     "pint",

src/qslib/cli.py

@@ -9,18 +9,22 @@
 from pathlib import Path
 
 import click
+from click_aliases import ClickAliasedGroup
 
 from qslib.common import Experiment, Machine
 from qslib.qs_is_protocol import (
     AccessLevelExceeded,
     AuthError,
     CommandError,
     InsufficientAccess,
+    NoMatch,
 )
 from qslib.scpi_commands import AccessLevel
+from .version import __version__
 
 
-@click.group()
+@click.group(cls=ClickAliasedGroup)
+@click.version_option(__version__, "--version", "-V")
 def cli():
     pass
 
@@ -278,20 +282,68 @@ def machine_status(machine: str) -> None:
 
     del m
 
+@cli.command(name="list-stored")
+@click.argument("machine_path", nargs=-1)
+@click.option("-v", "--verbose", is_flag=True, help="Show detailed file information")
+@click.option("-s", "--sort", type=click.Choice(['name', 'time', 'size', 'created']), default='name',
+              help="Sort by name, modification time, creation time, or file size")
+@click.option("-r", "--reverse", is_flag=True, help="Reverse the sort order")
+@click.option("-U", "--created", is_flag=True, help="Use creation time instead of modification time")
+def list_stored(machine_path: tuple[str, ...], verbose: bool, sort: str, reverse: bool, created: bool) -> None:
+    """List experiments stored on a machine.
+    
+    MACHINE_PATH can be either just a machine name/address, or machine:pattern where
+    pattern is used to filter results using shell-style wildcards (*, ?, etc)."""
+
+    for mp in machine_path:
+        if mp.startswith("["):
+            # Handle IPv6 address in brackets
+            closing_bracket = mp.find("]")
+            if closing_bracket == -1:
+                raise click.UsageError("Missing closing bracket for IPv6 address")
+
+            machine = mp[1:closing_bracket]
+            if len(mp) > closing_bracket + 1:
+                if mp[closing_bracket + 1] != ":":
+                    raise click.UsageError("Expected ':' after IPv6 address")
+                pattern = mp[closing_bracket + 2:]
+            else:
+                pattern = "*"
+        elif ":" in mp:
+            machine, pattern = mp.split(":", 1)
+        else:
+            machine = mp
+            pattern = "*"
 
-@cli.command()
-@click.argument("machine")
-def list_stored(machine: str) -> None:
-    """List experiments stored on a machine."""
-    m = Machine(
-        machine,
-        max_access_level="Observer",
-    )
-
-    with m:
-        for f in m.list_runs_in_storage():
-            click.echo(f)
+        m = Machine(
+            machine,
+            max_access_level="Observer",
+        )
 
+        with m:
+            if len(machine_path) > 1:
+                click.echo(f"{machine}:")
+            try:
+                files = m.list_runs_in_storage(pattern, verbose=True)
+
+                # Sort the files according to the specified criteria
+                if sort == 'name':
+                    files.sort(key=lambda x: x['path'], reverse=reverse)
+                elif sort == 'time':
+                    files.sort(key=lambda x: x['mtime'] if not created else x['ctime'], reverse=reverse)
+                elif sort == 'size':
+                    files.sort(key=lambda x: x['size'], reverse=reverse)
+
+            except NoMatch:
+                click.echo(f"No runs found matching {pattern}")
+                continue
+            for f in files:
+                if verbose:
+                    size_str = f"{f['size']/1000/1000:.1f}M" if f['size'] > 1000*1000 else f"{f['size']/1000:.1f}k"
+                    time_str = f['ctime'].strftime("%Y-%m-%d %H:%M") if created else f['mtime'].strftime("%Y-%m-%d %H:%M")
+                    click.echo(f"{size_str:>8}  {time_str}  {f['path']}")
+                else:
+                    click.echo(f['path'])
 
 @cli.command()
 @click.argument("machine")
@@ -315,6 +367,48 @@ def copy(
         exp = Experiment.from_machine(m, experiment)
         exp.save_file(output, update_files=False)
 
+@cli.command()
+@click.argument("path", nargs=-1)
+@click.option("-f", "--force/--no-force", default=False, help="Overwrite existing files")
+@click.option("-q", "--quiet/--no-quiet", default=False, help="Suppress output messages")
+@click.option("-o", "--output-dir", type=click.Path(), help="Output directory for saved files")
+def get_eds(path: tuple[str, ...], regex: bool, force: bool, quiet: bool, output_dir: str | None) -> None:
+    """Get all experiments matching a name from the machine.
+    
+    Each argument should be of the form machine:name, where machine is the machine address
+    and name is the glob pattern to match experiment names against.
+    """
+    if output_dir is not None:
+        output_path = Path(output_dir)
+        output_path.mkdir(parents=True, exist_ok=True)
+    else:
+        output_path = Path(".")
+
+    for machine_name in path:
+        try:
+            machine, name = machine_name.split(":", 1)
+        except ValueError:
+            click.echo(f"Error: argument '{machine_name}' must be in format 'machine:name'", err=True)
+            continue
+
+        m = Machine(machine, max_access_level="Observer")
+        runs = m.list_runs_in_storage(name)
+
+        for r in runs:
+            click.echo(f"Getting {r} from {machine}: ", nl=False)
+            exp = Experiment.from_machine(m, r)
+
+            oname = exp.runtitle_safe
+
+            output_file = output_path / (oname + ".eds")
+            if not force and output_file.exists():
+                if not quiet:
+                    click.echo(f"skipping, {output_file} already exists")
+                continue
+
+            if not quiet:
+                click.echo(f"saving to {output_file}")
+            exp.save_file(output_file, update_files=False)
 
 @dataclass
 class OutP:

src/qslib/machine.py

@@ -6,14 +6,15 @@
 
 import asyncio
 import base64
+from datetime import datetime
 import logging
 import re
 import zipfile
 from asyncio.futures import Future
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import wraps
-from typing import IO, TYPE_CHECKING, Any, Generator, Literal, cast, overload
+from typing import IO, TYPE_CHECKING, Any, Generator, Literal, cast, overload, TypedDict
 
 import nest_asyncio
 
@@ -22,14 +23,15 @@
 
 from ._util import _unwrap_tags
 from .protocol import Protocol
-from .qsconnection_async import QSConnectionAsync
+from .qsconnection_async import FileListInfo, QSConnectionAsync
 
 nest_asyncio.apply()
 
 from .base import MachineStatus, RunStatus  # noqa: E402
 
 log = logging.getLogger(__name__)
 
+
 if TYPE_CHECKING:  # pragma: no cover
     import matplotlib.pyplot as plt  # noqa: F401
 
@@ -326,7 +328,7 @@ def list_files(
         leaf: str = "FILE",
         verbose: Literal[True],
         recursive: bool = False,
-    ) -> list[dict[str, Any]]: ...
+    ) -> list[FileListInfo]: ...
 
     @overload
     def list_files(
@@ -336,7 +338,9 @@ def list_files(
         leaf: str = "FILE",
         verbose: Literal[False] = False,
         recursive: bool = False,
-    ) -> list[str]: ...
+    ) -> list[str] | list[FileListInfo]: ...
+
+
 
     @_ensure_connection(AccessLevel.Observer)
     def list_files(
@@ -346,7 +350,7 @@ def list_files(
         leaf: str = "FILE",
         verbose: bool = False,
         recursive: bool = False,
-    ) -> list[str] | list[dict[str, Any]]:
+    ) -> list[str] | list[FileListInfo]:
         loop = asyncio.get_event_loop()
         return loop.run_until_complete(
             self.connection.list_files(
@@ -391,7 +395,7 @@ def write_file(self, path: str, data: str | bytes) -> None:
         )
 
     @_ensure_connection(AccessLevel.Observer)
-    def list_runs_in_storage(self) -> list[str]:
+    def list_runs_in_storage(self, glob: str = "*", verbose: bool = False) -> list[str]:
         """List runs in machine storage.
 
         Returns
@@ -401,11 +405,17 @@ def list_runs_in_storage(self) -> list[str]:
             (to open as :any`Experiment`) or save_run_from_storage
             (to download and save it without opening.)
         """
-        x = self.run_command("FILE:LIST? public_run_complete:")
-        a = x.split("\n")[1:-1]
-        return [
-            re.sub("^public_run_complete:", "", s)[:-4] for s in a if s.endswith(".eds")
-        ]
+        if not glob.endswith("eds"):
+            glob = f"{glob}eds"
+        a = self.list_files(f"public_run_complete:{glob}", verbose=verbose)
+        if not verbose:
+            return [
+                re.sub("^public_run_complete:", "", s)[:-4] for s in a
+            ]
+        else:
+            for e in a:
+                e["path"] = re.sub("^public_run_complete:", "", e["path"])[:-4]
+            return a
 
     @_ensure_connection(AccessLevel.Observer)
     def load_run_from_storage(self, path: str) -> "Experiment":  # type: ignore

src/qslib/qsconnection_async.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import base64
+from datetime import datetime
 import hmac
 import io
 import logging
@@ -15,14 +16,23 @@
 import xml.etree.ElementTree as ET
 import zipfile
 from dataclasses import dataclass
-from typing import Any, Dict, List, Literal, Optional, Union, cast, overload
+from typing import Any, Dict, List, Literal, Optional, Union, cast, overload, TypedDict
 
 import pandas as pd
 
 from . import data
 from .qs_is_protocol import CommandError, Error, NoMatch, QS_IS_Protocol
 from .scpi_commands import AccessLevel, ArgList, SCPICommand
 
+class FileListInfo(TypedDict):
+    """Information about a file when verbose=True"""
+    path: str
+    type: str
+    size: int
+    mtime: datetime
+    atime: datetime
+    ctime: datetime
+
 log = logging.getLogger(__name__)
 
 
@@ -124,7 +134,7 @@ async def list_files(
         leaf: str = "FILE",
         verbose: Literal[True],
         recursive: bool = False,
-    ) -> list[dict[str, Any]]: ...
+    ) -> list[FileListInfo]: ...
 
     @overload
     async def list_files(
@@ -144,7 +154,7 @@ async def list_files(
         leaf: str = "FILE",
         verbose: bool = False,
         recursive: bool = False,
-    ) -> list[str] | list[dict[str, Any]]: ...
+    ) -> list[str] | list[FileListInfo]: ...
 
     async def list_files(
         self,
@@ -153,7 +163,7 @@ async def list_files(
         leaf: str = "FILE",
         verbose: bool = False,
         recursive: bool = False,
-    ) -> list[str] | list[dict[str, Any]]:
+    ) -> list[str] | list[FileListInfo]:
         if not verbose:
             if recursive:
                 raise NotImplementedError
@@ -162,7 +172,7 @@ async def list_files(
             v = (await self.run_command(f"{leaf}:LIST? -verbose {path}")).split("\n")[
                 1:-1
             ]
-            ret: list[dict[str, str | float | int]] = []
+            ret: list[FileListInfo] = []
             for x in v:
                 rm = re.match(
                     r'"([^"]+)" -type=(\S+) -size=(\S+) -mtime=(\S+) -atime=(\S+) -ctime=(\S+)$',
@@ -178,9 +188,9 @@ async def list_files(
                     d["path"] = rm.group(1)
                     d["type"] = rm.group(2)
                     d["size"] = int(rm.group(3))
-                    d["mtime"] = float(rm.group(4))
-                    d["atime"] = float(rm.group(5))
-                    d["ctime"] = float(rm.group(6))
+                    d["mtime"] = datetime.fromtimestamp(float(rm.group(4)))
+                    d["atime"] = datetime.fromtimestamp(float(rm.group(5)))
+                    d["ctime"] = datetime.fromtimestamp(float(rm.group(6)))
                 if d["type"] == "folder" and recursive:
                     ret += await self.list_files(
                         cast(str, d["path"]), leaf=leaf, verbose=True, recursive=True