instantiate streamdaq from id in cli and tests

miniscope_io/cli/common.py

@@ -0,0 +1,31 @@
+"""
+Shared CLI utils
+"""
+
+from os import PathLike
+from pathlib import Path
+from typing import Optional
+
+from click import Context, Parameter, ParamType
+
+
+class ConfigIDOrPath(ParamType):
+    """
+    A custom click type to accept either a config `id` or a path
+    as input, resolving relative paths first against
+    the current working directory and second against the user config directory.
+    """
+
+    name = "config-id-or-path"
+
+    def convert(
+        self, value: str | PathLike[str], param: Optional[Parameter], ctx: Optional[Context]
+    ) -> str | Path:
+        """
+        If something looks like a yaml file, return as a path, otherwise return unchanged.
+
+        Don't do validation here, the Config model will handle that on instantiation.
+        """
+        if value.endswith(".yaml") or value.endswith(".yml"):
+            value = Path(value)
+        return value

miniscope_io/cli/stream.py

@@ -8,6 +8,7 @@
 
 import click
 
+from miniscope_io.cli.common import ConfigIDOrPath
 from miniscope_io.stream_daq import StreamDaq
 
 
@@ -24,8 +25,13 @@ def _common_options(fn: Callable) -> Callable:
         "-c",
         "--device_config",
         required=True,
-        help="Path to device config YAML file for streamDaq (see models.stream.StreamDevConfig)",
-        type=click.Path(exists=True),
+        help=(
+            "Either a config `id` or a path to device config YAML file for streamDaq "
+            "(see models.stream.StreamDevConfig). If path is relative, treated as "
+            "relative to the current directory, and then if no matching file is found, "
+            "relative to the user `config_dir` (see `mio config --help`)."
+        ),
+        type=ConfigIDOrPath,
     )(fn)
     return fn
 

miniscope_io/models/config.py

@@ -102,7 +102,7 @@ def folder_exists(cls, v: Path) -> Path:
     @model_validator(mode="after")
     def paths_relative_to_basedir(self) -> "Config":
         """If relative paths are given, make them absolute relative to ``base_dir``"""
-        paths = ("log_dir",)
+        paths = ("log_dir", "config_dir")
         for path_name in paths:
             path = getattr(self, path_name)  # type: Path
             if not path.is_absolute():

miniscope_io/models/mixins.py

@@ -15,7 +15,7 @@
 
 from miniscope_io import CONFIG_DIR, Config
 from miniscope_io.logging import init_logger
-from miniscope_io.types import PythonIdentifier
+from miniscope_io.types import ConfigID, ConfigSource, PythonIdentifier, valid_config_id
 
 T = TypeVar("T")
 
@@ -82,20 +82,12 @@ class ConfigYAMLMixin(BaseModel, YAMLMixin):
      at the top of the file.
     """
 
-    id: str
+    id: ConfigID
     mio_model: PythonIdentifier = Field(None, validate_default=True)
     mio_version: str = version("miniscope-io")
 
     HEADER_FIELDS: ClassVar[tuple[str]] = ("id", "mio_model", "mio_version")
 
-    @field_validator("mio_model", mode="before")
-    @classmethod
-    def fill_mio_model(cls, v: Optional[str]) -> PythonIdentifier:
-        """Get name of instantiating model, if not provided"""
-        if v is None:
-            v = cls._model_name()
-        return v
-
     @classmethod
     def from_yaml(cls: Type[T], file_path: Union[str, Path]) -> T:
         """Instantiate this class by passing the contents of a yaml file as kwargs"""
@@ -117,16 +109,7 @@ def from_yaml(cls: Type[T], file_path: Union[str, Path]) -> T:
         return instance
 
     @classmethod
-    @property
-    def config_sources(cls: Type[T]) -> List[Path]:
-        """
-        Directories to search for config files, in order of priority
-        such that earlier sources are preferred over later sources.
-        """
-        return [Config().config_dir, CONFIG_DIR]
-
-    @classmethod
-    def from_id(cls: Type[T], id: str) -> T:
+    def from_id(cls: Type[T], id: ConfigID) -> T:
         """
         Instantiate a model from a config `id` specified in one of the .yaml configs in
         either the user :attr:`.Config.config_dir` or the packaged ``config`` dir.
@@ -149,6 +132,60 @@ def from_id(cls: Type[T], id: str) -> T:
                 continue
         raise KeyError(f"No config with id {id} found in {Config().config_dir}")
 
