Add LI-COR Odyssey Classic + DeviceCard for instrument provenance

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Unreleased
 
+### Added
+
+- LI-COR Odyssey Classic (model 9120) infrared imaging system at `pylabrobot.li_cor.odyssey`
+- `Scanning`, `ImageRetrieval`, and `InstrumentStatus` capabilities at `pylabrobot.capabilities.scanning.*`
+- `DeviceCard` class and `HasDeviceCard` mixin at `pylabrobot.device_card` for instrument identity / provenance metadata (model-base + per-instance two-tier cards)
+
 ## 0.2.1
 
 ### Added

mypy.ini

@@ -21,3 +21,6 @@ ignore_missing_imports = True
 
 [mypy-opentrons_shared_data.*]
 ignore_missing_imports = True
+
+[mypy-aiohttp.*]
+ignore_missing_imports = True

pylabrobot/capabilities/scanning/image_retrieval/__init__.py

@@ -0,0 +1,2 @@
+from .backend import ImageRetrievalBackend, ImageRetrievalError
+from .image_retrieval import ImageRetrieval

pylabrobot/capabilities/scanning/image_retrieval/backend.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+from typing import List
+
+from pylabrobot.capabilities.capability import CapabilityBackend
+
+
+class ImageRetrievalError(Exception):
+  """Capability-generic exception for image retrieval failures."""
+
+
+class ImageRetrievalBackend(CapabilityBackend, metaclass=ABCMeta):
+  """Abstract backend for the image retrieval capability.
+
+  Lists and downloads previously-acquired scans from instrument
+  storage. Independent of the scanning capability — scans persist
+  after the session that produced them, and lab users often retrieve
+  images they did not acquire themselves.
+  """
+
+  @abstractmethod
+  async def list_groups(self) -> List[str]:
+    """Return the names of scan groups available on the instrument."""
+
+  @abstractmethod
+  async def list_scans(self, group: str) -> List[str]:
+    """Return scan names within ``group``."""
+
+  @abstractmethod
+  async def download(self, group: str, scan_name: str) -> bytes:
+    """Download all channels for a scan, concatenated.
+
+    Vendors with multi-channel scans (e.g. Odyssey 700 / 800 nm) may
+    expose a ``download_channel`` extension on the concrete backend.
+    """

pylabrobot/capabilities/scanning/image_retrieval/image_retrieval.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import logging
+from typing import List
+
+from pylabrobot.capabilities.capability import Capability, need_capability_ready
+
+from .backend import ImageRetrievalBackend
+
+logger = logging.getLogger(__name__)
+
+
+class ImageRetrieval(Capability):
+  """Image retrieval capability — list and download saved scans."""
+
+  def __init__(self, backend: ImageRetrievalBackend):
+    super().__init__(backend=backend)
+    self.backend: ImageRetrievalBackend = backend
+
+  @need_capability_ready
+  async def list_groups(self) -> List[str]:
+    """Return the names of scan groups available on the instrument."""
+    return await self.backend.list_groups()
+
+  @need_capability_ready
+  async def list_scans(self, group: str) -> List[str]:
+    """Return scan names within ``group``."""
+    return await self.backend.list_scans(group)
+
+  @need_capability_ready
+  async def download(self, group: str, scan_name: str) -> bytes:
+    """Download all channels for a scan, concatenated."""
+    return await self.backend.download(group, scan_name)

pylabrobot/capabilities/scanning/image_retrieval/image_retrieval_tests.py

