bunch of typing, etc, to make CIs happy-ish

Author: drammock

pyproject.toml

@@ -21,11 +21,8 @@ classifiers = [
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3 :: Only",
-  "Programming Language :: Python :: 3.8",
-  "Programming Language :: Python :: 3.9",
-  "Programming Language :: Python :: 3.10",
-  "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
   "Topic :: Scientific/Engineering",
   "Typing :: Typed",
 ]
@@ -87,7 +84,7 @@ report.exclude_also = [
 
 [tool.mypy]
 files = ["src", "tests"]
-python_version = "3.8"
+python_version = "3.12"
 warn_unused_configs = true
 strict = true
 enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"]
@@ -143,7 +140,7 @@ isort.required-imports = ["from __future__ import annotations"]
 
 
 [tool.pylint]
-py-version = "3.8"
+py-version = "3.12"
 ignore-paths = [".*/_version.py"]
 reports.output-format = "colorized"
 similarities.ignore-imports = "yes"

src/streamsync/__init__.py

@@ -7,6 +7,6 @@
 from __future__ import annotations
 
 # from ._version import version as __version__
-from .streamsync import StreamSync
+from .streamsync import StreamSync, extract_audio_from_video
 
-__all__ = ["__version__", "StreamSync"]
+__all__ = ["__version__", "StreamSync", "extract_audio_from_video"]

src/streamsync/example_script.py

@@ -1,33 +1,36 @@
-import mne  # noqa F401
+from __future__ import annotations
+
+import mne
 
 from streamsync import StreamSync, extract_audio_from_video
 
 # load an MNE raw file
-raw = None
-cam1 = None
-flux1 = None
+raw = mne.io.read_raw("")
+cam1 = "path/to/video"
+flux1 = mne.io.read_raw("")
 my_events = []
 
 
 subjects = ["146a", "222b"]
 
 for subj in subjects:
-    # construct the filename/path
-    # load the Raw
-    # figure out where video files are & load them
-    audio1 = extract_audio_from_video(cam1)
-
+    # MEG data from SQUID
     ss = StreamSync(raw, "STIM001")
-    ss.add_stream(audio1)
-    ss.add_camera_events(my_events)
-    ss.add_stream(flux1)
+    # audio from camera
+    audio1 = extract_audio_from_video(cam1, channel=2)
+    ss.add_stream(audio1, channel=1)
+    # MEG data from FLUX
+    ss.add_stream(flux1, channel=127)
+    # synchronize
     result = ss.do_syncing()
+    # plot (to verify good sync)
     fig = ss.plot_sync()
+    fig.savefig(...)  # optional
+    # add camera events to SQUID raw as Annotations
     annot = ss.add_camera_events(my_events)
     raw.set_annotations(annot)
-    fig.savefig(...)
     if result < 0.7:
-        write_log_msg(f"subj {subj} had bad pulse syncing, aborting")  # noqa F821
+        write_log_msg(f"subj {subj} had bad pulse syncing")  # noqa: F821
         continue
 
     # apply maxfilter

src/streamsync/streamsync.py

@@ -1,3 +1,17 @@
+from __future__ import annotations
+
+import os
+
+import matplotlib.pyplot as plt
+import mne
+import numpy as np
+from numpy.typing import ArrayLike
+
+type Stream = mne.io.BaseRaw | ArrayLike
+type Channel = str | int
+type PathLike = str | bytes | os.PathLike
+
+
 class StreamSync:
     """Synchronize two data streams.
 
@@ -8,12 +22,20 @@ class StreamSync:
     time-warped to the timescale of the `Raw`.
     """
 
-    def __init__(self, reference_object, pulse_channel):
-        self.ref_stream = reference_object.get_chan(pulse_channel)
-        self.sfreq = reference_object.info["sfreq"]  # Hz
-        self.streams = []
+    def __init__(self, reference_object: Stream, pulse_channel: Channel) -> None:
+        self.ref_stream: ArrayLike = reference_object.get_chan(pulse_channel)
+        self.sfreq: float = reference_object.info["sfreq"]  # Hz
+        self.streams: list[Stream] = []
+
+    def add_camera_events(self, events: ArrayLike) -> mne.Annotations:  # noqa: ARG002
+        return mne.Annotations([], [], [])
 
-    def add_stream(self, stream, channel=None, events=None):  # noqa ARG002
+    def add_stream(
+        self,
+        stream: Stream,
+        channel: Channel,
+        events: ArrayLike | None = None,  # noqa: ARG002
+    ) -> StreamSync:
         """Add a new ``Raw`` or video stream, optionally with events.
 
         stream : Raw | wav
@@ -26,22 +48,27 @@ def add_stream(self, stream, channel=None, events=None):  # noqa ARG002
         """
         pulses = self._extract_pulse_sequence_from_stream(stream, channel=channel)
         self.streams.append(pulses)
+        return self
 
-    def _extract_pulse_sequence_from_stream(self, stream, channel):
+    def _extract_pulse_sequence_from_stream(
+        self,
+        stream: Stream,  # noqa: ARG002
+        channel: Channel,  # noqa: ARG002
+    ) -> ArrayLike:
         # TODO triage based on input type (e.g., if it's a Raw, pull out a stim chan,
         # if it's audio, just add it as-is)
-        pass
+        return np.array([])  # fake return
 
-    def do_syncing(self):
+    def do_syncing(self) -> float:
         """Synchronize all streams with the reference stream."""
         # TODO (waves hands) do the hard part.
         # TODO spit out a report of correlation/association between all pairs of streams
-        pass
+        return 0.0
 
-    def plot_sync(self):
-        pass
+    def plot_sync(self) -> plt.figure.Figure:
+        return plt.figure.Figure()
 
 
-def extract_audio_from_video(path_to_video, channel):  # noqa ARG001
+def extract_audio_from_video(path_to_video: PathLike, channel: Channel) -> ArrayLike:  # noqa: ARG001
     """Path can be a regex or glob to allow batch processing."""
-    pass
+    return np.array([])  # fake return

tests/test_package.py

@@ -2,7 +2,7 @@
 
 import importlib.metadata
 
-import ilabs_streamsync as m
+import streamsync as m
 
 
 def test_version():