+    @classmethod
+    def from_any(cls: Type[T], source: Union[ConfigSource, T]) -> T:
+        """
+        Try and instantiate a config model from any supported constructor.
+
+        Args:
+            source (:class:`.ConfigID`, :class:`.Path`, :class:`.PathLike[str]`):
+                Either
+
+                * the ``id`` of a config file in the user configs directory or builtin
+                * a relative ``Path`` to a config file, relative to the current working directory
+                * a relative ``Path`` to a config file, relative to the user config directory
+                * an absolute ``Path`` to a config file
+                * an instance of the class to be constructed (returned unchanged)
+
+        """
+        if isinstance(source, cls):
+            return source
+        elif valid_config_id(source):
+            return cls.from_id(source)
+        else:
+            source = Path(source)
+            if source.suffix in (".yaml", ".yml"):
+                if source.exists():
+                    # either relative to cwd or absolute
+                    return cls.from_yaml(source)
+                elif (
+                    not source.is_absolute()
+                    and (user_source := Config().config_dir / source).exists()
+                ):
+                    return cls.from_yaml(user_source)
+
+        raise ValueError(
+            f"Instance of config model {cls.__name__} could not be instantiated from "
+            f"{source} - id or file not found, or type not supported"
+        )
+
+    @field_validator("mio_model", mode="before")
+    @classmethod
+    def fill_mio_model(cls, v: Optional[str]) -> PythonIdentifier:
+        """Get name of instantiating model, if not provided"""
+        if v is None:
+            v = cls._model_name()
+        return v
+
+    @classmethod
+    @property
+    def config_sources(cls: Type[T]) -> List[Path]:
+        """
+        Directories to search for config files, in order of priority
+        such that earlier sources are preferred over later sources.
+        """
+        return [Config().config_dir, CONFIG_DIR]
+
     def _dump_data(self, **kwargs: Any) -> dict:
         """Ensure that header is prepended to model data"""
         return {**self._yaml_header(self), **super()._dump_data(**kwargs)}

miniscope_io/models/stream.py

@@ -10,7 +10,7 @@
 from miniscope_io import DEVICE_DIR
 from miniscope_io.models import MiniscopeConfig
 from miniscope_io.models.buffer import BufferHeader, BufferHeaderFormat
-from miniscope_io.models.mixins import YAMLMixin
+from miniscope_io.models.mixins import ConfigYAMLMixin
 from miniscope_io.models.sinks import CSVWriterConfig, StreamPlotterConfig
 
 
@@ -168,7 +168,7 @@ class StreamDevRuntime(MiniscopeConfig):
     )
 
 
-class StreamDevConfig(MiniscopeConfig, YAMLMixin):
+class StreamDevConfig(MiniscopeConfig, ConfigYAMLMixin):
     """
     Format model used to parse DAQ configuration yaml file (examples are in ./config)
     The model attributes are key-value pairs needed for reconstructing frames from data streams.

miniscope_io/stream_daq.py

@@ -27,6 +27,7 @@
     StreamDevConfig,
 )
 from miniscope_io.plots.headers import StreamPlotter
+from miniscope_io.types import ConfigSource
 
 HAVE_OK = False
 ok_error = None
@@ -79,8 +80,8 @@ class StreamDaq:
 
     def __init__(
         self,
-        device_config: Union[StreamDevConfig, Path],
-        header_fmt: Union[StreamBufferHeaderFormat, str] = "stream-buffer-header",
+        device_config: Union[StreamDevConfig, ConfigSource],
+        header_fmt: Union[StreamBufferHeaderFormat, ConfigSource] = "stream-buffer-header",
     ) -> None:
         """
         Constructer for the class.
