Convert core streamdaq routines to noob

.github/workflows/tests.yml

@@ -20,7 +20,7 @@ jobs:
       fail-fast: false
       matrix:
         os: ['ubuntu-latest', 'windows-latest', 'macos-14']
-        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.11", "3.12", "3.13", "3.14"]
         exclude:
           - os: 'windows-latest'
             python-version: "3.13"

mio/data/tubes/stream/stream-fpga.yaml

@@ -0,0 +1,35 @@
+noob_id: stream-fpga
+description: |
+  Source of fpga data for the stream device
+
+input:
+  config:
+    type: mio.devices.stream.StreamDevConfig
+    scope: tube
+  capture_binary:
+    type: pathlib.Path | None
+    scope: tube
+
+nodes:
+  fpga:
+    type: mio.devices.stream.nodes.iter_fpga
+    params:
+      config: input.config
+  split_buffers:
+    type: mio.devices.stream.nodes.SplitBuffers
+    params:
+      config: input.config
+    depends:
+      - chunk: fpga.chunk
+
+  # binary i/o
+  write_binary:
+    type: mio.io.append_binary
+    depends:
+      - path: input.capture_binary
+      - data: fpga.chunk
+
+  return:
+    type: return
+    depends:
+      - buffers: split_buffers.buffers

mio/data/tubes/stream/stream-frame.yaml

@@ -0,0 +1,11 @@
+noob_id: 1
+description: |
+  Frame assembly from buffers
+
+input:
+  config:
+    type: mio.devices.stream.StreamDevConfig
+    scope: tube
+  buffers:
+    type: list[bytes]
+    scope: process

mio/data/tubes/stream/stream.yaml

@@ -0,0 +1,23 @@
+noob_id: stream
+description: |
+  Combined core streamdev pipeline (WIP)
+
+input:
+  config:
+    type: mio.devices.stream.StreamDevConfig
+    scope: tube
+  capture_binary:
+    type: pathlib.Path | None
+    scope: tube
+
+nodes:
+  fpga:
+    type: tube
+    params:
+      tube: stream-fpga
+  frame-assembly:
+    type: tube
+    params:
+      tube: stream-frame
+    depends:
+      - buffers: fpga.buffers

mio/devices/base/headers.py

@@ -1,18 +1,12 @@
 """Base device headers"""
 
-import sys
 from collections.abc import Sequence
-from typing import Any, ClassVar
+from typing import Any, ClassVar, Self
 
 import pandera.pandas as pa
 
 from mio.models import Container, Table
 
