diff --git a/CHANGELOG.md b/CHANGELOG.md
index 220209251b6..99238cb93a9 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/mypy.ini b/mypy.ini
index 614e3886553..a6ac770eddb 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -21,3 +21,6 @@ ignore_missing_imports = True
[mypy-opentrons_shared_data.*]
ignore_missing_imports = True
+
+[mypy-aiohttp.*]
+ignore_missing_imports = True
diff --git a/pylabrobot/capabilities/scanning/__init__.py b/pylabrobot/capabilities/scanning/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/pylabrobot/capabilities/scanning/image_retrieval/__init__.py b/pylabrobot/capabilities/scanning/image_retrieval/__init__.py
new file mode 100644
index 00000000000..3a7d56f2a7e
--- /dev/null
+++ b/pylabrobot/capabilities/scanning/image_retrieval/__init__.py
@@ -0,0 +1,2 @@
+from .backend import ImageRetrievalBackend, ImageRetrievalError
+from .image_retrieval import ImageRetrieval
diff --git a/pylabrobot/capabilities/scanning/image_retrieval/backend.py b/pylabrobot/capabilities/scanning/image_retrieval/backend.py
new file mode 100644
index 00000000000..f6f82ae01ac
--- /dev/null
+++ b/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.
+ """
diff --git a/pylabrobot/capabilities/scanning/image_retrieval/image_retrieval.py b/pylabrobot/capabilities/scanning/image_retrieval/image_retrieval.py
new file mode 100644
index 00000000000..848c7a9935c
--- /dev/null
+++ b/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)
diff --git a/pylabrobot/capabilities/scanning/image_retrieval/image_retrieval_tests.py b/pylabrobot/capabilities/scanning/image_retrieval/image_retrieval_tests.py
new file mode 100644
index 00000000000..daa46153f69
--- /dev/null
+++ b/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()
diff --git a/pylabrobot/capabilities/scanning/instrument_status/__init__.py b/pylabrobot/capabilities/scanning/instrument_status/__init__.py
new file mode 100644
index 00000000000..f17ef263cbc
--- /dev/null
+++ b/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
diff --git a/pylabrobot/capabilities/scanning/instrument_status/backend.py b/pylabrobot/capabilities/scanning/instrument_status/backend.py
new file mode 100644
index 00000000000..254ed6ca970
--- /dev/null
+++ b/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."""
diff --git a/pylabrobot/capabilities/scanning/instrument_status/instrument_status.py b/pylabrobot/capabilities/scanning/instrument_status/instrument_status.py
new file mode 100644
index 00000000000..f3535874272
--- /dev/null
+++ b/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()
diff --git a/pylabrobot/capabilities/scanning/instrument_status/instrument_status_tests.py b/pylabrobot/capabilities/scanning/instrument_status/instrument_status_tests.py
new file mode 100644
index 00000000000..cb9403c4ff4
--- /dev/null
+++ b/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()
diff --git a/pylabrobot/capabilities/scanning/instrument_status/standard.py b/pylabrobot/capabilities/scanning/instrument_status/standard.py
new file mode 100644
index 00000000000..c99a95aa6ad
--- /dev/null
+++ b/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
diff --git a/pylabrobot/capabilities/scanning/scanning/__init__.py b/pylabrobot/capabilities/scanning/scanning/__init__.py
new file mode 100644
index 00000000000..74cea53ccfa
--- /dev/null
+++ b/pylabrobot/capabilities/scanning/scanning/__init__.py
@@ -0,0 +1,2 @@
+from .backend import ScanningBackend, ScanningError
+from .scanning import Scanning
diff --git a/pylabrobot/capabilities/scanning/scanning/backend.py b/pylabrobot/capabilities/scanning/scanning/backend.py
new file mode 100644
index 00000000000..b55d7ebd021
--- /dev/null
+++ b/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."""
diff --git a/pylabrobot/capabilities/scanning/scanning/scanning.py b/pylabrobot/capabilities/scanning/scanning/scanning.py
new file mode 100644
index 00000000000..dde012bbeee
--- /dev/null
+++ b/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()
diff --git a/pylabrobot/capabilities/scanning/scanning/scanning_tests.py b/pylabrobot/capabilities/scanning/scanning/scanning_tests.py
new file mode 100644
index 00000000000..1951af30cb1
--- /dev/null
+++ b/pylabrobot/capabilities/scanning/scanning/scanning_tests.py
@@ -0,0 +1,71 @@
+"""Tests for Scanning."""
+
+import unittest
+from typing import List, Optional, Tuple
+
+from pylabrobot.capabilities.scanning.scanning.backend import ScanningBackend
+from pylabrobot.capabilities.scanning.scanning.scanning import Scanning
+from pylabrobot.serializer import SerializableMixin
+
+
+class RecordingScanningBackend(ScanningBackend):
+ """Backend that records every call so tests can assert on the sequence."""
+
+ def __init__(self):
+ self.calls: List[Tuple[str, Optional[SerializableMixin]]] = []
+
+ async def configure(self, backend_params: Optional[SerializableMixin] = None) -> None:
+ self.calls.append(("configure", backend_params))
+
+ async def start(self) -> None:
+ self.calls.append(("start", None))
+
+ async def stop(self) -> None:
+ self.calls.append(("stop", None))
+
+ async def pause(self) -> None:
+ self.calls.append(("pause", None))
+
+ async def cancel(self) -> None:
+ self.calls.append(("cancel", None))
+
+
+class TestScanning(unittest.IsolatedAsyncioTestCase):
+ async def asyncSetUp(self):
+ self.backend = RecordingScanningBackend()
+ self.cap = Scanning(backend=self.backend)
+ await self.cap._on_setup()
+
+ async def test_configure_forwards_params(self):
+ sentinel = object()
+ await self.cap.configure(backend_params=sentinel) # type: ignore[arg-type]
+ self.assertEqual(self.backend.calls, [("configure", sentinel)])
+
+ async def test_full_verb_sequence(self):
+ await self.cap.configure()
+ await self.cap.start()
+ await self.cap.pause()
+ await self.cap.cancel()
+ await self.cap.stop()
+ self.assertEqual(
+ [name for name, _ in self.backend.calls],
+ ["configure", "start", "pause", "cancel", "stop"],
+ )
+
+ async def test_setup_finished_flag(self):
+ self.assertTrue(self.cap.setup_finished)
+ await self.cap._on_stop()
+ self.assertFalse(self.cap.setup_finished)
+
+ async def test_methods_require_setup(self):
+ backend = RecordingScanningBackend()
+ cap = Scanning(backend=backend)
+ with self.assertRaises(RuntimeError):
+ await cap.start()
+ with self.assertRaises(RuntimeError):
+ await cap.configure()
+ self.assertEqual(backend.calls, [])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/pylabrobot/device_card.py b/pylabrobot/device_card.py
new file mode 100644
index 00000000000..0846de924e1
--- /dev/null
+++ b/pylabrobot/device_card.py
@@ -0,0 +1,165 @@
+"""DeviceCard — machine-readable device description.
+
+A :class:`DeviceCard` is a metadata bag attached to a :class:`Device`.
+It carries:
+
+- **Identity** — persistent identifier (e.g. PIDInst Handle URI),
+ landing page, friendly name. Empty in the model-base card; each
+ deployment populates it with its own unit's identity.
+- **Capability specs** — operating ranges, supported settings, and
+ feature flags pulled from the operator's manual. Lets UIs and
+ validators reason about what a unit can actually do.
+- **Connection metadata** — protocol, port, auth method, discovery
+ mechanism. Configuration the model defines once, used by every
+ deployment.
+
+Two-tier design::
+
+ base = ODYSSEY_CLASSIC_BASE # ships with the device package
+ instance = DeviceCard.instance(identity={ # per-deployment
+ "pid": "http://hdl.handle.net/21.11157/psf97-zv353",
+ "landing_page": "https://b2inst.gwdg.de/records/psf97-zv353",
+ "name": "Odyssey, Lab 3 (WUR HAP)",
+ })
+ card = base.merge(instance) # effective deployed card
+
+Devices that carry a card declare the :class:`HasDeviceCard` mixin so
+the attribute is discoverable via type checks. Tooling that consumes
+identity (TIFF tagging, provenance writers, dataset registration) can
+duck-type ``hasattr(device, "card")`` or rely on the mixin.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class DeviceCard:
+ """Machine-readable device description with merge and introspection.
+
+ Fields:
+ name: Friendly model name (e.g. "Odyssey Classic").
+ vendor: Vendor name (e.g. "LI-COR Biosciences").
+ model: Model number / SKU (e.g. "9120").
+ capabilities: Per-capability spec sheet — keys are capability
+ names (``"scanning"``, ``"image_retrieval"``), values are dicts
+ of features / specs / ranges.
+ connection: Connection metadata (protocol, port, auth, network).
+ identity: Per-unit identity — PID, landing page, friendly name.
+ Empty in the model-base card; populated at the instance layer.
+ """
+
+ name: str = ""
+ vendor: str = ""
+ model: str = ""
+ capabilities: dict[str, dict[str, Any]] = field(default_factory=dict)
+ connection: dict[str, Any] = field(default_factory=dict)
+ identity: dict[str, Any] = field(default_factory=dict)
+
+ @classmethod
+ def instance(cls, **kwargs: Any) -> "DeviceCard":
+ """Build a partial card for instance-level overrides.
+
+ Use when populating per-deployment identity (PID, landing page) or
+ overriding a model-base spec for a non-standard unit.
+ """
+ return cls(**kwargs)
+
+ def merge(self, other: "DeviceCard") -> "DeviceCard":
+ """Deep-merge ``other`` on top of ``self``. Returns a new card.
+
+ Merge rules:
+ - Scalar fields (name, vendor, model): ``other`` wins if non-empty.
+ - capabilities: per-capability shallow merge; ``other`` keys
+ override; new capabilities from ``other`` are added.
+ - connection, identity: shallow dict merge, ``other`` wins on key
+ collision.
+
+ Neither input is mutated.
+ """
+ merged_caps = copy.deepcopy(self.capabilities)
+ for cap_name, cap_data in other.capabilities.items():
+ if cap_name in merged_caps:
+ merged_caps[cap_name].update(cap_data)
+ else:
+ merged_caps[cap_name] = copy.deepcopy(cap_data)
+
+ return DeviceCard(
+ name=other.name or self.name,
+ vendor=other.vendor or self.vendor,
+ model=other.model or self.model,
+ capabilities=merged_caps,
+ connection={**self.connection, **other.connection},
+ identity={**self.identity, **other.identity},
+ )
+
+ def has(self, capability: str) -> bool:
+ """Return True if the card declares ``capability``."""
+ return capability in self.capabilities
+
+ def get(self, capability: str, key: str, default: Any = None) -> Any:
+ """Return a single feature / spec value for ``capability``."""
+ return self.capabilities.get(capability, {}).get(key, default)
+
+ def features(self, capability: str) -> dict[str, bool]:
+ """Return all boolean feature flags for ``capability``."""
+ cap = self.capabilities.get(capability, {})
+ return {k: v for k, v in cap.items() if isinstance(v, bool)}
+
+ def specs(self, capability: str) -> dict[str, Any]:
+ """Return all non-boolean specs for ``capability``."""
+ cap = self.capabilities.get(capability, {})
+ return {k: v for k, v in cap.items() if not isinstance(v, bool)}
+
+ def to_dict(self) -> dict:
+ """Serialize to a JSON-compatible dictionary."""
+ return {
+ "name": self.name,
+ "vendor": self.vendor,
+ "model": self.model,
+ "capabilities": copy.deepcopy(self.capabilities),
+ "connection": copy.deepcopy(self.connection),
+ "identity": copy.deepcopy(self.identity),
+ }
+
+ @classmethod
+ def from_dict(cls, data: dict) -> "DeviceCard":
+ """Reconstruct from a dictionary (e.g. loaded from JSON)."""
+ return cls(
+ name=data.get("name", ""),
+ vendor=data.get("vendor", ""),
+ model=data.get("model", ""),
+ capabilities=data.get("capabilities", {}),
+ connection=data.get("connection", {}),
+ identity=data.get("identity", {}),
+ )
+
+ def to_json(self, indent: int = 2) -> str:
+ """Serialize to JSON string."""
+ return json.dumps(self.to_dict(), indent=indent)
+
+ @classmethod
+ def from_json(cls, json_str: str) -> "DeviceCard":
+ """Reconstruct from JSON string."""
+ return cls.from_dict(json.loads(json_str))
+
+
+class HasDeviceCard:
+ """Mixin for devices that carry a :class:`DeviceCard`.
+
+ Devices that want a card declare this mixin and assign ``self.card``
+ in their constructor. The mixin makes the attribute discoverable
+ via type checks::
+
+ if isinstance(device, HasDeviceCard):
+ embed_identity(device.card.identity)
+
+ Same shape as :class:`pylabrobot.capabilities.loading_tray.HasLoadingTray`
+ — a Device-attribute marker mixin, not a Backend mixin.
+ """
+
+ card: DeviceCard
diff --git a/pylabrobot/li_cor/__init__.py b/pylabrobot/li_cor/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/pylabrobot/li_cor/odyssey/__init__.py b/pylabrobot/li_cor/odyssey/__init__.py
new file mode 100644
index 00000000000..634362a5d2c
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/__init__.py
@@ -0,0 +1,61 @@
+"""LI-COR Odyssey Classic — public API."""
+
+from pylabrobot.li_cor.odyssey.chatterbox import (
+ OdysseyChatterboxDriver,
+ OdysseyImageRetrievalChatterboxBackend,
+ OdysseyInstrumentStatusChatterboxBackend,
+ OdysseyScanningChatterboxBackend,
+)
+from pylabrobot.li_cor.odyssey.device_card import ODYSSEY_CLASSIC_BASE
+from pylabrobot.li_cor.odyssey.driver import OdysseyDriver
+from pylabrobot.li_cor.odyssey.errors import (
+ OdysseyError,
+ OdysseyImageError,
+ OdysseyScanError,
+ OdysseyStatusError,
+)
+from pylabrobot.li_cor.odyssey.image_retrieval_backend import (
+ OdysseyImageRetrievalBackend,
+)
+from pylabrobot.li_cor.odyssey.instrument_status_backend import (
+ OdysseyInstrumentStatusBackend,
+ OdysseyState,
+ normalize_state,
+)
+from pylabrobot.li_cor.odyssey.odyssey import OdysseyClassic
+from pylabrobot.li_cor.odyssey.scanning_backend import (
+ DEFAULT_GROUP,
+ OdysseyScanningBackend,
+ OdysseyScanningParams,
+ StopResult,
+)
+from pylabrobot.li_cor.odyssey.tagging import (
+ DEFAULT_SOFTWARE_TAG,
+ build_identity_description,
+ tag_tiff_with_identity,
+)
+
+__all__ = [
+ "OdysseyClassic",
+ "OdysseyDriver",
+ "OdysseyScanningParams",
+ "ODYSSEY_CLASSIC_BASE",
+ "OdysseyScanningBackend",
+ "OdysseyImageRetrievalBackend",
+ "OdysseyInstrumentStatusBackend",
+ "OdysseyChatterboxDriver",
+ "OdysseyScanningChatterboxBackend",
+ "OdysseyImageRetrievalChatterboxBackend",
+ "OdysseyInstrumentStatusChatterboxBackend",
+ "OdysseyError",
+ "OdysseyScanError",
+ "OdysseyImageError",
+ "OdysseyStatusError",
+ "OdysseyState",
+ "normalize_state",
+ "StopResult",
+ "DEFAULT_GROUP",
+ "DEFAULT_SOFTWARE_TAG",
+ "build_identity_description",
+ "tag_tiff_with_identity",
+]
diff --git a/pylabrobot/li_cor/odyssey/chatterbox.py b/pylabrobot/li_cor/odyssey/chatterbox.py
new file mode 100644
index 00000000000..9b18ae13169
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/chatterbox.py
@@ -0,0 +1,205 @@
+"""Chatterbox path for the Odyssey Classic — no instrument required.
+
+Three backend-tier chatterbox classes (one per capability) share an
+:class:`_OdysseyChatterboxState` object that simulates the
+instrument's state machine and stored scans. A minimal
+:class:`OdysseyChatterboxDriver` overrides ``setup`` / ``stop`` to
+no-ops; it exists only because :class:`pylabrobot.device.Device`
+requires a :class:`Driver` instance — the chatterbox backends do
+not call it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import List, Optional
+
+from pylabrobot.capabilities.capability import BackendParams
+from pylabrobot.capabilities.scanning.image_retrieval import ImageRetrievalBackend
+from pylabrobot.capabilities.scanning.instrument_status import (
+ InstrumentStatusBackend,
+ InstrumentStatusReading,
+)
+from pylabrobot.capabilities.scanning.scanning import ScanningBackend
+from pylabrobot.device import Driver
+from pylabrobot.serializer import SerializableMixin
+
+from .driver import OdysseyDriver
+from .instrument_status_backend import OdysseyState
+from .scanning_backend import DEFAULT_GROUP, OdysseyScanningParams
+
+logger = logging.getLogger(__name__)
+
+
+class _OdysseyChatterboxState:
+ """Shared mutable state for the three chatterbox backends."""
+
+ def __init__(self) -> None:
+ self.scanner_state: OdysseyState = "Idle"
+ self.progress: float = 0.0
+ self.lid_open: bool = False
+ self.current_user: str = ""
+ self.current_scan_name: str = ""
+ self.current_group: str = ""
+ self.configured: bool = False
+ self.stop_was_partial: bool = False
+ # Pre-seed the default working group so the chatterbox mirrors
+ # the lab instrument's name space.
+ self.scans: dict[str, dict[str, bytes]] = {
+ DEFAULT_GROUP: {
+ "test_scan": b"CHATTERBOX_TIFF_DATA_700nm",
+ },
+ }
+
+
+class OdysseyChatterboxDriver(OdysseyDriver):
+ """No-op driver for chatterbox runs.
+
+ Bypasses the OdysseyDriver constructor's credential check and
+ overrides ``setup`` / ``stop`` so a Device can be wired with this
+ driver + chatterbox backends without contacting any instrument.
+ """
+
+ def __init__(self) -> None:
+ # Skip OdysseyDriver.__init__ — it requires real credentials.
+ # Call Driver.__init__ directly for instance-set registration.
+ Driver.__init__(self)
+ self._host = "chatterbox"
+ self._port = 0
+ self._timeout_seconds = 0.0
+ self._username = ""
+ self._password = ""
+ self._base_url = ""
+ self._auth = None
+ self._timeout = None
+ self._session = None
+
+ async def setup(self, backend_params: Optional[BackendParams] = None) -> None:
+ return None
+
+ async def stop(self) -> None:
+ return None
+
+ def serialize(self) -> dict:
+ return {"type": self.__class__.__name__}
+
+
+class OdysseyScanningChatterboxBackend(ScanningBackend):
+ """Chatterbox scanning backend — drives state with simulated progress."""
+
+ def __init__(self, state: _OdysseyChatterboxState) -> None:
+ super().__init__()
+ self._state = state
+
+ async def configure(self, backend_params: Optional[SerializableMixin] = None) -> None:
+ params = (
+ backend_params
+ if isinstance(backend_params, OdysseyScanningParams)
+ else OdysseyScanningParams()
+ )
+ self._state.configured = True
+ self._state.current_scan_name = params.name
+ self._state.current_group = params.group
+ self._state.scanner_state = "Configured"
+ self._state.stop_was_partial = False
+ logger.info("Chatterbox configured scan: %s", params.name)
+
+ async def start(self) -> None:
+ if not self._state.configured:
+ raise RuntimeError("Chatterbox scan not configured")
+ self._state.scanner_state = "Scanning"
+ self._state.progress = 0.0
+ for i in range(0, 101, 10):
+ if self._state.scanner_state != "Scanning":
+ return
+ self._state.progress = float(i)
+ await asyncio.sleep(0.05)
+ self._state.scanner_state = "Completed"
+ self._state.progress = 100.0
+ self._save_scan(partial=False)
+ self._state.configured = False
+
+ async def stop(self) -> None:
+ """Graceful stop — write a partial TIFF, transition to Stopped."""
+ if self._state.scanner_state == "Scanning":
+ self._save_scan(partial=True)
+ self._state.stop_was_partial = True
+ self._state.scanner_state = "Stopped"
+ self._state.configured = False
+
+ async def pause(self) -> None:
+ self._state.scanner_state = "Paused"
+
+ async def cancel(self) -> None:
+ self._state.scanner_state = "Idle"
+ self._state.progress = 0.0
+ self._state.configured = False
+ self._state.stop_was_partial = False
+
+ @property
+ def current_scan(self) -> tuple[str, str]:
+ """Return ``(group, name)`` for the most recently configured scan."""
+ return self._state.current_group, self._state.current_scan_name
+
+ def _save_scan(self, partial: bool) -> None:
+ group = self._state.current_group or DEFAULT_GROUP
+ name = self._state.current_scan_name or "scan"
+ if group not in self._state.scans:
+ self._state.scans[group] = {}
+ payload = b"CHATTERBOX_PARTIAL_TIFF_DATA" if partial else b"CHATTERBOX_TIFF_DATA"
+ self._state.scans[group][name] = payload
+
+
+class OdysseyImageRetrievalChatterboxBackend(ImageRetrievalBackend):
+ """Chatterbox image retrieval — reads from the shared state."""
+
+ def __init__(self, state: _OdysseyChatterboxState) -> None:
+ super().__init__()
+ self._state = state
+
+ async def list_groups(self) -> List[str]:
+ return list(self._state.scans.keys())
+
+ async def list_scans(self, group: str) -> List[str]:
+ return list(self._state.scans.get(group, {}).keys())
+
+ async def download(self, group: str, scan_name: str) -> bytes:
+ scans = self._state.scans.get(group, {})
+ if scan_name not in scans:
+ raise FileNotFoundError(f"Scan '{scan_name}' not found in group '{group}'")
+ return scans[scan_name]
+
+ async def download_channel(self, group: str, scan_name: str, channel: int) -> bytes:
+ """Mirrors the real backend's per-channel download.
+
+ The chatterbox stores one blob per scan rather than per channel,
+ so we return that blob for any requested channel — sufficient for
+ cross-capability orchestration tests (e.g. ``stop_and_save``).
+ """
+ return await self.download(group, scan_name)
+
+
+class OdysseyInstrumentStatusChatterboxBackend(InstrumentStatusBackend):
+ """Chatterbox status — reflects the shared state."""
+
+ def __init__(self, state: _OdysseyChatterboxState) -> None:
+ super().__init__()
+ self._state = state
+
+ async def read_status(self) -> InstrumentStatusReading:
+ return InstrumentStatusReading(
+ state=self._state.scanner_state,
+ current_user=self._state.current_user,
+ progress=self._state.progress,
+ time_remaining="",
+ lid_open=self._state.lid_open,
+ )
+
+
+__all__ = [
+ "OdysseyChatterboxDriver",
+ "OdysseyScanningChatterboxBackend",
+ "OdysseyImageRetrievalChatterboxBackend",
+ "OdysseyInstrumentStatusChatterboxBackend",
+]
diff --git a/pylabrobot/li_cor/odyssey/device_card.py b/pylabrobot/li_cor/odyssey/device_card.py
new file mode 100644
index 00000000000..08119ddb218
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/device_card.py
@@ -0,0 +1,69 @@
+"""Odyssey Classic device card — model base.
+
+Specs sourced from the operator's manual (publication 984-11712,
+version 3.0, edition D). The ``identity`` slot is empty: each
+deployment populates it with its own unit's PIDInst Handle URI,
+landing page, and friendly name via an instance card::
+
+ instance = DeviceCard.instance(identity={
+ "pid": "http://hdl.handle.net/21.11157/...",
+ "landing_page": "https://b2inst.gwdg.de/records/...",
+ "name": "Odyssey, Lab 3",
+ })
+ card = ODYSSEY_CLASSIC_BASE.merge(instance)
+"""
+
+from __future__ import annotations
+
+from pylabrobot.device_card import DeviceCard
+
+ODYSSEY_CLASSIC_BASE = DeviceCard(
+ name="Odyssey Classic",
+ vendor="LI-COR Biosciences",
+ model="9120",
+ capabilities={
+ "scanning": {
+ "resolutions_um": [21, 42, 84, 169, 337],
+ "quality_levels": ["lowest", "low", "medium", "high", "highest"],
+ "intensity_range": [0.5, 10.0],
+ "intensity_step": 0.5,
+ "low_intensity_range": [0.5, 2.0], # L0.5 to L2.0
+ "scan_area_cm": [25, 25],
+ "focus_offset_range_mm": [0.0, 4.0],
+ "scanning_speed_cm_s": [5, 40],
+ },
+ "channels": {
+ "700": {
+ "laser_wavelength_nm": 685,
+ "laser_type": "solid-state diode",
+ "laser_peak_power_mw": 80,
+ "detector": "silicon avalanche photodiode",
+ "dichroic_split_nm": 750,
+ },
+ "800": {
+ "laser_wavelength_nm": 785,
+ "laser_type": "solid-state diode",
+ "laser_peak_power_mw": 80,
+ "detector": "silicon avalanche photodiode",
+ "dichroic_split_nm": 810,
+ },
+ },
+ "image_retrieval": {
+ "format": "TIFF",
+ "channels_per_file": 1,
+ "storage_gb": 25,
+ },
+ "instrument_status": {
+ "states": ["Idle", "Scanning", "Paused"],
+ "lid_interlock": True,
+ },
+ },
+ connection={
+ "protocol": "http",
+ "port": 80,
+ "auth": "basic",
+ "network": "10/100Base-T Ethernet",
+ "discovery": "rendezvous/mdns",
+ "cgi_base": "/scanapp/nonjava",
+ },
+)
diff --git a/pylabrobot/li_cor/odyssey/driver.py b/pylabrobot/li_cor/odyssey/driver.py
new file mode 100644
index 00000000000..c1b92d2861a
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/driver.py
@@ -0,0 +1,309 @@
+"""LI-COR Odyssey Classic HTTP transport driver.
+
+Generic HTTP transport over Basic Auth for the Odyssey Classic
+embedded web server. Vendor-specific protocol — CGI paths, form
+encoding, response parsing — lives in the capability backends; this
+driver only ships bytes back and forth.
+
+Server: Apache/1.3.27 (Unix) (Red-Hat/Linux) mod_perl/1.23
+Auth realm: LICOR-Odyssey
+Transport: HTTP over 10/100Base-T Ethernet
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from typing import Any, Awaitable, Callable, Optional, TypeVar
+
+import aiohttp
+
+from pylabrobot.capabilities.capability import BackendParams
+from pylabrobot.device import Driver
+
+logger = logging.getLogger(__name__)
+
+
+# Transport errors worth retrying — connection-level transients only.
+RETRYABLE_EXCEPTIONS = (
+ aiohttp.ClientConnectionError,
+ asyncio.TimeoutError,
+ ConnectionResetError,
+ OSError,
+)
+_HTTP_RETRY_ATTEMPTS = 3
+_HTTP_RETRY_DELAY = 0.25
+
+_T = TypeVar("_T")
+
+# Environment variables for credentials.
+_CRED_ENV_USER = "ODYSSEY_USER"
+_CRED_ENV_PASS = "ODYSSEY_PASS"
+
+
+class OdysseyDriver(Driver):
+ """HTTP transport for the LI-COR Odyssey Classic.
+
+ Wraps an aiohttp.ClientSession with HTTP Basic Auth. Capability
+ backends share a single OdysseyDriver instance and call
+ :meth:`post` / :meth:`get` / :meth:`get_bytes` to exchange bytes
+ with the embedded web server.
+ """
+
+ def __init__(
+ self,
+ host: str,
+ username: str,
+ password: str,
+ port: int = 80,
+ timeout: float = 60.0,
+ ) -> None:
+ super().__init__()
+ if not username or not password:
+ raise ValueError(
+ "OdysseyDriver requires both username and password. "
+ f"Use OdysseyDriver.from_env() to read them from "
+ f"{_CRED_ENV_USER}/{_CRED_ENV_PASS}."
+ )
+ self._host = host
+ self._port = port
+ self._timeout_seconds = timeout
+ self._username = username
+ self._password = password
+ self._base_url = f"http://{host}:{port}"
+ self._auth = aiohttp.BasicAuth(username, password)
+ self._timeout = aiohttp.ClientTimeout(total=timeout)
+ self._session: Optional[aiohttp.ClientSession] = None
+ logger.info(
+ "OdysseyDriver initialised: host=%s %s=%s",
+ host,
+ _CRED_ENV_USER,
+ username,
+ )
+
+ def serialize(self) -> dict:
+ return {
+ **super().serialize(),
+ "host": self._host,
+ "port": self._port,
+ "timeout": self._timeout_seconds,
+ }
+
+ @classmethod
+ def from_env(
+ cls,
+ host: Optional[str] = None,
+ port: int = 80,
+ timeout: float = 60.0,
+ ) -> "OdysseyDriver":
+ """Construct from ODYSSEY_USER / ODYSSEY_PASS env vars.
+
+ Raises ValueError if either is missing — the driver does not
+ silently fall back to default credentials. If ``host`` is None,
+ ODYSSEY_HOST is read from the environment too.
+ """
+ username = os.environ.get(_CRED_ENV_USER, "")
+ password = os.environ.get(_CRED_ENV_PASS, "")
+ if not username or not password:
+ missing = [
+ name
+ for name, val in (
+ (_CRED_ENV_USER, username),
+ (_CRED_ENV_PASS, password),
+ )
+ if not val
+ ]
+ raise ValueError(f"Missing required environment variable(s): {', '.join(missing)}.")
+ if host is None:
+ host = os.environ.get("ODYSSEY_HOST", "")
+ if not host:
+ raise ValueError("No host provided and ODYSSEY_HOST is unset.")
+ return cls(
+ host=host,
+ username=username,
+ password=password,
+ port=port,
+ timeout=timeout,
+ )
+
+ @property
+ def base_url(self) -> str:
+ return self._base_url
+
+ async def setup(self, backend_params: Optional[BackendParams] = None) -> None:
+ """Open the HTTP session and verify reachability.
+
+ ``Connection: close`` is forced on every request: the embedded
+ Apache 1.3.27 doesn't always handle keep-alive cleanly when a
+ second request arrives before the first response is fully
+ consumed. Closing per request is a couple of ms slower but
+ eliminates the inter-request races we see in the field.
+
+ Auth verification is intentionally NOT done here. Backends
+ perform the first auth-protected call; a 401 surfaces there.
+ """
+ self._session = aiohttp.ClientSession(
+ auth=self._auth,
+ timeout=self._timeout,
+ headers={"Connection": "close"},
+ )
+ async with self._session.get(self._base_url) as resp:
+ if resp.status != 200:
+ raise ConnectionError(f"Cannot reach Odyssey at {self._base_url} (HTTP {resp.status})")
+ logger.info("Connected to Odyssey at %s", self._base_url)
+
+ async def stop(self) -> None:
+ """Close the HTTP session."""
+ if self._session is not None:
+ await self._session.close()
+ self._session = None
+
+ def _check_session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ raise RuntimeError("OdysseyDriver not set up")
+ return self._session
+
+ # -- Generic transport ---------------------------------------------------
+
+ async def post(
+ self,
+ path: str,
+ form_data: Optional[dict[str, str]] = None,
+ *,
+ allow_redirects: bool = False,
+ with_retry: bool = False,
+ ) -> tuple[int, str, dict[str, str]]:
+ """POST ``form_data`` to ``path``. Returns (status, body, headers)."""
+ if with_retry:
+ return await self._retry(self._post_once, path, form_data, allow_redirects)
+ return await self._post_once(path, form_data, allow_redirects)
+
+ async def _post_once(
+ self,
+ path: str,
+ form_data: Optional[dict[str, str]],
+ allow_redirects: bool,
+ ) -> tuple[int, str, dict[str, str]]:
+ session = self._check_session()
+ url = f"{self._base_url}{path}"
+ async with session.post(url, data=form_data, allow_redirects=allow_redirects) as resp:
+ body = await resp.text()
+ return resp.status, body, dict(resp.headers)
+
+ async def get(
+ self,
+ path: str,
+ params: Optional[dict[str, str]] = None,
+ *,
+ allow_redirects: bool = True,
+ with_retry: bool = False,
+ ) -> tuple[int, str, dict[str, str]]:
+ """GET ``path`` with ``params``. Returns (status, body, headers)."""
+ if with_retry:
+ return await self._retry(self._get_once, path, params, allow_redirects)
+ return await self._get_once(path, params, allow_redirects)
+
+ async def _get_once(
+ self,
+ path: str,
+ params: Optional[dict[str, str]],
+ allow_redirects: bool,
+ ) -> tuple[int, str, dict[str, str]]:
+ session = self._check_session()
+ url = f"{self._base_url}{path}"
+ async with session.get(url, params=params, allow_redirects=allow_redirects) as resp:
+ body = await resp.text()
+ return resp.status, body, dict(resp.headers)
+
+ async def get_bytes(
+ self,
+ path: str,
+ params: Optional[dict[str, str]] = None,
+ *,
+ with_retry: bool = False,
+ ) -> tuple[int, bytes, dict[str, str], Optional[int]]:
+ """GET binary content. Returns (status, bytes, headers, content_length).
+
+ ``content_length`` is the server-advertised ``Content-Length``
+ header (or None). Truncation checks belong on the caller — this
+ method just returns whatever the socket delivered.
+ """
+ if with_retry:
+ return await self._retry_bytes(path, params)
+ return await self._get_bytes_once(path, params)
+
+ async def _get_bytes_once(
+ self,
+ path: str,
+ params: Optional[dict[str, str]],
+ ) -> tuple[int, bytes, dict[str, str], Optional[int]]:
+ session = self._check_session()
+ url = f"{self._base_url}{path}"
+ async with session.get(url, params=params) as resp:
+ data = await resp.read()
+ return resp.status, data, dict(resp.headers), resp.content_length
+
+ async def _retry(
+ self,
+ method: Callable[..., Awaitable[_T]],
+ *args: Any,
+ ) -> _T:
+ last_exc: Optional[Exception] = None
+ for attempt in range(_HTTP_RETRY_ATTEMPTS):
+ try:
+ return await method(*args)
+ except RETRYABLE_EXCEPTIONS as exc:
+ last_exc = exc
+ if attempt < _HTTP_RETRY_ATTEMPTS - 1:
+ logger.warning(
+ "%s attempt %d/%d failed (%s) — retrying",
+ method.__name__,
+ attempt + 1,
+ _HTTP_RETRY_ATTEMPTS,
+ exc,
+ )
+ await asyncio.sleep(_HTTP_RETRY_DELAY)
+ continue
+ assert last_exc is not None
+ raise last_exc
+
+ async def _retry_bytes(
+ self,
+ path: str,
+ params: Optional[dict[str, str]],
+ ) -> tuple[int, bytes, dict[str, str], Optional[int]]:
+ last_exc: Optional[Exception] = None
+ for attempt in range(_HTTP_RETRY_ATTEMPTS):
+ try:
+ return await self._get_bytes_once(path, params)
+ except RETRYABLE_EXCEPTIONS as exc:
+ last_exc = exc
+ if attempt < _HTTP_RETRY_ATTEMPTS - 1:
+ logger.warning(
+ "GET bytes %s attempt %d/%d failed (%s) — retrying",
+ path,
+ attempt + 1,
+ _HTTP_RETRY_ATTEMPTS,
+ exc,
+ )
+ await asyncio.sleep(_HTTP_RETRY_DELAY)
+ continue
+ assert last_exc is not None
+ raise last_exc
+
+ # -- Admin / non-capability ---------------------------------------------
+
+ async def shutdown_instrument(self) -> str:
+ """Power off the instrument. WARNING: restart can take 30 minutes."""
+ logger.warning("Shutting down Odyssey instrument")
+ _, body, _ = await self.get(
+ "/scanapp/admin/admin/index",
+ params={"action": "InitiateShutdown"},
+ )
+ return body
+
+ async def get_instrument_info(self) -> str:
+ """Fetch instrument info page (serial, software version)."""
+ _, body, _ = await self.get("/scanapp/help/instinfo.pl")
+ return body
diff --git a/pylabrobot/li_cor/odyssey/errors.py b/pylabrobot/li_cor/odyssey/errors.py
new file mode 100644
index 00000000000..18e26b64baa
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/errors.py
@@ -0,0 +1,54 @@
+"""Odyssey driver exceptions — dual-base for capability + vendor reach.
+
+Each backend-level error inherits from BOTH the vendor's driver-level
+exception (:class:`OdysseyError`) AND the capability-generic exception
+(:class:`ScanningError`, :class:`ImageRetrievalError`,
+:class:`InstrumentStatusError`). A single raise is then catchable on
+either axis::
+
+ try:
+ await odyssey.scanning.start()
+ except ScanningError:
+ ... # capability-generic recovery (works for any scanning backend)
+ except OdysseyError:
+ ... # vendor-specific debugging
+"""
+
+from __future__ import annotations
+
+from pylabrobot.capabilities.scanning.image_retrieval import ImageRetrievalError
+from pylabrobot.capabilities.scanning.instrument_status import InstrumentStatusError
+from pylabrobot.capabilities.scanning.scanning import ScanningError
+
+
+class OdysseyError(Exception):
+ """Base exception for the LI-COR Odyssey Classic driver.
+
+ Raised at the connection / transport layer for protocol or HTTP
+ failures. Capability-specific raises use a subclass that also
+ inherits from the matching capability-generic exception.
+ """
+
+
+class OdysseyScanError(OdysseyError, ScanningError):
+ """Odyssey scanning capability failed.
+
+ Catchable as either :class:`OdysseyError` (vendor-specific) or
+ :class:`ScanningError` (capability-generic).
+ """
+
+
+class OdysseyImageError(OdysseyError, ImageRetrievalError):
+ """Odyssey image retrieval failed (download / list / preview)."""
+
+
+class OdysseyStatusError(OdysseyError, InstrumentStatusError):
+ """Odyssey status read failed."""
+
+
+__all__ = [
+ "OdysseyError",
+ "OdysseyScanError",
+ "OdysseyImageError",
+ "OdysseyStatusError",
+]
diff --git a/pylabrobot/li_cor/odyssey/image_retrieval_backend.py b/pylabrobot/li_cor/odyssey/image_retrieval_backend.py
new file mode 100644
index 00000000000..947f1a64476
--- /dev/null
+++ b/pylabrobot/li_cor/odyssey/image_retrieval_backend.py
@@ -0,0 +1,171 @@
+"""Odyssey image retrieval backend — protocol logic for /scan/image.
+
+Owns the image-retrieval protocol: XML query encoding, TIFF /
+JPEG / log GETs, and HTML parsing of the scan-list dropdown. The
+driver only ships the bytes back.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import List
+from urllib.parse import quote
+
+from pylabrobot.capabilities.scanning.image_retrieval import ImageRetrievalBackend
+
+from .driver import OdysseyDriver
+from .errors import OdysseyImageError
+
+logger = logging.getLogger(__name__)
+
+
+_IMAGE_BASE = "/scanapp/imaging/nonjava"
+_SCAN_LIST_PATH = "/scanapp/scan/nonjava/"
+_SCAN_IMAGE_PATH = "/scan/image"
+_SAVELOG_URL_PATH = f"{_IMAGE_BASE}/savelog.pl"
+
+
+def _tiff_xml(group: str, scan_name: str, channel: int) -> str:
+ """Build the XML query string for a TIFF download."""
+ return (
+ f""
+ f"{group}"
+ f"{scan_name}"
+ f"tiff"
+ f"{channel}"
+ f"0000"
+ f""
+ )
+
+
+def _jpeg_xml(
+ group: str,
+ scan_name: str,
+ contrast_700: int = 5,
+ contrast_800: int = 5,
+ channels: str = "700 800",
+ background: str = "black",
+ clip: tuple[int, int, int, int] = (0, 0, 0, 0),
+ vflip: bool = True,
+ hflip: bool = True,
+ zoom: int = 1,
+) -> str:
+ """Build the XML query string for a JPEG preview."""
+ x0, x1, y0, y1 = clip
+ return (
+ f""
+ f"{group}"
+ f"{scan_name}"
+ f"{zoom}"
+ f"{contrast_700}"
+ f"{contrast_800}"
+ f"{channels}"
+ f"{background}"
+ f"{x0}{x1}{y0}{y1}"
+ f"{'true' if vflip else 'false'}"
+ f"{'true' if hflip else 'false'}"
+ f""
+ )
+
+
+def _parse_select_options(html: str, select_name: str) -> List[str]:
+ """Extract