implement config from id in streaming daq

miniscope_io/data/config/wirefree/sd_layout_battery.yaml → miniscope_io/data/config/wirefree/sd-layout-battery.yaml

@@ -1,5 +1,5 @@
 id: wirefree-sd-layout-battery
-model: miniscope_io.models.sdcard.SDLayout
+mio_model: miniscope_io.models.sdcard.SDLayout
 mio_version: v5.0.0
 sectors:
   header: 1022

miniscope_io/data/config/wirefree/sd_layout.yaml → miniscope_io/data/config/wirefree/sd-layout.yaml

@@ -1,5 +1,5 @@
 id: wirefree-sd-layout
-model: miniscope_io.models.sdcard.SDLayout
+mio_model: miniscope_io.models.sdcard.SDLayout
 mio_version: v5.0.0
 sectors:
   header: 1022

miniscope_io/data/config/wireless/stream-buffer-header.yaml

@@ -1,5 +1,5 @@
-id: "stream-buffer-header"
-model: "miniscope_io.models.stream.StreamBufferHeaderFormat"
+id: stream-buffer-header
+mio_model: miniscope_io.models.stream.StreamBufferHeaderFormat
 mio_version: "v5.0.0"
 linked_list: 0
 frame_num: 1

miniscope_io/data/config/WLMS_v02_200px.yml → miniscope_io/data/config/wireless/wireless-200px.yml

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

miniscope_io/models/buffer.py

@@ -6,10 +6,12 @@
 from collections.abc import Sequence
 from typing import Type, TypeVar
 
+from miniscope_io.logging import init_logger
 from miniscope_io.models import Container, MiniscopeConfig
+from miniscope_io.models.mixins import ConfigYAMLMixin
 
 
-class BufferHeaderFormat(MiniscopeConfig):
+class BufferHeaderFormat(MiniscopeConfig, ConfigYAMLMixin):
     """
     Format model used to parse header at the beginning of every buffer.
 
@@ -86,9 +88,13 @@ def from_format(
         """
 
         header_data = dict()
-        for hd, header_index in format.model_dump().items():
+        for hd, header_index in format.model_dump(exclude=set(format.HEADER_FIELDS)).items():
             if header_index is not None:
-                header_data[hd] = vals[header_index]
+                try:
+                    header_data[hd] = vals[header_index]
+                except IndexError:
+                    init_logger("BufferHeader").exception(f"{header_index}")
+                    raise
 
         if construct:
             return cls.model_construct(**header_data)

miniscope_io/models/mixins.py

@@ -4,16 +4,18 @@
 """
 
 import re
+import shutil
 from importlib.metadata import version
 from itertools import chain
 from pathlib import Path
-from typing import Any, List, Literal, Optional, Type, TypeVar, Union, overload
+from typing import Any, ClassVar, List, Literal, Optional, Type, TypeVar, Union, overload
 
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, ValidationError, field_validator
 
 from miniscope_io import CONFIG_DIR, Config
 from miniscope_io.logging import init_logger
+from miniscope_io.types import PythonIdentifier
 
 T = TypeVar("T")
 
@@ -47,45 +49,84 @@ def to_yaml(self, path: Optional[Path] = None, **kwargs: Any) -> str:
         Dump the contents of this class to a yaml file, returning the
         contents of the dumped string
         """
-        data = self._dump_data(**kwargs)
-        data_str = yaml.dump(data, Dumper=YamlDumper, sort_keys=False)
-
+        data_str = self.to_yamls(**kwargs)
         if path:
             with open(path, "w") as file:
                 file.write(data_str)
 
         return data_str
 
+    def to_yamls(self, **kwargs: Any) -> str:
+        """
+        Dump the contents of this class to a yaml string
+
+        Args:
+            **kwargs: passed to :meth:`.BaseModel.model_dump`
+        """
+        data = self._dump_data(**kwargs)
+        return yaml.dump(data, Dumper=YamlDumper, sort_keys=False)
+
     def _dump_data(self, **kwargs: Any) -> dict:
         data = self.model_dump(**kwargs) if isinstance(self, BaseModel) else self.__dict__
         return data
 
 