@@ -0,0 +1,64 @@
+"""Tests for ImageRetrieval."""
+
+import unittest
+from typing import Dict, List
+
+from pylabrobot.capabilities.scanning.image_retrieval.backend import (
+  ImageRetrievalBackend,
+)
+from pylabrobot.capabilities.scanning.image_retrieval.image_retrieval import (
+  ImageRetrieval,
+)
+
+
+class InMemoryImageRetrievalBackend(ImageRetrievalBackend):
+  """Backend that serves a pre-loaded in-memory store."""
+
+  def __init__(self, store: Dict[str, Dict[str, bytes]]):
+    self._store = store
+
+  async def list_groups(self) -> List[str]:
+    return list(self._store.keys())
+
+  async def list_scans(self, group: str) -> List[str]:
+    return list(self._store.get(group, {}).keys())
+
+  async def download(self, group: str, scan_name: str) -> bytes:
+    return self._store[group][scan_name]
+
+
+class TestImageRetrieval(unittest.IsolatedAsyncioTestCase):
+  async def asyncSetUp(self):
+    self.store = {
+      "odyssey": {"scan_a": b"AAAA", "scan_b": b"BBBB"},
+      "public": {"shared": b"CCCC"},
+    }
+    self.backend = InMemoryImageRetrievalBackend(self.store)
+    self.cap = ImageRetrieval(backend=self.backend)
+    await self.cap._on_setup()
+
+  async def test_list_groups(self):
+    self.assertEqual(sorted(await self.cap.list_groups()), ["odyssey", "public"])
+
+  async def test_list_scans(self):
+    self.assertEqual(
+      sorted(await self.cap.list_scans("odyssey")),
+      ["scan_a", "scan_b"],
+    )
+    self.assertEqual(await self.cap.list_scans("missing"), [])
+
+  async def test_download(self):
+    self.assertEqual(await self.cap.download("odyssey", "scan_a"), b"AAAA")
+    self.assertEqual(await self.cap.download("public", "shared"), b"CCCC")
+
+  async def test_methods_require_setup(self):
+    backend = InMemoryImageRetrievalBackend({})
+    cap = ImageRetrieval(backend=backend)
+    with self.assertRaises(RuntimeError):
+      await cap.list_groups()
+    with self.assertRaises(RuntimeError):
+      await cap.download("g", "s")
+
+
+if __name__ == "__main__":
+  unittest.main()

pylabrobot/capabilities/scanning/instrument_status/__init__.py

@@ -0,0 +1,3 @@
+from .backend import InstrumentStatusBackend, InstrumentStatusError
+from .instrument_status import InstrumentStatus
+from .standard import InstrumentStatusReading

pylabrobot/capabilities/scanning/instrument_status/backend.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+
+from pylabrobot.capabilities.capability import CapabilityBackend
+
+from .standard import InstrumentStatusReading
+
+
+class InstrumentStatusError(Exception):
+  """Capability-generic exception for instrument status read failures."""
+
+
+class InstrumentStatusBackend(CapabilityBackend, metaclass=ABCMeta):
+  """Abstract backend for instrument status polling."""
+
+  @abstractmethod
+  async def read_status(self) -> InstrumentStatusReading:
+    """Return the current instrument status snapshot."""

pylabrobot/capabilities/scanning/instrument_status/instrument_status.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import logging
+
+from pylabrobot.capabilities.capability import Capability, need_capability_ready
+
+from .backend import InstrumentStatusBackend
+from .standard import InstrumentStatusReading
+
+logger = logging.getLogger(__name__)
+
+
+class InstrumentStatus(Capability):
+  """Instrument status capability — poll the device's state machine."""
+
+  def __init__(self, backend: InstrumentStatusBackend):
+    super().__init__(backend=backend)
+    self.backend: InstrumentStatusBackend = backend
+
+  @need_capability_ready
+  async def read_status(self) -> InstrumentStatusReading:
+    """Return the current instrument status snapshot."""
+    return await self.backend.read_status()

pylabrobot/capabilities/scanning/instrument_status/instrument_status_tests.py

@@ -0,0 +1,63 @@
+"""Tests for InstrumentStatus."""
+
+import unittest
+
+from pylabrobot.capabilities.scanning.instrument_status.backend import (
+  InstrumentStatusBackend,
+)
+from pylabrobot.capabilities.scanning.instrument_status.instrument_status import (
+  InstrumentStatus,
+)
+from pylabrobot.capabilities.scanning.instrument_status.standard import (
+  InstrumentStatusReading,
+)
+
+
+class StubInstrumentStatusBackend(InstrumentStatusBackend):
+  """Backend that returns a fixed reading and counts calls."""
+
+  def __init__(self, reading: InstrumentStatusReading):
+    self.reading = reading
+    self.call_count = 0
+
+  async def read_status(self) -> InstrumentStatusReading:
+    self.call_count += 1
+    return self.reading
+
+
+class TestInstrumentStatus(unittest.IsolatedAsyncioTestCase):
+  async def asyncSetUp(self):
+    self.reading = InstrumentStatusReading(
+      state="Scanning",
+      current_user="alice",
+      progress=37.5,
+      time_remaining="2 minutes",
+      lid_open=False,
+    )
+    self.backend = StubInstrumentStatusBackend(self.reading)
+    self.cap = InstrumentStatus(backend=self.backend)
+    await self.cap._on_setup()
+
+  async def test_read_status_returns_backend_reading(self):
+    result = await self.cap.read_status()
+    self.assertIs(result, self.reading)
+    self.assertEqual(self.backend.call_count, 1)
+
+  async def test_read_status_passes_through_fields(self):
+    result = await self.cap.read_status()
+    self.assertEqual(result.state, "Scanning")
+    self.assertEqual(result.current_user, "alice")
+    self.assertEqual(result.progress, 37.5)
+    self.assertEqual(result.time_remaining, "2 minutes")
+    self.assertFalse(result.lid_open)
+
+  async def test_read_status_requires_setup(self):
+    backend = StubInstrumentStatusBackend(self.reading)
+    cap = InstrumentStatus(backend=backend)
+    with self.assertRaises(RuntimeError):
+      await cap.read_status()
+    self.assertEqual(backend.call_count, 0)
+
+
+if __name__ == "__main__":
+  unittest.main()