-if sys.version_info <= (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
-
 
 class BufferHeader(Container):
     """

mio/devices/stream/config.py

@@ -231,3 +231,10 @@ def buffer_npix(self) -> list[int]:
         )
         quotient, remainder = divmod(px_per_frame, payload_bytes)
         return [payload_bytes] * int(quotient) + ([int(remainder)] if remainder else [])
+
+    @property
+    def read_length(self) -> int:
+        """
+        How many bytes to read from the FPGA per chunk, roughly the expected size of a buffer
+        """
+        return int(max(self.buffer_npix) * self.pix_depth / 8 / 16) * 16

mio/devices/stream/headers.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import sys
 import time
 from typing import TYPE_CHECKING, ClassVar
 
@@ -19,10 +18,7 @@
 if TYPE_CHECKING:
     from mio.devices.stream.config import StreamDevConfig
 
-if sys.version_info < (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
+from typing import Self
 
 
 class ADCScaling(MiniscopeConfig):

mio/devices/stream/nodes.py

@@ -9,11 +9,17 @@
 import queue
 import time
 from collections.abc import Callable, Generator, Iterator
+from functools import cached_property
 from pathlib import Path
+from typing import Annotated as A
 from typing import Any, Union
+from typing import Literal as L
 
 import numpy as np
 from bitstring import BitArray, Bits
+from noob import Name, Node
+from noob.event import MetaSignal
+from pydantic import PrivateAttr
 
 from mio import init_logger
 from mio.devices.stream import StreamBufferHeader, StreamDevConfig
@@ -33,6 +39,61 @@
     pass  # okDev stays None; error raised when actually trying to use FPGA
 
 
+def iter_fpga(config: StreamDevConfig) -> Generator[A[bytes, Name("chunk")], None, None]:
+    """
+    Iterate a raw binary stream from the FPGA in chunks
+    (not necessarily split into buffers)
+    """
+    # set up fpga interfaces
+    dev = init_okdev(config.bitstream, config.read_length)
+
+    while True:
+        yield next(dev)
+
+
+class SplitBuffers(Node):
+    """
+    Collect raw chunks from the FPGA, yield buffers
+
+    The data are passed in fixed chunks.
+    Then we concatenate the chunks and try to look for {attr}`.SplitBuffers.preamble` in the data.
+    The data between every pair of {attr}`.SplitBuffers.preamble`
+    is considered to be a single buffer and yielded.
+    """
+
+    config: StreamDevConfig
+    _buffer: BitArray = PrivateAttr(default_factory=BitArray)
+
+    @cached_property
+    def preamble(self) -> Bits:
+        """
+        The preamble that indicates the start of a buffer.
+
+        If ``config.reverse_header_bits`` is true, the preamble is bit-flipped
+        """
+        pre = Bits(self.config.preamble)
+        if self.config.reverse_header_bits:
+            pre = pre[::-1]
+        return pre
+
+    def process(self, chunk: bytes) -> A[list[bytes] | L[MetaSignal.NoEvent], Name("chunks")]:
+        """
+        Append the chunk to the buffer, return any split buffers, if found.
+        """
+        self._buffer += BitArray(chunk)
+        pos = list(self._buffer.findall(self.preamble))
+        buffers = [self._buffer[start:stop].tobytes() for start, stop in zip(pos[:-1], pos[1:])]
+        if buffers:
+            self._buffer = self._buffer[pos[-1] :]
+            return buffers
+        else:
+            return MetaSignal.NoEvent
+
+    def deinit(self) -> None:
+        """Clear the internal buffer"""
+        self._buffer = BitArray()
+
+
 def exact_iter(f: Callable, sentinel: Any) -> Generator[Any, None, None]:
     """
     A version of :func:`iter` that compares with `is` rather than `==`

mio/interfaces/mocks.py

@@ -10,13 +10,8 @@
 # ruff: noqa: D102
 
 import os
-import sys
 from pathlib import Path
-
-if sys.version_info < (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
+from typing import Self
 
 from mio.exceptions import EndOfRecordingException
 

mio/interfaces/opalkelly.py

@@ -2,12 +2,7 @@
 Interfaces for OpalKelly (model number?) FPGAs
 """
 
-import sys
-
-if sys.version_info < (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
+from typing import Self
 
 from mio.exceptions import (
     DeviceConfigurationError,

mio/io/binary.py

@@ -0,0 +1,12 @@
+"""Raw bytes i/o"""
+
+from pathlib import Path
+
+
+def append_binary(path: Path | None, data: bytes) -> None:
+    """Just append some binary to a path!"""
+    if path is None:
+        # FIXME: Still working how we want enabling/disabling binary i/o to look like in streamdev
+        return
+    with open(path, "ab") as f:
+        f.write(data)

mio/models/dataset.py

@@ -56,11 +56,10 @@
 """
 
 import re
-import sys
 import warnings
 from pathlib import Path
 from typing import Annotated as A
-from typing import Any, Literal, TypeAlias
+from typing import Any, Literal, Self, TypeAlias, TypedDict
 
 import pandas as pd
 from numpydantic import NDArraySchema
@@ -80,11 +79,6 @@
 from mio.models.process import NoisePatchConfig
 from mio.utils import _format_ranges
 
-if sys.version_info < (3, 11):
-    from typing_extensions import Self, TypedDict
-else:
-    from typing import Self, TypedDict
-
 VIDEO_EXTENSIONS = (".avi", ".mp4")
 RECORDING_TYPES = Literal["raw", "stitched"]
 DERIVATION_TYPES = Literal["stitched"]

mio/models/devupdate.py

@@ -2,17 +2,12 @@
 Models for device update commands.
 """
 
-import sys
 from enum import Enum
+from typing import Self
 
 import serial.tools.list_ports
 from pydantic import BaseModel, field_validator, model_validator
 
-if sys.version_info < (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
-
 
 class DeviceCommand(Enum):
     """Commands for device."""

mio/models/mixins.py

@@ -5,7 +5,6 @@
 
 import re
 import shutil
-import sys
 from collections.abc import Iterator
 from importlib.metadata import version
 from itertools import chain
@@ -14,6 +13,8 @@
     Any,
     ClassVar,
     Literal,
+    NotRequired,
+    Self,
     TypedDict,
     TypeVar,
     overload,
@@ -22,11 +23,6 @@
 import yaml
 from pydantic import BaseModel, Field, ValidationError, field_validator
 
-if sys.version_info < (3, 11):
-    from typing_extensions import NotRequired, Self
-else:
-    from typing import NotRequired, Self
-
 from mio.types import ConfigID, ConfigSource, PythonIdentifier, valid_config_id
 
 T = TypeVar("T")

mio/models/models.py

@@ -2,20 +2,14 @@
 Base and meta model classes.
 """
 
-import sys
 from pathlib import Path
-from typing import Any, ClassVar
+from typing import Any, ClassVar, Self
 
 import pandas as pd
 import pandera.pandas as pa
 from pandera.typing import DataFrame
 from pydantic import BaseModel
 
-if sys.version_info < (3, 11):
-    from typing_extensions import Self
-else:
-    from typing import Self
-
 
 class MiniscopeIOModel(BaseModel):
     """

mio/process/frame_helper.py

@@ -4,8 +4,8 @@
 
 from __future__ import annotations
 
-import sys
 from abc import abstractmethod
+from typing import TypedDict
 
 import cv2
 import numpy as np
@@ -18,11 +18,6 @@
     NoisePatchConfig,
 )
 
-if sys.version_info < (3, 11):
-    from typing_extensions import TypedDict
-else:
-    from typing import TypedDict
-
 logger = init_logger("frame_helper")
 
 

mio/types.py

@@ -6,16 +6,14 @@
 import sys
 from os import PathLike
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated, Any, TypeAlias
 
 from pydantic import AfterValidator, Field
 
-if sys.version_info < (3, 10) or sys.version_info < (3, 13):
-    from typing import TypeAlias
-
+if sys.version_info < (3, 13):
     from typing_extensions import TypeIs
 else:
-    from typing import TypeAlias, TypeIs
+    from typing import TypeIs
 
 CONFIG_ID_PATTERN = r"[\w\-\/#]+"
 """

mio/utils.py

@@ -83,3 +83,8 @@ def _format_ranges(indices: list[int] | set[int]) -> list[str]:
 
     ranges.append(f"{start}-{end}" if start != end else str(start))
     return ranges
+
+
+def add_noob_sources() -> list[Path]:
+    """Provide the tubes directory so that noob can find it!"""
+    return [Path(__file__).parent / "data" / "tubes"]