-class ConfigYAMLMixin(YAMLMixin):
+class ConfigYAMLMixin(BaseModel, YAMLMixin):
     """
     Yaml Mixin class that always puts a header consisting of
 
      * `id` - unique identifier for this config
-     * `model` - fully-qualified module path to model class
+     * `mio_model` - fully-qualified module path to model class
      * `mio_version` - version of miniscope-io when this model was created
+
+     at the top of the file.
     """
 
-    HEADER_FIELDS = {"id", "model", "mio_version"}
+    id: str
+    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"""
         with open(file_path) as file:
             config_data = yaml.safe_load(file)
-        instance = cls(**config_data)
 
         # fill in any missing fields in the source file needed for a header
-        cls._complete_header(instance, config_data, file_path)
+        config_data = cls._complete_header(config_data, file_path)
+        try:
+            instance = cls(**config_data)
+        except ValidationError:
+            if (backup_path := file_path.with_suffix(".yaml.bak")).exists():
+                init_logger("config").debug(
+                    f"Model instantiation failed, restoring modified backup from {backup_path}..."
+                )
+                shutil.copy(backup_path, file_path)
+            raise
 
         return instance
 
     @classmethod
-    def from_id(cls, id: str) -> T:
+    @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:
         """
         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.
@@ -95,10 +136,14 @@ def from_id(cls, id: str) -> T:
             this method does not yet validate that the config matches the model loading it
 
         """
-        for config_file in chain(Config().config_dir.rglob("*.y*ml"), CONFIG_DIR.rglob("*.y*ml")):
+        globs = [src.rglob("*.y*ml") for src in cls.config_sources]
+        for config_file in chain(*globs):
             try:
                 file_id = yaml_peek("id", config_file)
                 if file_id == id:
+                    init_logger("config").debug(
+                        "Model for %s found at %s", cls._model_name(), config_file
+                    )
                     return cls.from_yaml(config_file)
             except KeyError:
                 continue
@@ -109,30 +154,60 @@ def _dump_data(self, **kwargs: Any) -> dict:
         return {**self._yaml_header(self), **super()._dump_data(**kwargs)}
 
     @classmethod
-    def _yaml_header(cls, instance: T) -> dict:
+    def _model_name(cls) -> PythonIdentifier:
+        return f"{cls.__module__}.{cls.__name__}"
+
+    @classmethod
+    def _yaml_header(cls, instance: Union[T, dict]) -> dict:
+        if isinstance(instance, dict):
+            model_id = instance.get("id", None)
+            mio_model = instance.get("mio_model", cls._model_name())
+            mio_version = instance.get("mio_version", version("miniscope_io"))
+        else:
+            model_id = getattr(instance, "id", None)
+            mio_model = getattr(instance, "mio_model", cls._model_name())
+            mio_version = getattr(instance, "mio_version", version("miniscope_io"))
+
+        if model_id is None:
+            # if missing an id, try and recover with model default cautiously
+            # so we throw the exception during validation and not here, for clarity.
+            model_id = getattr(cls.model_fields.get("id", None), "default", None)
+            if type(model_id).__name__ == "PydanticUndefinedType":
+                model_id = None
+
         return {
-            "id": instance.id,
-            "model": f"{cls.__module__}.{cls.__name__}",
-            "mio_version": version("miniscope_io"),
+            "id": model_id,
+            "mio_model": mio_model,
+            "mio_version": mio_version,
         }
 
     @classmethod
-    def _complete_header(
-        cls: Type[T], instance: T, data: dict, file_path: Union[str, Path]
-    ) -> None:
+    def _complete_header(cls: Type[T], data: dict, file_path: Union[str, Path]) -> dict:
         """fill in any missing fields in the source file needed for a header"""
 
-        missing_fields = cls.HEADER_FIELDS - set(data.keys())
-        if missing_fields:
+        missing_fields = set(cls.HEADER_FIELDS) - set(data.keys())
+        keys = tuple(data.keys())
+        out_of_order = len(keys) >= 3 and keys[0:3] != cls.HEADER_FIELDS
+
+        if missing_fields or out_of_order:
+            if missing_fields:
+                msg = f"Missing required header fields {missing_fields} in config model "
+                f"{str(file_path)}. Updating file (preserving backup)..."
+            else:
+                msg = f"Header keys were present, but either not at the start of {str(file_path)} "
+                "or in out of order. Updating file (preserving backup)..."
             logger = init_logger(cls.__name__)
-            logger.warning(
-                f"Missing required header fields {missing_fields} in config model "
-                f"{str(file_path)}. Updating file..."
-            )
-            header = cls._yaml_header(instance)
+            logger.warning(msg)
+            logger.debug(data)
+
+            header = cls._yaml_header(data)
             data = {**header, **data}
+            if CONFIG_DIR not in file_path.parents:
+                shutil.copy(file_path, file_path.with_suffix(".yaml.bak"))
             with open(file_path, "w") as yfile:
-                yaml.safe_dump(data, yfile, sort_keys=False)
+                yaml.dump(data, yfile, Dumper=YamlDumper, sort_keys=False)
+
+        return data
 
 
 @overload
@@ -177,9 +252,13 @@ def yaml_peek(
         str
     """
     if root:
-        pattern = re.compile(rf"^(?P<key>{key}):\s*\"*\'*(?P<value>\S.*?)\"*\'*$")
+        pattern = re.compile(
+            rf"^(?P<key>{key}):\s*\"*\'*(?P<value>\S.*?)\"*\'*$", flags=re.MULTILINE
+        )
     else:
-        pattern = re.compile(rf"^\s*(?P<key>{key}):\s*\"*\'*(?P<value>\S.*?)\"*\'*$")
+        pattern = re.compile(
+            rf"^\s*(?P<key>{key}):\s*\"*\'*(?P<value>\S.*?)\"*\'*$", flags=re.MULTILINE
+        )
 
     res = None
     if first:

miniscope_io/models/sdcard.py

@@ -92,6 +92,8 @@ class SDBufferHeaderFormat(BufferHeaderFormat):
     Positions in the header for each frame
     """
 
+    id: str = "sd-buffer-header"
+
     length: int = 0
     linked_list: int = 1
     frame_num: int = 2

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 ConfigYAMLMixin, YAMLMixin
+from miniscope_io.models.mixins import YAMLMixin
 from miniscope_io.models.sinks import CSVWriterConfig, StreamPlotterConfig
 
 
@@ -61,7 +61,7 @@ def scale_input_voltage(self, voltage_raw: float) -> float:
         return voltage_raw / 2**self.bitdepth * self.ref_voltage * self.vin_div_factor
 
 
-class StreamBufferHeaderFormat(BufferHeaderFormat, ConfigYAMLMixin):
+class StreamBufferHeaderFormat(BufferHeaderFormat):
     """
     Refinements of :class:`.BufferHeaderFormat` for
     :class:`~miniscope_io.stream_daq.StreamDaq`

miniscope_io/types.py

@@ -2,6 +2,26 @@
 Type and type annotations
 """
 
-from typing import Tuple, Union
+import sys
+from typing import Annotated, Tuple, Union
 
-Range = Union[Tuple[int, int], Tuple[float, float]]
+from pydantic import AfterValidator
+
+if sys.version_info < (3, 10):
+    from typing_extensions import TypeAlias
+else:
+    from typing import TypeAlias
+
+
+def _is_identifier(val: str) -> str:
+    for part in val.split("."):
+        assert part.isidentifier(), f"{part} is not a valid python identifier within {val}"
+    return val
+
+
+Range: TypeAlias = Union[Tuple[int, int], Tuple[float, float]]
+PythonIdentifier: TypeAlias = Annotated[str, AfterValidator(_is_identifier)]
+"""
+A valid python identifier, including globally namespace pathed like 
+module.submodule.ClassName
+"""

pyproject.toml

@@ -118,7 +118,7 @@ omit = [
 ]
 
 [tool.ruff]
-target-version = "py311"
+target-version = "py39"
 include = ["miniscope_io/**/*.py", "pyproject.toml"]
 exclude = ["docs", "tests", "miniscope_io/vendor", "noxfile.py"]
 line-length = 100