@@ -97,11 +98,10 @@ def __init__(
             Header format used to parse information from buffer header,
             by default `MetadataHeaderFormat()`.
         """
-        if isinstance(device_config, (str, Path)):
-            device_config = StreamDevConfig.from_yaml(device_config)
 
         self.logger = init_logger("streamDaq")
-        self.config = device_config
+        self.config = StreamDevConfig.from_any(device_config)
+        self.header_fmt = StreamBufferHeaderFormat.from_any(header_fmt)
         if isinstance(header_fmt, str):
             self.header_fmt = StreamBufferHeaderFormat.from_id(header_fmt)
         elif isinstance(header_fmt, StreamBufferHeaderFormat):

miniscope_io/types.py

@@ -2,18 +2,42 @@
 Type and type annotations
 """
 
+import re
 import sys
-from typing import Annotated, Tuple, Union
+from os import PathLike
+from pathlib import Path
+from typing import Annotated, Any, Tuple, Union
 
-from pydantic import AfterValidator
+from pydantic import AfterValidator, Field
 
 if sys.version_info < (3, 10):
-    from typing_extensions import TypeAlias
-else:
+    from typing_extensions import TypeAlias, TypeIs
+elif sys.version_info < (3, 13):
     from typing import TypeAlias
 
+    from typing_extensions import TypeIs
+else:
+    from typing import TypeAlias, TypeIs
+
+CONFIG_ID_PATTERN = r"[\w\-\/#]+"
+"""
+Any alphanumeric string (\w), as well as
+- ``-``
+- ``/``
+- ``#``
+(to allow hierarchical IDs as well as fragment IDs).
+
+Specficially excludes ``.`` to avoid confusion between IDs, paths, and python module names
+
+May be made less restrictive in the future, will not be made more restrictive.
+"""
+
 
 def _is_identifier(val: str) -> str:
+    # private validation method to validate the parts of a fully-qualified python identifier
+    # defined first and not made public bc used as a validator,
+    # distinct from a boolean "is_{x}" check
+
     for part in val.split("."):
         assert part.isidentifier(), f"{part} is not a valid python identifier within {val}"
     return val
@@ -25,3 +49,18 @@ def _is_identifier(val: str) -> str:
 A valid python identifier, including globally namespace pathed like 
 module.submodule.ClassName
 """
+ConfigID: TypeAlias = Annotated[str, Field(pattern=CONFIG_ID_PATTERN)]
+"""
+A string that refers to a config file by the ``id`` field in that config
+"""
+ConfigSource: TypeAlias = Union[Path, PathLike[str], ConfigID]
+"""
+Union of all types of config sources
+"""
+
+
+def valid_config_id(val: Any) -> TypeIs[ConfigID]:
+    """
+    Checks whether a string is a valid config id.
+    """
+    return bool(re.fullmatch(CONFIG_ID_PATTERN, val))

pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "rich>=13.6.0",
     "pyyaml>=6.0.1",
     "click>=8.1.7",
+    'typing-extensions>=4.10.0; python_version<"3.13"'
 ]
 
 readme = "README.md"

tests/conftest.py

@@ -6,6 +6,10 @@
 import pytest
 import yaml
 
+from miniscope_io.models.mixins import ConfigYAMLMixin
+
+from .fixtures import *
+
 DATA_DIR = Path(__file__).parent / "data"
 CONFIG_DIR = DATA_DIR / "config"
 MOCK_DIR = Path(__file__).parent / "mock"
@@ -33,6 +37,21 @@ def mock_okdev(monkeypatch):
     monkeypatch.setattr(stream_daq, "okDev", okDevMock)
 
 
+@pytest.fixture(scope="session", autouse=True)
+def mock_config_source(monkeypatch_session):
+    """
+    Add the `tests/data/config` directory to the config sources for the entire testing session
+    """
+    current_sources = ConfigYAMLMixin.config_sources
+
+    @classmethod
+    @property
+    def _config_sources(cls: type[ConfigYAMLMixin]) -> list[Path]:
+        return [CONFIG_DIR, *current_sources]
+
+    monkeypatch_session.setattr(ConfigYAMLMixin, "config_sources", _config_sources)
+
+
 @pytest.fixture()
 def set_okdev_input(monkeypatch):
     """

tests/data/config/preamble_hex.yml

@@ -1,3 +1,7 @@
+id: test-wireless-preamble-hex
+mio_model: miniscope_io.models.stream.StreamDevConfig
+mio_version: "v5.0.0"
+
 # capture device. "OK" (Opal Kelly) or "UART"
 device: "OK"
 

tests/data/config/stream_daq_test_200px.yml

@@ -1,3 +1,7 @@
+id: test-wireless-200px
+mio_model: miniscope_io.models.stream.StreamDevConfig
+mio_version: "v5.0.0"
+
 # capture device. "OK" (Opal Kelly) or "UART"
 device: "OK"
 

tests/data/config/wireless_example.yml

@@ -1,3 +1,7 @@
+id: test-wireless-example
+mio_model: miniscope_io.models.stream.StreamDevConfig
+mio_version: "v5.0.0"
+
 # capture device. "OK" (Opal Kelly) or "UART"
 device: "OK"
 

tests/fixtures.py

@@ -3,6 +3,7 @@
 
 import pytest
 import yaml
+from _pytest.monkeypatch import MonkeyPatch
 
 from miniscope_io.io import SDCard
 from miniscope_io.models.data import Frames
@@ -87,3 +88,13 @@ def _yaml_config(id: str, data: dict, path: Optional[Path] = None) -> Path:
         return path
 
     return _yaml_config
+
+
+@pytest.fixture(scope="session")
+def monkeypatch_session() -> MonkeyPatch:
+    """
+    Monkeypatch you can use at the session scope!
+    """
+    mpatch = MonkeyPatch()
+    yield mpatch
+    mpatch.undo()