pylabrobot/capabilities/scanning/instrument_status/standard.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class InstrumentStatusReading:
+  """Generic instrument status snapshot.
+
+  Vendor backends populate the fields they have a value for; missing
+  fields keep their defaults.
+  """
+
+  state: str
+  current_user: str = ""
+  progress: float = 0.0
+  time_remaining: str = ""
+  lid_open: bool = False

pylabrobot/capabilities/scanning/scanning/__init__.py

@@ -0,0 +1,2 @@
+from .backend import ScanningBackend, ScanningError
+from .scanning import Scanning

pylabrobot/capabilities/scanning/scanning/backend.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+from typing import Optional
+
+from pylabrobot.capabilities.capability import CapabilityBackend
+from pylabrobot.serializer import SerializableMixin
+
+
+class ScanningError(Exception):
+  """Capability-generic exception for scanning failures.
+
+  Vendor backends should raise a subclass that ALSO inherits from the
+  vendor's driver-level exception, so callers can catch on either axis
+  (capability-generic or vendor-specific).
+  """
+
+
+class ScanningBackend(CapabilityBackend, metaclass=ABCMeta):
+  """Abstract backend for the scanning capability.
+
+  Concrete backends configure and control flatbed-style fluorescence
+  / luminescence scans. The capability does not assume a plate or
+  well grid — Odyssey-style flatbed imagers and gel docs fit;
+  plate-aware microscopy belongs under :class:`Microscopy`.
+  """
+
+  @abstractmethod
+  async def configure(self, backend_params: Optional[SerializableMixin] = None) -> None:
+    """Set up the next scan with vendor-specific parameters."""
+
+  @abstractmethod
+  async def start(self) -> None:
+    """Begin acquisition. Backend must be configured first."""
+
+  @abstractmethod
+  async def stop(self) -> None:
+    """Graceful stop — finish current line, save partial output."""
+
+  @abstractmethod
+  async def pause(self) -> None:
+    """Pause acquisition. Resume by calling :meth:`start` again."""
+
+  @abstractmethod
+  async def cancel(self) -> None:
+    """Abort acquisition and discard any partial output."""

pylabrobot/capabilities/scanning/scanning/scanning.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from pylabrobot.capabilities.capability import Capability, need_capability_ready
+from pylabrobot.serializer import SerializableMixin
+
+from .backend import ScanningBackend
+
+logger = logging.getLogger(__name__)
+
+
+class Scanning(Capability):
+  """Flatbed scanning capability — fluorescence / luminescence imager control."""
+
+  def __init__(self, backend: ScanningBackend):
+    super().__init__(backend=backend)
+    self.backend: ScanningBackend = backend
+
+  @need_capability_ready
+  async def configure(self, backend_params: Optional[SerializableMixin] = None) -> None:
+    """Set up the next scan with vendor-specific parameters."""
+    await self.backend.configure(backend_params=backend_params)
+
+  @need_capability_ready
+  async def start(self) -> None:
+    """Begin acquisition."""
+    await self.backend.start()
+
+  @need_capability_ready
+  async def stop(self) -> None:
+    """Graceful stop — finish current line, save partial output."""
+    await self.backend.stop()
+
+  @need_capability_ready
+  async def pause(self) -> None:
+    """Pause acquisition. Resume by calling :meth:`start` again."""
+    await self.backend.pause()
+
+  @need_capability_ready
+  async def cancel(self) -> None:
+    """Abort acquisition and discard any partial output."""
+    await self.backend.cancel()