diff --git a/docs/bridge-devices.md b/docs/bridge-devices.md new file mode 100644 index 00000000..c2b0e0bd --- /dev/null +++ b/docs/bridge-devices.md @@ -0,0 +1,254 @@ +# Python Bridge Devices + +This document explains how pymmcore-nano enables Python-implemented devices +to be registered as real devices in CMMCore's device registry, so that all +CMMCore features (auto-shutter, config groups, device enumeration, property +system, etc.) work transparently. + +## Overview + +``` + Python device object C++ bridge device CMMCore + (your code) (bridge_devices.h) (upstream) + ┌──────────────┐ ┌──────────────────┐ ┌───────────┐ + │ MyCamera │◄── GIL ────▶│ PyBridgeCamera │◄──────▶│ │ + │ .snap_image │ forwarding │ : CCameraBase<> │ normal │ device │ + │ .get_exp.. │ │ │ device │ Manager_ │ + │ .set_exp.. │ │ owns nb::object │ calls │ │ + └──────────────┘ └──────────────────┘ └───────────┘ +``` + +Each Python device is wrapped in a C++ "bridge" class that inherits from the +real MM device base (`CCameraBase`, `CShutterBase`, etc.). The bridge +acquires the GIL and forwards every MM method call to the Python object. +CMMCore sees the bridge as a normal device — it lives in `deviceManager_`, +has properties in `PropertyCollection`, and participates in auto-shutter, +config groups, and state enumeration. + +## How It Works (no upstream patches) + +The bridge uses MMCore's existing `MockDeviceAdapter` infrastructure +(originally designed for unit testing). The flow: + +1. Python creates a `DeviceAdapter` and registers device classes +2. `core.loadPyDeviceAdapter(name, adapter)` registers it with CMMCore +3. `core.loadDevice(label, adapterName, deviceName)` instantiates the Python + class and wraps it in the appropriate bridge +4. `core.initializeDevice(label)` calls `Initialize()` on the bridge, which + calls `py_device.initialize(bridge)` with a `PropertyBridge` for + registering properties + +## Two Loading APIs + +### `loadPyDeviceAdapter` — adapter with multiple device classes + +```python +from pymmcore_nano import CMMCore, DeviceAdapter, DeviceType + +adapter = DeviceAdapter() +adapter.add_device_class("MyCam", MyCameraClass, DeviceType.CameraDevice, + "My custom camera") +adapter.add_device_class("MyShutter", MyShutterClass, DeviceType.ShutterDevice, + "My custom shutter") + +core = CMMCore() +core.loadPyDeviceAdapter("MyHardware", adapter) + +# Standard CMMCore device discovery and loading: +core.getAvailableDevices("MyHardware") # ["MyCam", "MyShutter"] +core.loadDevice("Cam1", "MyHardware", "MyCam") +core.initializeDevice("Cam1") +``` + +#### Benefits + +- All devices from one adapter share the same `LoadedDeviceAdapter` mutex, + matching the behavior of real C++ adapters (important for devices that share + a communication bus). +- Easier discovery on the application side +- Could be supported by python entry-points to allow third-party packages to + provide device adapter libraries without requiring explicit registration code + +### `loadPyDevice` — convenience for a single pre-instantiated device + +```python +cam = MyCameraClass() +core.loadPyDevice("Cam1", cam, DeviceType.CameraDevice) +core.initializeDevice("Cam1") +``` + +This creates a single-device adapter internally. The Python object is shared +(not copied) — the bridge holds a reference to the same instance. + +#### Benefits + +- Simpler for users who just want to directly instantiate and load a single + python device without needing to create an adapter library to wrap it. + +## Device Protocols + +The C++ bridge expects Python device objects to implement specific methods. +These are documented as `typing.Protocol` classes in +`pymmcore_nano.protocols`: + +- `PyDevice` — base: `initialize(bridge)`, `shutdown()`, `busy()` +- `PyCamera` — camera methods (snap, exposure, ROI, binning, etc.) +- `PyShutter` — `set_open()`, `get_open()`, `fire()` +- `PyStage` — single-axis positioning +- `PyXYStage` — dual-axis positioning (step-based) +- `PyState` — filter wheel / turret (`get_number_of_positions()`) +- `PyAutoFocus` — continuous/incremental focus, offset, scores +- `PyGeneric` — properties only (no device-specific methods) +- `PyHub` — peripheral discovery (`detect_installed_devices()`) +- `PySLM` — spatial light modulator (image display, exposure) + +These protocols are `@runtime_checkable`. The bridge does not enforce them +at registration time — missing methods will raise `AttributeError` at call +time, just as a missing method on any Python object would. + +## Property System + +MM devices have a string-based property system (name/value pairs with +optional types, limits, and allowed values). The bridge integrates with +this through the `PropertyBridge` object. + +### How it works + +During `initialize(bridge)`, the Python device registers properties: + +```python +def initialize(self, bridge): + bridge.create_property( + "Gain", "1.0", PropertyType.Float, read_only=False, + getter=lambda: self._gain, + setter=lambda v: setattr(self, '_gain', float(v)), + limits=(0.0, 100.0) + ) + # can also set limits separately: + # bridge.set_property_limits("Gain", 0.0, 100.0) + + bridge.create_property( + "Mode", "Normal", PropertyType.String, read_only=False, + getter=lambda: self._mode, + setter=lambda v: setattr(self, '_mode', v), + allowed_values=["Normal", "Fast", "Slow"] + ) + # can also set allowed values separately: + # bridge.set_allowed_values("Mode", ["Normal", "Fast", "Slow"]) +``` + +The `PropertyBridge`: + +- Wraps `CDeviceBase::CreateProperty()` with an `MM::ActionLambda` that + calls the Python getter/setter +- On `BeforeGet` (CMMCore reads the property): calls `getter()`, converts + to string, stores in the MM property +- On `AfterSet` (CMMCore writes the property): reads the string value from + the MM property, passes to `setter(value_str)` +- `set_property_limits()` and `set_allowed_values()` wrap the corresponding + `CDeviceBase` methods + +After `initialize()` returns, the `PropertyBridge` is invalidated. Calling +its methods after that raises `RuntimeError`. This is enforced via a shared +validity flag that survives even if Python stores a reference to the bridge. + +### Property lifecycle + +``` +bridge.create_property("Gain", ...) + → CDeviceBase::CreateProperty("Gain", "1.0", Float, false, ActionLambda) + → MM::PropertyCollection stores MM::FloatProperty with the lambda + +core.getProperty("dev", "Gain") + → CDeviceBase::GetProperty → PropertyCollection::Get + → MM::FloatProperty::Update → ActionLambda(BeforeGet) + → GIL acquire → getter() → pProp->Set(str(value)) + → returns string value to CMMCore + +core.setProperty("dev", "Gain", "42.5") + → CDeviceBase::SetProperty → PropertyCollection::Set + → validates limits (0.0-100.0) → MM::FloatProperty::Set("42.5") + → MM::FloatProperty::Apply → ActionLambda(AfterSet) + → GIL acquire → pProp->Get(val) → setter("42.5000") +``` + +CDeviceBase handles validation (limits, allowed values, read-only checks) +before the lambda is ever called. + +## Key Files + +| File | Purpose | +|------|---------| +| `src/bridge_devices.h` | All C++ bridge device classes + PropertyBridge + PyBridgeAdapter | +| `src/_pymmcore_nano.cc` | nanobind bindings for `DeviceAdapter`, `PropertyBridge`, `loadPyDevice`, `loadPyDeviceAdapter` | +| `src/pymmcore_nano/protocols.py` | Python `Protocol` classes documenting the bridge contract | +| `tests/test_bridge_devices.py` | Tests for all device types, properties, and adapter loading | + +## Supported Device Types + +| Type | Bridge Class | Base Class | Python Protocol | +|------|-------------|------------|-----------------| +| Camera | `PyBridgeCamera` | `CCameraBase<>` | `PyCamera` | +| Shutter | `PyBridgeShutter` | `CShutterBase<>` | `PyShutter` | +| Stage | `PyBridgeStage` | `CStageBase<>` | `PyStage` | +| XYStage | `PyBridgeXYStage` | `CXYStageBase<>` | `PyXYStage` | +| State | `PyBridgeState` | `CStateDeviceBase<>` | `PyState` | +| AutoFocus | `PyBridgeAutoFocus` | `CAutoFocusBase<>` | `PyAutoFocus` | +| Generic | `PyBridgeGeneric` | `CGenericBase<>` | `PyGeneric` | +| Hub | `PyBridgeHub` | `HubBase<>` | `PyHub` | +| SLM | `PyBridgeSLM` | `CSLMBase<>` | `PySLM` | + +## Adding a New Device Type + +1. In `bridge_devices.h`: + - Create a new class inheriting from the appropriate `C*Base<>` + - Implement pure virtuals that don't have base defaults + - Use `py_get`, `py_set`, `py_call` helpers for method forwarding + - Use `initializeWithBridge(this, py_)` in `Initialize()` + - Store `deviceName_` for `GetName()` +2. Add the type to `createBridgeDevice()` switch +3. In `protocols.py`: add a `Py*` protocol class +4. In `tests/test_bridge_devices.py`: add a minimal stub and test +5. Rebuild and run tests + +## GIL and Threading + +- Every bridge method that calls into Python acquires the GIL via + `nb::gil_scoped_acquire` +- `GetImageBuffer()` returns a C++-owned buffer pointer without the GIL +- GIL acquisition is re-entrant (safe for nested calls) +- `std::atomic capturing_` is used for `IsCapturing()` to avoid + GIL acquisition on a hot path + +## Lifecycle and Cleanup + +- Bridge devices hold `nb::object` references to Python devices +- Destructors acquire the GIL before releasing references (`py_.reset()`) +- All destructors are wrapped in `try/catch(...)` to prevent + `std::terminate` if GIL acquisition fails during shutdown +- `PyBridgeAdapter` destructors similarly acquire GIL before clearing + the device vector +- Adapter capsules are stored in a module-level `_bridge_adapters` dict + (not a static C++ map) to ensure cleanup happens during Python GC, + not during static destruction after interpreter shutdown +- `PropertyBridge` getter/setter callables are wrapped in a + `shared_ptr` whose destructor acquires the GIL — this + handles the case where `~CDeviceBase` destroys `ActionLambda` captures + without the GIL + +## Known Limitations + +- **Sequence acquisition**: The bridge's `StartSequenceAcquisition` calls + into Python, but there is no mechanism for Python to push frames into + CMMCore's circular buffer via `InsertImage`. This requires further design. +- **`IsCapturing` flag**: Managed by the C++ bridge, not forwarded to Python. + If the Python device's sequence finishes independently, the flag may be + stale. +- **RGB cameras**: `GetNumberOfComponents()` defaults to 1 (from + `CCameraBase`). Not currently forwarded to Python. +- **Adapter cleanup on CMMCore destruction**: The `MockDeviceAdapter` API + provides no unload notification. Adapter capsules may outlive the CMMCore + that references them (benign — they hold Python objects, not C++ pointers). +- **SLM `set_image` array shapes**: The 8-bit overload passes a flat + `uint8` array of total bytes; the 32-bit overload passes a `uint32` + array of pixel count. Python implementations must handle both. diff --git a/meson.build b/meson.build index e9ab9488..77a28c19 100644 --- a/meson.build +++ b/meson.build @@ -88,6 +88,7 @@ custom_target( py.install_sources( files( 'src/pymmcore_nano/__init__.py', + 'src/pymmcore_nano/protocols.py', 'src/pymmcore_nano/py.typed', ), subdir: 'pymmcore_nano', diff --git a/src/_pymmcore_nano.cc b/src/_pymmcore_nano.cc index 41f9debe..afe793a2 100644 --- a/src/_pymmcore_nano.cc +++ b/src/_pymmcore_nano.cc @@ -12,6 +12,8 @@ #include "MMEventCallback.h" #include "ModuleInterface.h" +#include "bridge_devices.h" + namespace nb = nanobind; using namespace nb::literals; @@ -176,7 +178,7 @@ np_array create_metadata_array(CMMCore &core, void *pBuf, const Metadata md) { } void validate_slm_image(const nb::ndarray &pixels, long expectedWidth, - long expectedHeight, long bytesPerPixel) { + long expectedHeight, long bytesPerPixel, long nComponents) { // Check dtype if (pixels.dtype() != nb::dtype()) { throw std::invalid_argument("Pixel array type is wrong. Expected uint8."); @@ -186,7 +188,7 @@ void validate_slm_image(const nb::ndarray &pixels, long expectedWidth, if (pixels.ndim() != 2 && pixels.ndim() != 3) { throw std::invalid_argument( "Pixels must be a 2D numpy array [h,w] of uint8, or a 3D numpy array " - "[h,w,c] of uint8 with 3 color channels [R,G,B]."); + "[h,w,c] of uint8 with 3 or 4 color channels."); } // Check shape @@ -198,17 +200,13 @@ void validate_slm_image(const nb::ndarray &pixels, long expectedWidth, std::to_string(pixels.shape(1)) + ")."); } - // Check total bytes - long expectedBytes = expectedWidth * expectedHeight * bytesPerPixel; - if (pixels.nbytes() != expectedBytes) { + // Check total bytes (accounts for multi-component pixels like RGB) + long expectedBytes = expectedWidth * expectedHeight * bytesPerPixel * nComponents; + if (static_cast(pixels.nbytes()) != expectedBytes) { throw std::invalid_argument("Image size is wrong for this SLM. Expected " + std::to_string(expectedBytes) + " bytes, but received " + - std::to_string(pixels.nbytes()) + - " bytes. Does this SLM support RGB?"); + std::to_string(pixels.nbytes()) + " bytes."); } - - // Ensure C-contiguous layout - // TODO } ///////////////// Trampoline class for MMEventCallback /////////////////// @@ -280,6 +278,19 @@ class PyMMEventCallback : public MMEventCallback { } }; +//////////////////////////////////////////////////////////////////////////// +///////////////// Bridge adapter storage helper //////////////////////////// +//////////////////////////////////////////////////////////////////////////// + +// Register a PyBridgeAdapter with CMMCore. CMMCore takes ownership of the +// adapter (via unique_ptr in LoadedDeviceAdapterImplMock). The adapter is +// destroyed when CMMCore unloads it (unloadLibrary/destructor). +void registerAndStoreBridgeAdapter(CMMCore &core, const std::string &adapterName, + std::unique_ptr adapter) { + adapter->markLoaded(); + core.loadMockDeviceAdapter(adapterName.c_str(), adapter.release()); +} + //////////////////////////////////////////////////////////////////////////// ///////////////// main _pymmcore_nano module definition /////////////////// //////////////////////////////////////////////////////////////////////////// @@ -290,6 +301,54 @@ NB_MODULE(_pymmcore_nano, m) { m.doc() = "Python bindings for MMCore"; + // DeviceAdapter — Python-visible wrapper for PyBridgeAdapter. + // Allows Python to build an adapter by adding device classes, then + // register it with CMMCore. This keeps device discovery logic in Python. + nb::class_( + m, "DeviceAdapter", + "A collection of Python device classes that acts as an MM device " + "adapter library. Add device classes with add_device_class(), then " + "register with core.loadPyDeviceAdapter(name, adapter).") + .def(nb::init<>()) + .def("add_device_class", &PyBridgeAdapter::addDeviceClass, "name"_a, "device_class"_a, + "device_type"_a, "description"_a, + nb::sig("def add_device_class(self, name: str, device_class: type," + " device_type: DeviceType, description: str) -> None")); + + // PropertyHandle — returned by create_property() for dynamic + // constraint updates (limits, allowed values). Valid for the + // device's entire lifetime. + nb::class_( + m, "PropertyHandle", + "Handle for updating an MM property's limits or allowed values. " + "Returned by the create_property() callable passed to initialize().") + .def("set_limits", &PropertyHandle::setLimits, "lower"_a, "upper"_a) + .def("set_allowed_values", &PropertyHandle::setAllowedValues, "values"_a) + .def("set_sequence_max_length", &PropertyHandle::setSequenceMaxLength, "max_length"_a); + + // DeviceCallbacks — bridge between Python device objects and C++ + // infrastructure they can't access directly. In C++, a device adapter + // calls this->SetPositionLabel(), this->GetCoreCallback()->OnExposureChanged(), + // etc. — it's calling methods on itself or on the core callback. But in + // the bridge, the Python device and the C++ bridge device are two separate + // objects, so DeviceCallbacks provides the channel for Python devices to + // reach back into CMMCore notifications and device base class methods. + // Passed to initialize() as the second argument. Valid for the + // device's entire lifetime. + nb::class_(m, "DeviceCallbacks", + "Notification callbacks for communicating device state changes " + "to CMMCore. Passed to initialize() as the second argument.") + .def("on_property_changed", &DeviceCallbacks::onPropertyChanged, "name"_a, "value"_a) + .def("on_properties_changed", &DeviceCallbacks::onPropertiesChanged) + .def("on_stage_position_changed", &DeviceCallbacks::onStagePositionChanged, "pos"_a) + .def("on_xy_stage_position_changed", &DeviceCallbacks::onXYStagePositionChanged, "x"_a, + "y"_a) + .def("on_exposure_changed", &DeviceCallbacks::onExposureChanged, "exposure"_a) + .def("on_shutter_open_changed", &DeviceCallbacks::onShutterOpenChanged, "open"_a) + .def("log_message", &DeviceCallbacks::logMessage, "msg"_a, "debug_only"_a = false) + .def("acq_finished", &DeviceCallbacks::acqFinished, "status_code"_a = 0) + .def("set_position_label", &DeviceCallbacks::setPositionLabel, "pos"_a, "label"_a); + /////////////////// Module Attributes /////////////////// m.attr("DEVICE_INTERFACE_VERSION") = CMMCore::getMMDeviceDeviceInterfaceVersion(); @@ -1061,21 +1120,30 @@ MMCore will send notifications on internal events using this interface .def("getPixelSizeAffine", [](CMMCore &self) { std::vector v; - { nb::gil_scoped_release gil; v = self.getPixelSizeAffine(); } + { + nb::gil_scoped_release gil; + v = self.getPixelSizeAffine(); + } return nb::make_tuple(v[0], v[1], v[2], v[3], v[4], v[5]); }, nb::sig("def getPixelSizeAffine(self) -> tuple[float, float, float, float, float, float]")) .def("getPixelSizeAffine", [](CMMCore &self, bool cached) { std::vector v; - { nb::gil_scoped_release gil; v = self.getPixelSizeAffine(cached); } + { + nb::gil_scoped_release gil; + v = self.getPixelSizeAffine(cached); + } return nb::make_tuple(v[0], v[1], v[2], v[3], v[4], v[5]); }, "cached"_a, nb::sig("def getPixelSizeAffine(self, cached: bool) -> tuple[float, float, float, float, float, float]")) .def("getPixelSizeAffineByID", [](CMMCore &self, const char *resolutionID) { std::vector v; - { nb::gil_scoped_release gil; v = self.getPixelSizeAffineByID(resolutionID); } + { + nb::gil_scoped_release gil; + v = self.getPixelSizeAffineByID(resolutionID); + } return nb::make_tuple(v[0], v[1], v[2], v[3], v[4], v[5]); }, "resolutionID"_a, nb::sig("def getPixelSizeAffineByID(self, resolutionID: str) -> tuple[float, float, float, float, float, float]")) @@ -1238,7 +1306,34 @@ MMCore will send notifications on internal events using this interface } RGIL) .def("popNextImage", [](CMMCore &self) -> np_array { - return create_image_array(self, self.popNextImage()); + // Use the MD fast-path: CircularBuffer stores Width/Height/PixelType + // tags, so we can build the np_array without querying the camera + // (which would cross the Python bridge for PyBridgeCamera devices). + // + // TRADEOFF: For C++ cameras, this replaces 4 inline int accesses + // (create_image_array) with 2 std::stoi + a PixelType string-compare + // per frame — measured ~5% slowdown in a tight-pop loop, invisible + // in any realistic acquisition. For PyBridgeCamera (python devices) + // it saves 4 Python bridge crossings (~8us) per frame, which is a + // huge win (see bridge_devices.h StartSequenceAcquisition comment). + // + // ALTERNATIVES if the C++ regression ever matters: + // (a) Cache dims in pymmcore-plus' popNextImage wrapper at + // startSequenceAcquisition and reuse — only works when going + // through CMMCorePlus. + // (b) Cache dims in CMMCore itself at startSequenceAcquisition + // (cleanest, but touches upstream MMCore). + // (c) Expose popNextImageFast as a separate method, leaving + // popNextImage unchanged, and opt-in from pymmcore-plus for + // the python-device case. + // + // Safety: CircularBuffer::InsertImage always writes Width/Height/ + // PixelType (CircularBuffer.cpp ~264-280). If the tags are missing + // or the PixelType is exotic, create_metadata_array falls back to + // create_image_array(self, pBuf) automatically. + Metadata md; + auto img = self.popNextImageMD(md); + return create_metadata_array(self, img, md); } RGIL) // this is a new overload that returns both the image and the metadata // not present in the original C++ API @@ -1565,7 +1660,9 @@ MMCore will send notifications on internal events using this interface long expectedWidth = self.getSLMWidth(slmLabel); long expectedHeight = self.getSLMHeight(slmLabel); long bytesPerPixel = self.getSLMBytesPerPixel(slmLabel); - validate_slm_image(pixels, expectedWidth, expectedHeight, bytesPerPixel); + long nComponents = self.getSLMNumberOfComponents(slmLabel); + validate_slm_image(pixels, expectedWidth, expectedHeight, bytesPerPixel, + nComponents); // Cast the numpy array to a pointer to unsigned char self.setSLMImage(slmLabel, reinterpret_cast(pixels.data())); @@ -1602,9 +1699,11 @@ MMCore will send notifications on internal events using this interface long expectedWidth = self.getSLMWidth(slmLabel); long expectedHeight = self.getSLMHeight(slmLabel); long bytesPerPixel = self.getSLMBytesPerPixel(slmLabel); + long nComponents = self.getSLMNumberOfComponents(slmLabel); std::vector inputVector; for (auto &image : imageSequence) { - validate_slm_image(image, expectedWidth, expectedHeight, bytesPerPixel); + validate_slm_image(image, expectedWidth, expectedHeight, bytesPerPixel, + nComponents); inputVector.push_back(reinterpret_cast(image.data())); } self.loadSLMSequence(slmLabel, inputVector); @@ -1692,5 +1791,40 @@ MMCore will send notifications on internal events using this interface "peripheralLabel"_a RGIL) .def("getLoadedPeripheralDevices", &CMMCore::getLoadedPeripheralDevices, "hubLabel"_a RGIL) + + // Python Bridge Devices + .def("loadPyDevice", + [](CMMCore& self, const char* label, nb::object py_device, + MM::DeviceType type) { + // Convenience: create a single-device adapter and load it. + // Counter ensures unique adapter names across load/unload + // cycles for the same label. + static std::atomic counter{0}; + auto adapter = std::make_unique(); + adapter->addDevice(label, py_device, type); + + std::string adapterName = "_PyBridge_" + std::to_string(counter.fetch_add(1)); + registerAndStoreBridgeAdapter(self, adapterName, std::move(adapter)); + { + nb::gil_scoped_release release; + self.loadDevice(label, adapterName.c_str(), label); + } + }, + "label"_a, "py_device"_a, "type"_a) + + .def("loadPyDeviceAdapter", + [](CMMCore& self, const char* adapterName, + PyBridgeAdapter* adapter) { + // Move the adapter contents into a new owned instance. + // The Python-side DeviceAdapter is left in a "loaded" + // state — add_device/add_device_class will raise if + // called again. + auto owned = std::make_unique(std::move(*adapter)); + registerAndStoreBridgeAdapter(self, adapterName, std::move(owned)); + }, + "adapter_name"_a, "adapter"_a, + nb::sig("def loadPyDeviceAdapter(self, adapter_name: str," + " adapter: DeviceAdapter) -> None")) + ; } diff --git a/src/bridge_devices.h b/src/bridge_devices.h new file mode 100644 index 00000000..68d423fc --- /dev/null +++ b/src/bridge_devices.h @@ -0,0 +1,1408 @@ +// Bridge device classes that forward MM::Device calls to Python objects. +// These enable Python-implemented devices to be registered as real devices +// in CMMCore's device registry via the MockDeviceAdapter infrastructure. + +#pragma once + +#include +#include +#include +#include + +#include "DeviceBase.h" +#include "ImageMetadata.h" +#include "MMDevice.h" +#include "MockDeviceAdapter.h" + +#include +#include +#include +#include + +namespace nb = nanobind; + +// ============================================================================ +// Helpers — factor out GIL + cast boilerplate +// ============================================================================ + +// Catch nb::python_error from Python calls and rethrow as std::runtime_error +// with the Python traceback preserved. CMMCore's DeviceInstance layer catches +// std::exception, logs via LOG_ERROR, and wraps in CMMError — so the Python +// error info surfaces through MM's existing error pipeline. + +template T py_get(const nb::object &py, const char *attr) { + nb::gil_scoped_acquire gil; + try { + return nb::cast(py.attr(attr)()); + } catch (nb::python_error &e) { + std::string msg = e.what(); + e.restore(); + PyErr_Clear(); + throw std::runtime_error(msg); + } +} + +template +void py_set(const nb::object &py, const char *attr, Args &&...args) { + nb::gil_scoped_acquire gil; + try { + py.attr(attr)(std::forward(args)...); + } catch (nb::python_error &e) { + std::string msg = e.what(); + e.restore(); + PyErr_Clear(); + throw std::runtime_error(msg); + } +} + +template +int py_call(const nb::object &py, const char *attr, Args &&...args) { + nb::gil_scoped_acquire gil; + try { + py.attr(attr)(std::forward(args)...); + } catch (nb::python_error &e) { + std::string msg = e.what(); + e.restore(); + PyErr_Clear(); + throw std::runtime_error(msg); + } + return DEVICE_OK; +} + +// GIL-acquiring wrapper for inline Python calls that aren't simple +// get/set/call. Catches nb::python_error and rethrows as std::runtime_error. +template auto py_invoke(F &&fn) -> decltype(fn()) { + nb::gil_scoped_acquire gil; + try { + return fn(); + } catch (nb::python_error &e) { + std::string msg = e.what(); + e.restore(); + PyErr_Clear(); + throw std::runtime_error(msg); + } +} + +// ============================================================================ +// PropertyHandle — returned by create_property(), allows dynamic updates +// to a single property's limits and allowed values. Valid for the device's +// entire lifetime (the dev_ pointer lives as long as CMMCore owns the device). +// ============================================================================ + +class PropertyHandle { + MM::Device *dev_ = nullptr; + std::string name_; + std::shared_ptr> alive_; + // Shared with the ActionLambda in PyCallbacks so runtime updates propagate. + std::shared_ptr> seqMaxLength_; + + struct Vtable { + int (*setPropertyLimits)(MM::Device *, const char *, double, double); + int (*setAllowedValues)(MM::Device *, const char *, std::vector &); + }; + Vtable vt_{}; + + void checkAlive() const { + if (!alive_ || !*alive_) + throw std::runtime_error("Device has been unloaded"); + } + + public: + PropertyHandle() = delete; + + template + PropertyHandle(TDevice *dev, std::string name, std::shared_ptr> alive, + std::shared_ptr> seqMaxLength = nullptr) + : dev_(dev), name_(std::move(name)), alive_(std::move(alive)), + seqMaxLength_(std::move(seqMaxLength)) { + vt_.setPropertyLimits = [](MM::Device *d, const char *n, double lo, double hi) -> int { + return static_cast(d)->SetPropertyLimits(n, lo, hi); + }; + vt_.setAllowedValues = [](MM::Device *d, const char *n, + std::vector &vals) -> int { + return static_cast(d)->SetAllowedValues(n, vals); + }; + } + + void setLimits(double lo, double hi) { + checkAlive(); + vt_.setPropertyLimits(dev_, name_.c_str(), lo, hi); + } + + void setAllowedValues(std::vector values) { + checkAlive(); + vt_.setAllowedValues(dev_, name_.c_str(), values); + } + + void setSequenceMaxLength(long maxLength) { + checkAlive(); + if (seqMaxLength_) + *seqMaxLength_ = maxLength; + } +}; + +// ============================================================================ +// DeviceCallbacks — passed to Python's initialize() so the device can send +// notifications to CMMCore (property changes, position updates, etc.). +// Valid for the device's entire lifetime. Shares the alive_ flag with the +// bridge device. +// ============================================================================ + +class DeviceCallbacks { + MM::Device *dev_ = nullptr; + MM::Core *cb_ = nullptr; + std::shared_ptr> alive_; + + // Type-erased SetPositionLabel — only populated for state devices. + using SetPosLabelFn = int (*)(MM::Device *, long, const char *); + SetPosLabelFn setPositionLabel_ = nullptr; + + void checkAlive() const { + if (!alive_ || !*alive_) + throw std::runtime_error("Device has been unloaded"); + } + + public: + DeviceCallbacks() = delete; + + DeviceCallbacks(MM::Device *dev, MM::Core *cb, std::shared_ptr> alive) + : dev_(dev), cb_(cb), alive_(std::move(alive)) {} + + // Called by initializeWithPropertyFactory for state devices. + void enableSetPositionLabel(SetPosLabelFn fn) { setPositionLabel_ = fn; } + + void onPropertyChanged(const std::string &name, const std::string &value) { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnPropertyChanged(dev_, name.c_str(), value.c_str()); + } + + void onPropertiesChanged() { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnPropertiesChanged(dev_); + } + + void onStagePositionChanged(double pos) { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnStagePositionChanged(dev_, pos); + } + + void onXYStagePositionChanged(double x, double y) { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnXYStagePositionChanged(dev_, x, y); + } + + void onExposureChanged(double newExposure) { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnExposureChanged(dev_, newExposure); + } + + void onShutterOpenChanged(bool open) { + checkAlive(); + nb::gil_scoped_release release; + cb_->OnShutterOpenChanged(dev_, open); + } + + void logMessage(const std::string &msg, bool debugOnly = false) { + checkAlive(); + nb::gil_scoped_release release; + cb_->LogMessage(dev_, msg.c_str(), debugOnly); + } + + void acqFinished(int statusCode = 0) { + checkAlive(); + nb::gil_scoped_release release; + cb_->AcqFinished(dev_, statusCode); + } + + void setPositionLabel(long pos, const std::string &label) { + checkAlive(); + if (!setPositionLabel_) + throw std::runtime_error("setPositionLabel is only available on State devices"); + setPositionLabel_(dev_, pos, label.c_str()); + } +}; + +// ============================================================================ +// createPropertyFactory — builds the create_property callable that is passed +// to Python's initialize(create_property, notify). +// +// Each call to create_property registers an MM property on the C++ device +// with an ActionLambda for get/set, and returns a PropertyHandle for +// dynamic constraint updates. +// ============================================================================ + +// GIL-safe destructor for Python objects captured in ActionLambda. +// CDeviceBase may destroy the ActionLambda without the GIL held. +struct PyCallbacks { + nb::object getter; + nb::object setter; + // Sequence callbacks — none() if not sequenceable. + nb::object seq_loader; // (list[str]) -> None + nb::object seq_starter; // () -> None + nb::object seq_stopper; // () -> None + // Shared with PropertyHandle so runtime updates propagate. + std::shared_ptr> seq_max_length; + + ~PyCallbacks() { + try { + nb::gil_scoped_acquire gil; + getter.reset(); + setter.reset(); + seq_loader.reset(); + seq_starter.reset(); + seq_stopper.reset(); + } catch (...) { + } + } +}; + +// Build an ActionLambda that forwards property actions to Python callables. +// Handles get/set and optionally sequencing (IsSequenceable, AfterLoadSequence, +// StartSequence, StopSequence). +// Returns the shared seq_max_length pointer (may be null if not sequenceable) +// so createPropertyFactory can pass it to PropertyHandle. +inline std::pair, std::shared_ptr>> +makePropertyAction(nb::object getter, nb::object setter, long seqMaxLength, + nb::object seqLoader, nb::object seqStarter, nb::object seqStopper) { + bool hasGetSet = !getter.is_none() || !setter.is_none(); + // maxLength == 0 means not sequenceable — callbacks are ignored even if + // provided, since CMMCore won't invoke them on a non-sequenceable property. + bool hasSeq = seqMaxLength > 0; + if (!hasGetSet && !hasSeq) + return {nullptr, nullptr}; + + auto seqMaxPtr = std::make_shared>(seqMaxLength); + auto cbs = std::make_shared( + PyCallbacks{getter, setter, seqLoader, seqStarter, seqStopper, seqMaxPtr}); + auto functor = std::make_unique( + [cbs](MM::PropertyBase *pProp, MM::ActionType eAct) -> int { + nb::gil_scoped_acquire gil; + try { + if (eAct == MM::BeforeGet && !cbs->getter.is_none()) { + auto val = cbs->getter(); + pProp->Set(nb::cast(nb::str(val)).c_str()); + } else if (eAct == MM::AfterSet && !cbs->setter.is_none()) { + std::string val; + pProp->Get(val); + cbs->setter(nb::str(val.c_str())); + } else if (eAct == MM::IsSequenceable) { + long maxLen = cbs->seq_max_length ? cbs->seq_max_length->load() : 0; + pProp->SetSequenceable(maxLen); + } else if (eAct == MM::AfterLoadSequence && !cbs->seq_loader.is_none()) { + // Convert the C++ string sequence to a Python list + auto seq = pProp->GetSequence(); + nb::list py_seq; + for (auto &s : seq) + py_seq.append(nb::str(s.c_str())); + cbs->seq_loader(py_seq); + } else if (eAct == MM::StartSequence && !cbs->seq_starter.is_none()) { + cbs->seq_starter(); + } else if (eAct == MM::StopSequence && !cbs->seq_stopper.is_none()) { + cbs->seq_stopper(); + } + } catch (nb::python_error &e) { + std::string msg = e.what(); + e.restore(); + PyErr_Clear(); + throw std::runtime_error(msg); + } + return DEVICE_OK; + }); + return {std::move(functor), seqMaxPtr}; +} + +// ============================================================================ +// createPropertyFactory — builds the create_property callable that is passed +// to Python's initialize(create_property). +// +// Each call registers an MM property and returns a PropertyHandle for +// dynamic constraint updates. The factory is invalidated after initialize() +// returns — calling it later raises RuntimeError. +// ============================================================================ + +template +nb::object createPropertyFactory(TDevice *dev, std::shared_ptr> canCreate, + std::shared_ptr> alive) { + // Type-erased CreateProperty + auto doCreate = [](MM::Device *d, const char *name, const char *val, MM::PropertyType t, + bool ro, MM::ActionFunctor *act, bool preInit) -> int { + return static_cast(d)->CreateProperty(name, val, t, ro, act, preInit); + }; + + return nb::cpp_function( + [dev, doCreate, canCreate, alive]( + const std::string &name, const std::string &defaultValue, int mmType, bool readOnly, + nb::object getter, nb::object setter, bool preInit, nb::object limits, + nb::object allowedValues, long sequenceMaxLength, nb::object sequenceLoader, + nb::object sequenceStarter, nb::object sequenceStopper) -> PropertyHandle { + if (!*canCreate) + throw std::runtime_error("create_property() can only be called during " + "initialize()"); + + auto [action, seqMaxPtr] = + makePropertyAction(getter, setter, sequenceMaxLength, sequenceLoader, + sequenceStarter, sequenceStopper); + + int ret = doCreate(dev, name.c_str(), defaultValue.c_str(), + static_cast(mmType), readOnly, action.get(), + preInit); + if (ret == DEVICE_OK) + action.release(); + + PropertyHandle handle(dev, name, alive, seqMaxPtr); + + if (!limits.is_none()) { + double lo = nb::cast(limits[nb::int_(0)]); + double hi = nb::cast(limits[nb::int_(1)]); + handle.setLimits(lo, hi); + } + if (!allowedValues.is_none()) { + std::vector vals; + for (auto v : allowedValues) + vals.push_back(nb::cast(nb::str(v))); + handle.setAllowedValues(vals); + } + + return handle; + }, + nb::arg("name"), nb::arg("default_value"), nb::arg("mm_type"), nb::arg("read_only"), + nb::kw_only(), nb::arg("getter") = nb::none(), nb::arg("setter") = nb::none(), + nb::arg("pre_init") = false, nb::arg("limits") = nb::none(), + nb::arg("allowed_values") = nb::none(), nb::arg("sequence_max_length") = 0, + nb::arg("sequence_loader") = nb::none(), nb::arg("sequence_starter") = nb::none(), + nb::arg("sequence_stopper") = nb::none()); +} + +// ============================================================================ +// initializeWithPropertyFactory — calls Python's +// initialize(create_property, notify) +// The create_property callable is invalidated after initialize() returns. +// The DeviceCallbacks object remains valid for the device's lifetime. +// ============================================================================ + +template +int initializeWithPropertyFactory(TDevice *dev, nb::object &py, + std::shared_ptr> alive, + MM::Core *coreCallback) { + auto canCreate = std::make_shared>(true); + nb::object factory = createPropertyFactory(dev, canCreate, alive); + + // Create DeviceCallbacks — valid for the device's lifetime. + // Heap-allocated, Python takes ownership. + auto *notify = new DeviceCallbacks(dev, coreCallback, alive); + + // Enable SetPositionLabel for state devices. + if constexpr (std::is_base_of_v, TDevice>) { + notify->enableSetPositionLabel([](MM::Device *d, long pos, const char *label) -> int { + return static_cast(d)->SetPositionLabel(pos, label); + }); + } + + nb::object py_notify = nb::cast(notify, nb::rv_policy::take_ownership); + + try { + py.attr("initialize_bridge")(factory, py_notify); + } catch (...) { + *canCreate = false; + throw; + } + *canCreate = false; + return DEVICE_OK; +} + +// ============================================================================ +// Common helper for all bridge device classes. +// Provides shared state and behavior for Initialize/Shutdown/Busy/GetName. +// ============================================================================ + +template class PyBridgeDeviceBase { + protected: + nb::object py_; + std::string deviceName_; + std::string deviceDescription_; + std::shared_ptr> alive_ = std::make_shared>(true); + + PyBridgeDeviceBase(nb::object py_dev, std::string name, std::string description = "") + : py_(std::move(py_dev)), deviceName_(std::move(name)), + deviceDescription_(std::move(description)) {} + + ~PyBridgeDeviceBase() { + try { + nb::gil_scoped_acquire gil; + py_.reset(); + } catch (...) { + } + } + + int initializeCommon(TDevice *dev, MM::Core *coreCallback) { + nb::gil_scoped_acquire gil; + return initializeWithPropertyFactory(dev, py_, alive_, coreCallback); + } + + int shutdownCommon() { + *alive_ = false; + return py_call(py_, "shutdown"); + } + + bool busyCommon() { return py_get(py_, "busy"); } + + void getNameCommon(char *name) const { + CDeviceUtils::CopyLimitedString(name, deviceName_.c_str()); + } + + void getDescriptionCommon(char *desc) const { + CDeviceUtils::CopyLimitedString(desc, deviceDescription_.c_str()); + } +}; + +#define PYBRIDGE_COMMON_OVERRIDES(ClassName) \ + public: \ + ClassName(nb::object py_dev, std::string name, std::string description = "") \ + : PyBridgeDeviceBase(std::move(py_dev), std::move(name), \ + std::move(description)) {} \ + int Initialize() override { \ + return this->initializeCommon(this, this->GetCoreCallback()); \ + } \ + int Shutdown() override { return this->shutdownCommon(); } \ + bool Busy() override { return this->busyCommon(); } \ + void GetName(char *name) const override { this->getNameCommon(name); } \ + void GetDescription(char *desc) const override { this->getDescriptionCommon(desc); } + +// ============================================================================ +// PyBridgeCamera +// ============================================================================ + +class PyBridgeCamera : public CCameraBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeCamera) + + ~PyBridgeCamera() { + try { + nb::gil_scoped_acquire gil; + img_arr_.reset(); + } catch (...) { + } + } + + nb::object img_arr_; // holds ndarray from get_image_buffer() to prevent GC + + // -- MM::Camera: getters -- + unsigned GetImageWidth() const override { return py_get(py_, "get_image_width"); } + unsigned GetImageHeight() const override { + return py_get(py_, "get_image_height"); + } + unsigned GetImageBytesPerPixel() const override { + return py_get(py_, "get_bytes_per_pixel"); + } + unsigned GetNumberOfComponents() const override { + return py_get(py_, "get_number_of_components"); + } + unsigned GetNumberOfChannels() const override { + return py_get(py_, "get_number_of_channels"); + } + int GetChannelName(unsigned channel, char *name) override { + return py_invoke([&]() -> int { + auto pyName = nb::cast(py_.attr("get_channel_name")(channel)); + CDeviceUtils::CopyLimitedString(name, pyName.c_str()); + return DEVICE_OK; + }); + } + unsigned GetBitDepth() const override { return py_get(py_, "get_bit_depth"); } + long GetImageBufferSize() const override { + return py_get(py_, "get_image_buffer_size"); + } + double GetExposure() const override { return py_get(py_, "get_exposure"); } + int GetBinning() const override { return py_get(py_, "get_binning"); } + + // -- MM::Camera: setters -- + void SetExposure(double ms) override { py_set(py_, "set_exposure", ms); } + int SetBinning(int bin) override { return py_call(py_, "set_binning", bin); } + + // -- MM::Camera: snap + buffer -- + int SnapImage() override { return py_call(py_, "snap_image"); } + + const unsigned char *GetImageBuffer() override { + return py_invoke([&]() -> const unsigned char * { + // Zero-copy: store the Python array to prevent GC and return its + // data pointer directly. The c_contig constraint will trigger an + // implicit copy only if the array is non-contiguous (e.g. a slice). + img_arr_ = py_.attr("get_image_buffer")(); + auto nd = nb::cast>(img_arr_); + return static_cast(nd.data()); + }); + } + + const unsigned char *GetImageBuffer(unsigned channelNr) override { + return py_invoke([&]() -> const unsigned char * { + img_arr_ = py_.attr("get_image_buffer")(channelNr); + auto nd = nb::cast>(img_arr_); + return static_cast(nd.data()); + }); + } + + // -- MM::Camera: ROI -- + int SetROI(unsigned x, unsigned y, unsigned w, unsigned h) override { + return py_call(py_, "set_roi", x, y, w, h); + } + int ClearROI() override { return py_call(py_, "clear_roi"); } + + int GetROI(unsigned &x, unsigned &y, unsigned &w, unsigned &h) override { + return py_invoke([&]() -> int { + auto roi = py_.attr("get_roi")(); + x = nb::cast(roi[nb::int_(0)]); + y = nb::cast(roi[nb::int_(1)]); + w = nb::cast(roi[nb::int_(2)]); + h = nb::cast(roi[nb::int_(3)]); + return DEVICE_OK; + }); + } + + // -- MM::Camera: exposure sequencing -- + std::vector exposureSeq_; + + int IsExposureSequenceable(bool &f) const override { + f = py_get(py_, "is_exposure_sequenceable"); + return DEVICE_OK; + } + int GetExposureSequenceMaxLength(long &nrEvents) const override { + nrEvents = py_get(py_, "get_exposure_sequence_max_length"); + return DEVICE_OK; + } + int ClearExposureSequence() override { + exposureSeq_.clear(); + return DEVICE_OK; + } + int AddToExposureSequence(double exposureTime_ms) override { + exposureSeq_.push_back(exposureTime_ms); + return DEVICE_OK; + } + int SendExposureSequence() const override { + return py_invoke([&]() -> int { + nb::list py_seq; + for (double v : exposureSeq_) + py_seq.append(v); + py_.attr("load_exposure_sequence")(py_seq); + return DEVICE_OK; + }); + } + int StartExposureSequence() override { return py_call(py_, "start_exposure_sequence"); } + int StopExposureSequence() override { return py_call(py_, "stop_exposure_sequence"); } + + // -- MM::Camera: sequence acquisition -- + bool IsCapturing() override { return py_get(py_, "is_capturing"); } + + int StartSequenceAcquisition(long numImages, double interval_ms, + bool stopOnOverflow) override { + int ret = GetCoreCallback()->PrepareForAcq(this); + if (ret != DEVICE_OK) + return ret; + + // Cache image dimensions once for the whole sequence. These shouldn't + // change during a running acquisition and querying them on every frame + // crosses the Python bridge 4 times per frame. + unsigned w = GetImageWidth(); + unsigned h = GetImageHeight(); + unsigned bpp = GetImageBytesPerPixel(); + unsigned nComp = GetNumberOfComponents(); + + return py_invoke([&]() -> int { + // Create an insert_image callable that pushes a frame into + // CMMCore's circular buffer. Python calls this per frame. + auto *self = this; + // insert_image returns False on buffer overflow so Python + // can stop acquisition when stopOnOverflow is set. + nb::object inserter = nb::cpp_function( + [self, w, h, bpp, nComp](nb::object arr, nb::object metadata) -> bool { + auto nd = nb::cast>(arr); + + // Build serialized metadata in MMCore's format + Metadata md; + if (!metadata.is_none()) { + nb::dict d = nb::cast(metadata); + for (auto [key, val] : d) { + auto k = nb::cast(nb::str(key)); + auto v = nb::cast(nb::str(val)); + md.PutImageTag(k.c_str(), v); + } + } + std::string mdStr = md.Serialize(); + + int ret; + { + nb::gil_scoped_release release; + ret = self->GetCoreCallback()->InsertImage( + self, static_cast(nd.data()), w, h, bpp, + nComp, mdStr.c_str()); + } + return ret == DEVICE_OK; + }, + nb::arg("image"), nb::arg("metadata") = nb::none()); + + py_.attr("start_sequence_acquisition")(numImages, interval_ms, inserter); + return DEVICE_OK; + }); + } + + int StartSequenceAcquisition(double interval_ms) override { + return StartSequenceAcquisition(LONG_MAX, interval_ms, false); + } + + int StopSequenceAcquisition() override { return py_call(py_, "stop_sequence_acquisition"); } +}; + +// ============================================================================ +// PyBridgeShutter +// ============================================================================ + +class PyBridgeShutter : public CShutterBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeShutter) + + // -- MM::Shutter -- + int SetOpen(bool open) override { return py_call(py_, "set_open", open); } + + int GetOpen(bool &open) override { + open = py_get(py_, "get_open"); + return DEVICE_OK; + } + + int Fire(double deltaT) override { return py_call(py_, "fire", deltaT); } +}; + +// ============================================================================ +// PyBridgeStage +// ============================================================================ + +class PyBridgeStage : public CStageBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeStage) + + // -- MM::Stage: position -- + int SetPositionUm(double pos) override { return py_call(py_, "set_position_um", pos); } + int GetPositionUm(double &pos) override { + pos = py_get(py_, "get_position_um"); + return DEVICE_OK; + } + int SetRelativePositionUm(double d) override { + return py_call(py_, "set_relative_position_um", d); + } + int SetPositionSteps(long steps) override { + return py_call(py_, "set_position_steps", steps); + } + int GetPositionSteps(long &steps) override { + steps = py_get(py_, "get_position_steps"); + return DEVICE_OK; + } + int SetAdapterOriginUm(double d) override { + return py_call(py_, "set_adapter_origin_um", d); + } + int SetOrigin() override { return py_call(py_, "set_origin"); } + int GetLimits(double &lo, double &hi) override { + return py_invoke([&]() -> int { + auto lim = py_.attr("get_limits")(); + lo = nb::cast(lim[nb::int_(0)]); + hi = nb::cast(lim[nb::int_(1)]); + return DEVICE_OK; + }); + } + + // -- MM::Stage: motion -- + int Move(double velocity) override { return py_call(py_, "move", velocity); } + int Stop() override { return py_call(py_, "stop"); } + int Home() override { return py_call(py_, "home"); } + + // -- MM::Stage: focus -- + int GetFocusDirection(MM::FocusDirection &direction) override { + direction = static_cast(py_get(py_, "get_focus_direction")); + return DEVICE_OK; + } + bool IsContinuousFocusDrive() const override { + return py_get(py_, "is_continuous_focus_drive"); + } + + // -- MM::Stage: sequencing -- + std::vector stageSeq_; + + int IsStageSequenceable(bool &f) const override { + f = py_get(py_, "is_stage_sequenceable"); + return DEVICE_OK; + } + int GetStageSequenceMaxLength(long &nrEvents) const override { + nrEvents = py_get(py_, "get_stage_sequence_max_length"); + return DEVICE_OK; + } + int ClearStageSequence() override { + stageSeq_.clear(); + return DEVICE_OK; + } + int AddToStageSequence(double position) override { + stageSeq_.push_back(position); + return DEVICE_OK; + } + int SendStageSequence() override { + return py_invoke([&]() -> int { + nb::list py_seq; + for (double v : stageSeq_) + py_seq.append(v); + py_.attr("load_stage_sequence")(py_seq); + return DEVICE_OK; + }); + } + int StartStageSequence() override { return py_call(py_, "start_stage_sequence"); } + int StopStageSequence() override { return py_call(py_, "stop_stage_sequence"); } +}; + +// ============================================================================ +// PyBridgeXYStage +// ============================================================================ + +class PyBridgeXYStage : public CXYStageBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeXYStage) + + // -- MM::XYStage: position (um) -- + int SetPositionUm(double x, double y) override { + return py_call(py_, "set_position_um", x, y); + } + int GetPositionUm(double &x, double &y) override { + return py_invoke([&]() -> int { + auto pos = py_.attr("get_position_um")(); + x = nb::cast(pos[nb::int_(0)]); + y = nb::cast(pos[nb::int_(1)]); + return DEVICE_OK; + }); + } + int SetRelativePositionUm(double dx, double dy) override { + return py_call(py_, "set_relative_position_um", dx, dy); + } + int SetAdapterOriginUm(double x, double y) override { + return py_call(py_, "set_adapter_origin_um", x, y); + } + + // -- MM::XYStage: position (steps) -- + int SetPositionSteps(long x, long y) override { + return py_call(py_, "set_position_steps", x, y); + } + int GetPositionSteps(long &x, long &y) override { + return py_invoke([&]() -> int { + auto pos = py_.attr("get_position_steps")(); + x = nb::cast(pos[nb::int_(0)]); + y = nb::cast(pos[nb::int_(1)]); + return DEVICE_OK; + }); + } + int SetRelativePositionSteps(long x, long y) override { + return py_call(py_, "set_relative_position_steps", x, y); + } + + // -- MM::XYStage: motion -- + int Home() override { return py_call(py_, "home"); } + int Stop() override { return py_call(py_, "stop"); } + int Move(double vx, double vy) override { return py_call(py_, "move", vx, vy); } + + // -- MM::XYStage: origin -- + int SetOrigin() override { return py_call(py_, "set_origin"); } + int SetXOrigin() override { return py_call(py_, "set_x_origin"); } + int SetYOrigin() override { return py_call(py_, "set_y_origin"); } + + // -- MM::XYStage: limits + step size -- + int GetLimitsUm(double &xMin, double &xMax, double &yMin, double &yMax) override { + return py_invoke([&]() -> int { + auto lim = py_.attr("get_limits_um")(); + xMin = nb::cast(lim[nb::int_(0)]); + xMax = nb::cast(lim[nb::int_(1)]); + yMin = nb::cast(lim[nb::int_(2)]); + yMax = nb::cast(lim[nb::int_(3)]); + return DEVICE_OK; + }); + } + int GetStepLimits(long &xMin, long &xMax, long &yMin, long &yMax) override { + return py_invoke([&]() -> int { + auto lim = py_.attr("get_step_limits")(); + xMin = nb::cast(lim[nb::int_(0)]); + xMax = nb::cast(lim[nb::int_(1)]); + yMin = nb::cast(lim[nb::int_(2)]); + yMax = nb::cast(lim[nb::int_(3)]); + return DEVICE_OK; + }); + } + double GetStepSizeXUm() override { return py_get(py_, "get_step_size_x_um"); } + double GetStepSizeYUm() override { return py_get(py_, "get_step_size_y_um"); } + + // -- MM::XYStage: sequencing -- + std::vector> xySeq_; + + int IsXYStageSequenceable(bool &f) const override { + f = py_get(py_, "is_xy_stage_sequenceable"); + return DEVICE_OK; + } + int GetXYStageSequenceMaxLength(long &nrEvents) const override { + nrEvents = py_get(py_, "get_xy_stage_sequence_max_length"); + return DEVICE_OK; + } + int ClearXYStageSequence() override { + xySeq_.clear(); + return DEVICE_OK; + } + int AddToXYStageSequence(double positionX, double positionY) override { + xySeq_.emplace_back(positionX, positionY); + return DEVICE_OK; + } + int SendXYStageSequence() override { + return py_invoke([&]() -> int { + nb::list py_seq; + for (auto &[x, y] : xySeq_) + py_seq.append(nb::make_tuple(x, y)); + py_.attr("load_xy_stage_sequence")(py_seq); + return DEVICE_OK; + }); + } + int StartXYStageSequence() override { return py_call(py_, "start_xy_stage_sequence"); } + int StopXYStageSequence() override { return py_call(py_, "stop_xy_stage_sequence"); } +}; + +// ============================================================================ +// PyBridgeState +// ============================================================================ + +class PyBridgeState : public CStateDeviceBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeState) + + // -- MM::State -- + // CStateDeviceBase provides defaults for SetPosition, GetPosition, + // GetPositionLabel, SetPositionLabel, GetLabelPosition, SetGateOpen, + // GetGateOpen — all driven by the "State" and "Label" properties. + // The only pure virtual remaining is GetNumberOfPositions. + unsigned long GetNumberOfPositions() const override { + return py_invoke( + [&]() { return nb::cast(py_.attr("get_number_of_positions")()); }); + } +}; + +// ============================================================================ +// PyBridgeAutoFocus +// ============================================================================ + +class PyBridgeAutoFocus : public CAutoFocusBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeAutoFocus) + + // -- MM::AutoFocus -- + int SetContinuousFocusing(bool state) override { + return py_call(py_, "set_continuous_focusing", state); + } + int GetContinuousFocusing(bool &state) override { + state = py_get(py_, "get_continuous_focusing"); + return DEVICE_OK; + } + bool IsContinuousFocusLocked() override { + return py_get(py_, "is_continuous_focus_locked"); + } + int FullFocus() override { return py_call(py_, "full_focus"); } + int IncrementalFocus() override { return py_call(py_, "incremental_focus"); } + int GetLastFocusScore(double &score) override { + score = py_get(py_, "get_last_focus_score"); + return DEVICE_OK; + } + int GetCurrentFocusScore(double &score) override { + score = py_get(py_, "get_current_focus_score"); + return DEVICE_OK; + } + int GetOffset(double &offset) override { + offset = py_get(py_, "get_offset"); + return DEVICE_OK; + } + int SetOffset(double offset) override { return py_call(py_, "set_offset", offset); } +}; + +// ============================================================================ +// PyBridgeSignalIO +// ============================================================================ + +class PyBridgeSignalIO : public CSignalIOBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeSignalIO) + + // -- MM::SignalIO: core -- + int SetGateOpen(bool open) override { return py_call(py_, "set_gate_open", open); } + int GetGateOpen(bool &open) override { + open = py_get(py_, "get_gate_open"); + return DEVICE_OK; + } + int SetSignal(double volts) override { return py_call(py_, "set_signal", volts); } + int GetSignal(double &volts) override { + volts = py_get(py_, "get_signal"); + return DEVICE_OK; + } + int GetLimits(double &minVolts, double &maxVolts) override { + return py_invoke([&]() -> int { + auto lim = py_.attr("get_limits")(); + minVolts = nb::cast(lim[nb::int_(0)]); + maxVolts = nb::cast(lim[nb::int_(1)]); + return DEVICE_OK; + }); + } + + // -- MM::SignalIO: DA sequencing -- + std::vector daSeq_; + + int IsDASequenceable(bool &f) const override { + f = py_get(py_, "is_da_sequenceable"); + return DEVICE_OK; + } + int GetDASequenceMaxLength(long &nrEvents) const override { + nrEvents = py_get(py_, "get_da_sequence_max_length"); + return DEVICE_OK; + } + int ClearDASequence() override { + daSeq_.clear(); + return DEVICE_OK; + } + int AddToDASequence(double voltage) override { + daSeq_.push_back(voltage); + return DEVICE_OK; + } + int SendDASequence() override { + return py_invoke([&]() -> int { + nb::list py_seq; + for (double v : daSeq_) + py_seq.append(v); + py_.attr("load_da_sequence")(py_seq); + return DEVICE_OK; + }); + } + int StartDASequence() override { return py_call(py_, "start_da_sequence"); } + int StopDASequence() override { return py_call(py_, "stop_da_sequence"); } +}; + +// ============================================================================ +// PyBridgeMagnifier +// ============================================================================ + +class PyBridgeMagnifier : public CMagnifierBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeMagnifier) + + double GetMagnification() override { return py_get(py_, "get_magnification"); } +}; + +// ============================================================================ +// PyBridgeSerial +// ============================================================================ + +class PyBridgeSerial : public CSerialBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeSerial) + + MM::PortType GetPortType() const override { + return static_cast(py_get(py_, "get_port_type")); + } + + int SetCommand(const char *command, const char *term) override { + return py_invoke([&]() -> int { + py_.attr("set_command")(std::string(command), + term ? std::string(term) : std::string()); + return DEVICE_OK; + }); + } + + int GetAnswer(char *txt, unsigned maxChars, const char *term) override { + return py_invoke([&]() -> int { + auto answer = nb::cast( + py_.attr("get_answer")(term ? std::string(term) : std::string())); + strncpy(txt, answer.c_str(), maxChars); + if (maxChars > 0) + txt[maxChars - 1] = '\0'; + return DEVICE_OK; + }); + } + + int Write(const unsigned char *buf, unsigned long bufLen) override { + return py_invoke([&]() -> int { + // Pass as Python bytes object + nb::object py_bytes = nb::steal( + PyBytes_FromStringAndSize(reinterpret_cast(buf), bufLen)); + py_.attr("write")(py_bytes); + return DEVICE_OK; + }); + } + + int Read(unsigned char *buf, unsigned long bufLen, unsigned long &charsRead) override { + return py_invoke([&]() -> int { + nb::object result = py_.attr("read")(bufLen); + Py_buffer view; + if (PyObject_GetBuffer(result.ptr(), &view, PyBUF_SIMPLE) != 0) + throw nb::python_error(); + charsRead = std::min(static_cast(view.len), bufLen); + std::memcpy(buf, view.buf, charsRead); + PyBuffer_Release(&view); + return DEVICE_OK; + }); + } + + int Purge() override { return py_call(py_, "purge"); } +}; + +// ============================================================================ +// PyBridgeGalvo +// ============================================================================ + +class PyBridgeGalvo : public CGalvoBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeGalvo) + + // -- MM::Galvo: position + illumination -- + int PointAndFire(double x, double y, double time_us) override { + return py_call(py_, "point_and_fire", x, y, time_us); + } + int SetSpotInterval(double pulseInterval_us) override { + return py_call(py_, "set_spot_interval", pulseInterval_us); + } + int SetPosition(double x, double y) override { return py_call(py_, "set_position", x, y); } + int GetPosition(double &x, double &y) override { + return py_invoke([&]() -> int { + auto pos = py_.attr("get_position")(); + x = nb::cast(pos[nb::int_(0)]); + y = nb::cast(pos[nb::int_(1)]); + return DEVICE_OK; + }); + } + int SetIlluminationState(bool on) override { + return py_call(py_, "set_illumination_state", on); + } + + // -- MM::Galvo: range -- + double GetXRange() override { return py_get(py_, "get_x_range"); } + double GetXMinimum() override { return py_get(py_, "get_x_minimum"); } + double GetYRange() override { return py_get(py_, "get_y_range"); } + double GetYMinimum() override { return py_get(py_, "get_y_minimum"); } + + // -- MM::Galvo: polygons -- + int AddPolygonVertex(int polygonIndex, double x, double y) override { + return py_call(py_, "add_polygon_vertex", polygonIndex, x, y); + } + int DeletePolygons() override { return py_call(py_, "delete_polygons"); } + int LoadPolygons() override { return py_call(py_, "load_polygons"); } + int SetPolygonRepetitions(int repetitions) override { + return py_call(py_, "set_polygon_repetitions", repetitions); + } + int RunPolygons() override { return py_call(py_, "run_polygons"); } + + // -- MM::Galvo: sequence -- + int RunSequence() override { return py_call(py_, "run_sequence"); } + int StopSequence() override { return py_call(py_, "stop_sequence"); } + + // -- MM::Galvo: channel -- + int GetChannel(char *channelName) override { + return py_invoke([&]() -> int { + auto name = nb::cast(py_.attr("get_channel")()); + CDeviceUtils::CopyLimitedString(channelName, name.c_str()); + return DEVICE_OK; + }); + } +}; + +// ============================================================================ +// PyBridgeGeneric +// ============================================================================ + +class PyBridgeGeneric : public CGenericBase, + private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeGeneric) +}; + +// Forward declaration — PyBridgeHub::DetectInstalledDevices needs this. +inline MM::Device *createBridgeDevice(nb::object py_dev, MM::DeviceType type, + const std::string &name, + const std::string &description = ""); + +// ============================================================================ +// PyBridgeHub +// ============================================================================ + +class PyBridgeHub : public HubBase, private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeHub) + + // Discover peripherals by calling Python's detect_installed_devices(), + // which returns a list of (name, py_device, device_type) tuples. + // Each is wrapped in a bridge device and registered with HubBase. + int DetectInstalledDevices() override { + ClearInstalledDevices(); + return py_invoke([&]() -> int { + nb::object peripherals = py_.attr("detect_installed_devices")(); + for (auto item : peripherals) { + auto tup = nb::cast(item); + auto name = nb::cast(tup[0]); + nb::object py_dev = tup[1]; + auto type = nb::cast(tup[2]); + // Extract description from the Python object's class docstring. + std::string desc; + nb::object py_type = + nb::borrow(reinterpret_cast(Py_TYPE(py_dev.ptr()))); + nb::object doc = py_type.attr("__doc__"); + if (!doc.is_none()) + desc = nb::cast(nb::str(doc)); + MM::Device *pDev = createBridgeDevice(py_dev, type, name, desc); + if (pDev) + AddInstalledDevice(pDev); + } + return DEVICE_OK; + }); + } +}; + +// ============================================================================ +// PyBridgeSLM +// ============================================================================ + +class PyBridgeSLM : public CSLMBase, private PyBridgeDeviceBase { + PYBRIDGE_COMMON_OVERRIDES(PyBridgeSLM) + + // -- MM::SLM -- + int SetImage(unsigned char *pixels) override { + ensureSLMDimsCached(); + size_t h = cachedH_, w = cachedW_; + size_t pixDepth = cachedNComp_ * cachedBpp_; + return py_invoke([&]() -> int { + nb::ndarray arr; + if (pixDepth == 1) + arr = nb::ndarray(pixels, {h, w}); + else + arr = nb::ndarray(pixels, {h, w, pixDepth}); + py_.attr("set_image")(arr); + return DEVICE_OK; + }); + } + int SetImage(unsigned int *pixels) override { + ensureSLMDimsCached(); + size_t h = cachedH_, w = cachedW_; + return py_invoke([&]() -> int { + auto arr = nb::ndarray(pixels, {h, w}); + py_.attr("set_image")(arr); + return DEVICE_OK; + }); + } + int DisplayImage() override { return py_call(py_, "display_image"); } + int SetPixelsTo(unsigned char intensity) override { + return py_call(py_, "set_pixels_to", intensity); + } + int SetPixelsTo(unsigned char r, unsigned char g, unsigned char b) override { + return py_call(py_, "set_pixels_to_rgb", r, g, b); + } + int SetExposure(double interval_ms) override { + return py_call(py_, "set_exposure", interval_ms); + } + double GetExposure() override { return py_get(py_, "get_exposure"); } + unsigned GetWidth() override { return py_get(py_, "get_width"); } + unsigned GetHeight() override { return py_get(py_, "get_height"); } + unsigned GetNumberOfComponents() override { + return py_get(py_, "get_number_of_components"); + } + unsigned GetBytesPerPixel() override { + return py_get(py_, "get_bytes_per_pixel"); + } + // Cached SLM dimensions — populated once to avoid repeated Python + // bridge crossings in SetImage / AddToSLMSequence / SendSLMSequence. + unsigned cachedW_ = 0, cachedH_ = 0, cachedNComp_ = 0, cachedBpp_ = 0; + bool slmDimsCached_ = false; + + void ensureSLMDimsCached() { + if (!slmDimsCached_) { + cachedW_ = GetWidth(); + cachedH_ = GetHeight(); + cachedNComp_ = GetNumberOfComponents(); + cachedBpp_ = GetBytesPerPixel(); + slmDimsCached_ = true; + } + } + + // -- MM::SLM: sequencing -- + std::vector> slmSeq8_; + std::vector> slmSeq32_; + bool usingSeq32_ = false; + + int IsSLMSequenceable(bool &f) const override { + f = py_get(py_, "is_slm_sequenceable"); + return DEVICE_OK; + } + int GetSLMSequenceMaxLength(long &nrEvents) const override { + nrEvents = py_get(py_, "get_slm_sequence_max_length"); + return DEVICE_OK; + } + int ClearSLMSequence() override { + slmSeq8_.clear(); + slmSeq32_.clear(); + usingSeq32_ = false; + return DEVICE_OK; + } + int AddToSLMSequence(const unsigned char *const image) override { + ensureSLMDimsCached(); + size_t nbytes = (size_t)cachedH_ * cachedW_ * cachedNComp_ * cachedBpp_; + slmSeq8_.emplace_back(image, image + nbytes); + usingSeq32_ = false; + return DEVICE_OK; + } + int AddToSLMSequence(const unsigned int *const image) override { + ensureSLMDimsCached(); + size_t npixels = (size_t)cachedH_ * cachedW_; + slmSeq32_.emplace_back(image, image + npixels); + usingSeq32_ = true; + return DEVICE_OK; + } + int SendSLMSequence() override { + ensureSLMDimsCached(); + size_t h = cachedH_, w = cachedW_; + size_t pixDepth = (size_t)cachedNComp_ * cachedBpp_; + return py_invoke([&]() -> int { + nb::list py_seq; + if (usingSeq32_) { + for (auto &buf : slmSeq32_) { + auto arr = + nb::ndarray(buf.data(), {h, w}); + py_seq.append(arr); + } + } else { + for (auto &buf : slmSeq8_) { + nb::ndarray arr; + if (pixDepth == 1) + arr = nb::ndarray(buf.data(), {h, w}); + else + arr = nb::ndarray(buf.data(), + {h, w, pixDepth}); + py_seq.append(arr); + } + } + py_.attr("load_slm_sequence")(py_seq); + return DEVICE_OK; + }); + } + int StartSLMSequence() override { return py_call(py_, "start_slm_sequence"); } + int StopSLMSequence() override { return py_call(py_, "stop_slm_sequence"); } +}; + +#undef PYBRIDGE_COMMON_OVERRIDES + +// ============================================================================ +// Helper: create the right bridge device for a given MM::DeviceType +// ============================================================================ + +inline MM::Device *createBridgeDevice(nb::object py_dev, MM::DeviceType type, + const std::string &name, const std::string &description) { + switch (type) { + case MM::CameraDevice: return new PyBridgeCamera(py_dev, name, description); + case MM::ShutterDevice: return new PyBridgeShutter(py_dev, name, description); + case MM::StageDevice: return new PyBridgeStage(py_dev, name, description); + case MM::XYStageDevice: return new PyBridgeXYStage(py_dev, name, description); + case MM::StateDevice: return new PyBridgeState(py_dev, name, description); + case MM::SLMDevice: return new PyBridgeSLM(py_dev, name, description); + case MM::AutoFocusDevice: return new PyBridgeAutoFocus(py_dev, name, description); + case MM::SignalIODevice: return new PyBridgeSignalIO(py_dev, name, description); + case MM::GalvoDevice: return new PyBridgeGalvo(py_dev, name, description); + case MM::MagnifierDevice: return new PyBridgeMagnifier(py_dev, name, description); + case MM::SerialDevice: return new PyBridgeSerial(py_dev, name, description); + case MM::GenericDevice: return new PyBridgeGeneric(py_dev, name, description); + case MM::HubDevice: return new PyBridgeHub(py_dev, name, description); + default: + throw std::runtime_error("No Python bridge for device type " + std::to_string(type)); + } +} + +// ============================================================================ +// PyBridgeAdapter — implements MockDeviceAdapter for Python bridge devices. +// +// Supports two modes: +// 1. Pre-instantiated: addDevice(name, py_instance, type) +// CreateDevice returns a bridge wrapping the existing instance. +// 2. Class-based: addDeviceClass(name, py_class, type, description) +// CreateDevice instantiates the Python class on demand. +// +// Both modes can be mixed in the same adapter, and all devices from +// one adapter share the same LoadedDeviceAdapter mutex. +// ============================================================================ + +class PyBridgeAdapter : public MockDeviceAdapter { + struct DeviceEntry { + std::string name; + std::string description; + nb::object py_obj; // either an instance or a class + MM::DeviceType type; + bool is_class; // true = call py_obj() to instantiate + }; + + std::vector devices_; + bool loaded_ = false; + + public: + PyBridgeAdapter() = default; + + // Move constructor: leaves the source marked as loaded so it + // rejects further modifications. + PyBridgeAdapter(PyBridgeAdapter &&other) noexcept + : devices_(std::move(other.devices_)), loaded_(other.loaded_) { + other.loaded_ = true; + } + PyBridgeAdapter &operator=(PyBridgeAdapter &&) = delete; + PyBridgeAdapter(const PyBridgeAdapter &) = delete; + PyBridgeAdapter &operator=(const PyBridgeAdapter &) = delete; + + ~PyBridgeAdapter() { + try { + nb::gil_scoped_acquire gil; + devices_.clear(); + } catch (...) { + } + } + + // Register a pre-instantiated Python device (for loadPyDevice). + void addDevice(const std::string &name, nb::object py_dev, MM::DeviceType type) { + if (loaded_) + throw std::runtime_error("Cannot add devices after adapter has been loaded"); + devices_.push_back({name, "Python bridge device", std::move(py_dev), type, false}); + } + + // Register a Python device class (for loadPyDeviceAdapter). + // CreateDevice will call py_cls() to instantiate. + void addDeviceClass(const std::string &name, nb::object py_cls, MM::DeviceType type, + const std::string &description) { + if (loaded_) + throw std::runtime_error("Cannot add devices after adapter has been loaded"); + devices_.push_back({name, description, std::move(py_cls), type, true}); + } + + // Mark as loaded — called by registerAndStoreBridgeAdapter after + // the adapter has been registered with CMMCore. + void markLoaded() { loaded_ = true; } + + void InitializeModuleData(RegisterDeviceFunc registerDevice) override { + for (auto &d : devices_) { + registerDevice(d.name.c_str(), d.type, d.description.c_str()); + } + } + + MM::Device *CreateDevice(const char *name) override { + nb::gil_scoped_acquire gil; + for (auto &d : devices_) { + if (d.name == name) { + try { + nb::object py_dev = d.is_class ? d.py_obj() : d.py_obj; + return createBridgeDevice(py_dev, d.type, d.name, d.description); + } catch (nb::python_error &e) { + e.restore(); + PyErr_Clear(); + return nullptr; + } catch (...) { + return nullptr; + } + } + } + return nullptr; + } + + void DeleteDevice(MM::Device *device) override { delete device; } +}; diff --git a/src/pymmcore_nano/__init__.py b/src/pymmcore_nano/__init__.py index 22e7e92b..c00f6010 100644 --- a/src/pymmcore_nano/__init__.py +++ b/src/pymmcore_nano/__init__.py @@ -1,9 +1,8 @@ -from ._pymmcore_nano import * # noqa -from ._pymmcore_nano import __version__ as __version__ # type: ignore [attr-defined] - -import sys import importlib.util +import sys +from ._pymmcore_nano import * # pyright: ignore # noqa: F403 +from ._pymmcore_nano import __version__ as __version__ # type: ignore [attr-defined] _pymmcore_spec = importlib.util.find_spec("pymmcore") diff --git a/src/pymmcore_nano/protocols.py b/src/pymmcore_nano/protocols.py new file mode 100644 index 00000000..d6fe9d11 --- /dev/null +++ b/src/pymmcore_nano/protocols.py @@ -0,0 +1,304 @@ +"""Protocols defining what the C++ bridge expects from Python device objects. + +THESE ARE MANUALLY DEFINED. + +These *should* mirror the methods that bridge_devices.h calls on Python device +objects via nanobind. Update these when bridge device classes change. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + + import numpy as np + + from . import DeviceCallbacks, PropertyHandle + + +class CreatePropertyFn(Protocol): + """Callable passed to initialize() for registering MM properties.""" + + def __call__( + self, + name: str, + default_value: str, + mm_type: int, + read_only: bool, + *, + getter: Callable[[], str | int | float] | None = None, + setter: Callable[[str], None] | None = None, + pre_init: bool = False, + limits: tuple[float, float] | None = None, + allowed_values: Sequence[str] | None = None, + sequence_max_length: int = 0, + sequence_loader: Callable[[list[str]], None] | None = None, + sequence_starter: Callable[[], None] | None = None, + sequence_stopper: Callable[[], None] | None = None, + ) -> PropertyHandle: ... + + +@runtime_checkable +class PyDevice(Protocol): + """Base protocol for all Python bridge devices.""" + + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + """Initialize the device. + + Receives two callables: + - `create_property(...)` for registering properties with CMMCore. See + `CreatePropertyFn` for details. + - `notify(...)` for sending property change notifications to CMMCore. See + `DeviceCallbacks` for details. + + Because this signature is different from the standard `initialize()` method of + MM devices, the bridge calls this `initialize_bridge()` instead. This leaves + downstream devices free to implement `initialize()` as desired (which + will likely be called inside of `initialize_bridge()`). + """ + ... + + def shutdown(self) -> None: ... + def busy(self) -> bool: ... + + +@runtime_checkable +class PyCamera(PyDevice, Protocol): + """Protocol for Python camera devices.""" + + def get_image_width(self) -> int: ... + def get_image_height(self) -> int: ... + def get_bytes_per_pixel(self) -> int: ... + def get_number_of_components(self) -> int: ... + def get_number_of_channels(self) -> int: ... + def get_channel_name(self, channel: int) -> str: ... + def get_bit_depth(self) -> int: ... + def get_image_buffer_size(self) -> int: ... + def get_exposure(self) -> float: ... + def get_binning(self) -> int: ... + def set_binning(self, bin: int) -> None: ... + def set_roi(self, x: int, y: int, w: int, h: int) -> None: ... + def clear_roi(self) -> None: ... + def set_exposure(self, ms: float) -> None: ... + def snap_image(self) -> None: ... + def get_image_buffer(self, channel: int = 0) -> np.ndarray: ... + def get_roi(self) -> tuple[int, int, int, int]: ... + def is_exposure_sequenceable(self) -> bool: ... + def get_exposure_sequence_max_length(self) -> int: ... + def load_exposure_sequence(self, sequence: list[float]) -> None: ... + def start_exposure_sequence(self) -> None: ... + def stop_exposure_sequence(self) -> None: ... + def is_capturing(self) -> bool: ... + def start_sequence_acquisition( + self, + num_images: int, + interval_ms: float, + insert_image: Callable[[np.ndarray, dict | None], bool], + ) -> None: + """Start sequence acquisition. + + The device owns the acquisition loop — call `insert_image(array, + metadata)` for each frame, synchronously or from a background thread. + `insert_image` returns True on success, False when CMMCore's circular + buffer is full. Stop acquiring when it returns False. + """ + ... + + def stop_sequence_acquisition(self) -> None: ... + + +@runtime_checkable +class PyShutter(PyDevice, Protocol): + """Protocol for Python shutter devices.""" + + def get_open(self) -> bool: ... + def set_open(self, open: bool) -> None: ... + def fire(self, delta_t: float) -> None: ... + + +@runtime_checkable +class PyStage(PyDevice, Protocol): + """Protocol for Python single-axis stage devices.""" + + # position + def set_position_um(self, pos: float) -> None: ... + def get_position_um(self) -> float: ... + def set_relative_position_um(self, d: float) -> None: ... + def set_position_steps(self, steps: int) -> None: ... + def get_position_steps(self) -> int: ... + def set_adapter_origin_um(self, d: float) -> None: ... + def set_origin(self) -> None: ... + def get_limits(self) -> tuple[float, float]: ... + # motion + def move(self, velocity: float) -> None: ... + def stop(self) -> None: ... + def home(self) -> None: ... + # focus + def get_focus_direction(self) -> int: ... + def is_continuous_focus_drive(self) -> bool: ... + # sequencing + def is_stage_sequenceable(self) -> bool: ... + def get_stage_sequence_max_length(self) -> int: ... + def load_stage_sequence(self, positions: list[float]) -> None: ... + def start_stage_sequence(self) -> None: ... + def stop_stage_sequence(self) -> None: ... + + +@runtime_checkable +class PyXYStage(PyDevice, Protocol): + """Protocol for Python XY stage devices.""" + + # position (um) + def set_position_um(self, x: float, y: float) -> None: ... + def get_position_um(self) -> tuple[float, float]: ... + def set_relative_position_um(self, dx: float, dy: float) -> None: ... + def set_adapter_origin_um(self, x: float, y: float) -> None: ... + # position (steps) + def set_position_steps(self, x: int, y: int) -> None: ... + def get_position_steps(self) -> tuple[int, int]: ... + def set_relative_position_steps(self, x: int, y: int) -> None: ... + # motion + def home(self) -> None: ... + def stop(self) -> None: ... + def move(self, vx: float, vy: float) -> None: ... + # origin + def set_origin(self) -> None: ... + def set_x_origin(self) -> None: ... + def set_y_origin(self) -> None: ... + # limits + step size + def get_limits_um(self) -> tuple[float, float, float, float]: ... + def get_step_limits(self) -> tuple[int, int, int, int]: ... + def get_step_size_x_um(self) -> float: ... + def get_step_size_y_um(self) -> float: ... + # sequencing + def is_xy_stage_sequenceable(self) -> bool: ... + def get_xy_stage_sequence_max_length(self) -> int: ... + def load_xy_stage_sequence(self, positions: list[tuple[float, float]]) -> None: ... + def start_xy_stage_sequence(self) -> None: ... + def stop_xy_stage_sequence(self) -> None: ... + + +@runtime_checkable +class PySignalIO(PyDevice, Protocol): + """Protocol for Python SignalIO (DA/AD) devices.""" + + def set_gate_open(self, open: bool) -> None: ... + def get_gate_open(self) -> bool: ... + def set_signal(self, volts: float) -> None: ... + def get_signal(self) -> float: ... + def get_limits(self) -> tuple[float, float]: ... + # sequencing + def is_da_sequenceable(self) -> bool: ... + def get_da_sequence_max_length(self) -> int: ... + def load_da_sequence(self, sequence: list[float]) -> None: ... + def start_da_sequence(self) -> None: ... + def stop_da_sequence(self) -> None: ... + + +@runtime_checkable +class PyMagnifier(PyDevice, Protocol): + """Protocol for Python magnifier devices.""" + + def get_magnification(self) -> float: ... + + +@runtime_checkable +class PySerial(PyDevice, Protocol): + """Protocol for Python serial port devices.""" + + def get_port_type(self) -> int: ... + def set_command(self, command: str, term: str) -> None: ... + def get_answer(self, term: str) -> str: ... + def write(self, data: bytes) -> None: ... + def read(self, max_bytes: int) -> bytes: ... + def purge(self) -> None: ... + + +@runtime_checkable +class PyGalvo(PyDevice, Protocol): + """Protocol for Python galvo scanner devices.""" + + # position + illumination + def point_and_fire(self, x: float, y: float, time_us: float) -> None: ... + def set_spot_interval(self, pulse_interval_us: float) -> None: ... + def set_position(self, x: float, y: float) -> None: ... + def get_position(self) -> tuple[float, float]: ... + def set_illumination_state(self, on: bool) -> None: ... + # range + def get_x_range(self) -> float: ... + def get_x_minimum(self) -> float: ... + def get_y_range(self) -> float: ... + def get_y_minimum(self) -> float: ... + # polygons + def add_polygon_vertex(self, polygon_index: int, x: float, y: float) -> None: ... + def delete_polygons(self) -> None: ... + def load_polygons(self) -> None: ... + def set_polygon_repetitions(self, repetitions: int) -> None: ... + def run_polygons(self) -> None: ... + # sequence + def run_sequence(self) -> None: ... + def stop_sequence(self) -> None: ... + # channel + def get_channel(self) -> str: ... + + +@runtime_checkable +class PyState(PyDevice, Protocol): + """Protocol for Python state devices (filter wheel, turret, etc.).""" + + def get_number_of_positions(self) -> int: ... + + +@runtime_checkable +class PyAutoFocus(PyDevice, Protocol): + """Protocol for Python auto-focus devices.""" + + def set_continuous_focusing(self, state: bool) -> None: ... + def get_continuous_focusing(self) -> bool: ... + def is_continuous_focus_locked(self) -> bool: ... + def full_focus(self) -> None: ... + def incremental_focus(self) -> None: ... + def get_last_focus_score(self) -> float: ... + def get_current_focus_score(self) -> float: ... + def get_offset(self) -> float: ... + def set_offset(self, offset: float) -> None: ... + + +@runtime_checkable +class PyGeneric(PyDevice, Protocol): + """Protocol for Python generic devices (properties only).""" + + +@runtime_checkable +class PyHub(PyDevice, Protocol): + """Protocol for Python hub devices.""" + + def detect_installed_devices(self) -> Sequence[tuple[str, object, int]]: + """Return peripherals as (name, py_device, device_type) tuples.""" + ... + + +@runtime_checkable +class PySLM(PyDevice, Protocol): + """Protocol for Python SLM devices.""" + + def set_image(self, pixels: np.ndarray) -> None: ... + def display_image(self) -> None: ... + def set_pixels_to(self, intensity: int) -> None: ... + def set_pixels_to_rgb(self, r: int, g: int, b: int) -> None: ... + def set_exposure(self, interval_ms: float) -> None: ... + def get_exposure(self) -> float: ... + def get_width(self) -> int: ... + def get_height(self) -> int: ... + def get_number_of_components(self) -> int: ... + def get_bytes_per_pixel(self) -> int: ... + # sequencing + def is_slm_sequenceable(self) -> bool: ... + def get_slm_sequence_max_length(self) -> int: ... + def load_slm_sequence(self, images: list[np.ndarray]) -> None: ... + def start_slm_sequence(self) -> None: ... + def stop_slm_sequence(self) -> None: ... diff --git a/subprojects/mmcore.wrap b/subprojects/mmcore.wrap index fe0b666e..91440167 100644 --- a/subprojects/mmcore.wrap +++ b/subprojects/mmcore.wrap @@ -2,6 +2,7 @@ url = https://github.com/micro-manager/mmcore.git revision = 8399a00b7210fd0775f3fe51392828a69c9aa57b depth = 1 +diff_files = mmcore/0001-unload-adapter-from-registry.patch, mmcore/0002-mock-adapter-ownership-transfer.patch, mmcore/0003-validate-sequence-lengths-in-load-methods.patch [provide] dependency_names = mmcore diff --git a/subprojects/packagefiles/mmcore/0001-unload-adapter-from-registry.patch b/subprojects/packagefiles/mmcore/0001-unload-adapter-from-registry.patch new file mode 100644 index 00000000..93c09f83 --- /dev/null +++ b/subprojects/packagefiles/mmcore/0001-unload-adapter-from-registry.patch @@ -0,0 +1,13 @@ +diff --git a/PluginManager.cpp b/PluginManager.cpp +index c12eda0..ed0bf79 100644 +--- a/PluginManager.cpp ++++ b/PluginManager.cpp +@@ -207,6 +207,8 @@ CPluginManager::UnloadPluginLibrary(const char* moduleName) + { + throw CMMError("Cannot unload device adapter " + ToQuotedString(moduleName), e); + } ++ ++ moduleMap_.erase(it); + } + + diff --git a/subprojects/packagefiles/mmcore/0002-mock-adapter-ownership-transfer.patch b/subprojects/packagefiles/mmcore/0002-mock-adapter-ownership-transfer.patch new file mode 100644 index 00000000..8a7abccc --- /dev/null +++ b/subprojects/packagefiles/mmcore/0002-mock-adapter-ownership-transfer.patch @@ -0,0 +1,63 @@ +diff --git a/LoadableModules/LoadedDeviceAdapterImplMock.h b/LoadableModules/LoadedDeviceAdapterImplMock.h +index 1aa3af7..2439fdb 100644 +--- a/LoadableModules/LoadedDeviceAdapterImplMock.h ++++ b/LoadableModules/LoadedDeviceAdapterImplMock.h +@@ -25,6 +25,8 @@ + + #include "RegisteredDeviceCollection.h" + ++#include ++ + + namespace mmcore { + namespace internal { +@@ -32,10 +34,10 @@ namespace internal { + class LoadedDeviceAdapterImplMock : public LoadedDeviceAdapterImpl + { + public: +- explicit LoadedDeviceAdapterImplMock(MockDeviceAdapter* impl) +- : impl_(impl) {} ++ explicit LoadedDeviceAdapterImplMock(std::unique_ptr impl) ++ : impl_(std::move(impl)) {} + +- void Unload() override {}; ++ void Unload() override { impl_.reset(); } + + void InitializeModuleData() override; + long GetModuleVersion() const override; +@@ -49,7 +51,7 @@ public: + void DeleteDevice(MM::Device* device) override; + + private: +- MockDeviceAdapter* impl_; ++ std::unique_ptr impl_; + MM::internal::RegisteredDeviceCollection registeredDevices_; + }; + +diff --git a/MockDeviceAdapter.h b/MockDeviceAdapter.h +index f5036d8..e1ea8ec 100644 +--- a/MockDeviceAdapter.h ++++ b/MockDeviceAdapter.h +@@ -27,6 +27,8 @@ + struct MockDeviceAdapter { + using RegisterDeviceFunc = std::function; + ++ virtual ~MockDeviceAdapter() = default; ++ + virtual void InitializeModuleData(RegisterDeviceFunc registerDevice) = 0; + virtual MM::Device* CreateDevice(const char* name) = 0; + virtual void DeleteDevice(MM::Device* device) = 0; +diff --git a/PluginManager.cpp b/PluginManager.cpp +index ed0bf79..6021c5f 100644 +--- a/PluginManager.cpp ++++ b/PluginManager.cpp +@@ -184,7 +184,8 @@ CPluginManager::LoadMockAdapter(const std::string& name, MockDeviceAdapter* impl + } + + moduleMap_[name] = std::make_shared( +- name, std::make_unique(impl)); ++ name, std::make_unique( ++ std::unique_ptr(impl))); + } + + diff --git a/subprojects/packagefiles/mmcore/0003-validate-sequence-lengths-in-load-methods.patch b/subprojects/packagefiles/mmcore/0003-validate-sequence-lengths-in-load-methods.patch new file mode 100644 index 00000000..f06fc8ea --- /dev/null +++ b/subprojects/packagefiles/mmcore/0003-validate-sequence-lengths-in-load-methods.patch @@ -0,0 +1,52 @@ +diff --git a/MMCore.cpp b/MMCore.cpp +--- a/MMCore.cpp ++++ b/MMCore.cpp +@@ -2375,6 +2375,13 @@ void CMMCore::loadStageSequence(const char* label, std::vector positionS + std::shared_ptr pStage = + deviceManager_->GetDeviceOfType(label); + ++ unsigned long maxLength = getStageSequenceMaxLength(label); ++ if (positionSequence.size() > maxLength) { ++ throw CMMError("The length of the requested stage sequence (" + ToString(positionSequence.size()) + ++ ") exceeds the maximum allowed (" + ToString(maxLength) + ++ ") by the stage " + ToQuotedString(label)); ++ } ++ + mmi::DeviceModuleLockGuard guard(pStage); + + int ret; +@@ -2528,9 +2535,21 @@ void CMMCore::loadXYStageSequence(const char* label, + std::vector xSequence, + std::vector ySequence) MMCORE_LEGACY_THROW(CMMError) + { ++ if (xSequence.size() != ySequence.size()) { ++ throw CMMError("xSequence and ySequence must have the same length (got " + ++ ToString(xSequence.size()) + " and " + ToString(ySequence.size()) + ")"); ++ } ++ + std::shared_ptr pStage = + deviceManager_->GetDeviceOfType(label); + ++ unsigned long maxLength = getXYStageSequenceMaxLength(label); ++ if (xSequence.size() > maxLength) { ++ throw CMMError("The length of the requested XY stage sequence (" + ToString(xSequence.size()) + ++ ") exceeds the maximum allowed (" + ToString(maxLength) + ++ ") by the XY stage " + ToQuotedString(label)); ++ } ++ + mmi::DeviceModuleLockGuard guard(pStage); + + int ret; +@@ -6488,6 +6507,12 @@ void CMMCore::loadSLMSequence(const char* deviceLabel, std::vector pSLM = + deviceManager_->GetDeviceOfType(deviceLabel); + ++ unsigned long maxLength = getSLMSequenceMaxLength(deviceLabel); ++ if (imageSequence.size() > maxLength) { ++ throw CMMError("The length of the requested SLM sequence (" + ToString(imageSequence.size()) + ++ ") exceeds the maximum allowed (" + ToString(maxLength) + ++ ") by the SLM " + ToQuotedString(deviceLabel)); ++ } + + mmi::DeviceModuleLockGuard guard(pSLM); + int ret = pSLM->ClearSLMSequence(); diff --git a/tests/test_bridge_devices.py b/tests/test_bridge_devices.py new file mode 100644 index 00000000..83bd00ef --- /dev/null +++ b/tests/test_bridge_devices.py @@ -0,0 +1,1451 @@ +"""Tests for Python bridge devices via MockDeviceAdapter.""" + +from __future__ import annotations + +import threading +import time +from typing import TYPE_CHECKING + +import numpy as np +import pymmcore_nano as pmn +from pymmcore_nano import CMMCore, DeviceAdapter, DeviceType + +if TYPE_CHECKING: + from collections.abc import Callable + + from pymmcore_nano import DeviceCallbacks + from pymmcore_nano.protocols import CreatePropertyFn + + +class MinimalDevice: + """Shared base for all minimal test devices.""" + + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + pass + + def shutdown(self) -> None: + pass + + def busy(self) -> bool: + return False + + +class MinimalCamera(MinimalDevice): + """Minimal Python camera that satisfies the bridge interface.""" + + def __init__(self, width: int = 64, height: int = 32) -> None: + self._width = width + self._height = height + self._exposure = 10.0 + self._buf: np.ndarray | None = None + self._gain: float = 1.0 + self._mode: str = "Normal" + self._capturing = False + self._notify = None + + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + super().initialize_bridge(create_property, notify) + self._notify = notify + self._gain_prop = create_property( + "Gain", + "1.0", + 2, + False, + getter=lambda: self._gain, + setter=lambda v: setattr(self, "_gain", float(v)), + limits=(0.0, 100.0), + ) + create_property( + "Mode", + "Normal", + 1, + False, + getter=lambda: self._mode, + setter=lambda v: setattr(self, "_mode", str(v)), + allowed_values=["Normal", "Fast", "Slow"], + ) + + def snap_image(self) -> None: + # Fill with a recognizable pattern + self._buf = np.arange(self._width * self._height, dtype=np.uint8).reshape( + self._height, self._width + ) + + def get_image_buffer(self, channel: int = 0) -> np.ndarray: + assert self._buf is not None + return self._buf + + def is_exposure_sequenceable(self) -> bool: + return False + + def get_image_width(self) -> int: + return self._width + + def get_image_height(self) -> int: + return self._height + + def get_bytes_per_pixel(self) -> int: + return 1 + + def get_number_of_components(self) -> int: + return 1 + + def get_number_of_channels(self) -> int: + return 1 + + def get_channel_name(self, channel: int) -> str: + return "" + + def get_bit_depth(self) -> int: + return 8 + + def get_image_buffer_size(self) -> int: + return self._width * self._height + + def get_exposure(self) -> float: + return self._exposure + + def set_exposure(self, ms: float) -> None: + self._exposure = ms + + def get_binning(self) -> int: + return 1 + + def set_binning(self, b: int) -> None: + pass + + def set_roi(self, x: int, y: int, w: int, h: int) -> None: + pass + + def get_roi(self) -> tuple[int, int, int, int]: + return (0, 0, self._width, self._height) + + def clear_roi(self) -> None: + pass + + def is_capturing(self) -> bool: + return self._capturing + + def start_sequence_acquisition( + self, + n: int, + interval_ms: float, + insert_image: Callable[[np.ndarray, dict | None], bool], + ) -> None: + self._stop_event = threading.Event() + self._capturing = True + + def run(): + count = 0 + try: + while not self._stop_event.is_set(): + if n is not None and n < 2**62 and count >= n: + break + img = np.full( + (self._height, self._width), count % 256, dtype=np.uint8 + ) + if not insert_image(img, {"frame": count}): + break + count += 1 + time.sleep(interval_ms / 1000.0) + finally: + self._capturing = False + if self._notify is not None: + self._notify.acq_finished() + + self._acq_thread = threading.Thread(target=run, daemon=True) + self._acq_thread.start() + + def stop_sequence_acquisition(self) -> None: + if hasattr(self, "_stop_event"): + self._stop_event.set() + if hasattr(self, "_acq_thread"): + self._acq_thread.join(timeout=5.0) + + +class MinimalShutter(MinimalDevice): + """Minimal Python shutter for the bridge.""" + + def __init__(self) -> None: + self._open = False + + def set_open(self, state: bool) -> None: + self._open = state + + def get_open(self) -> bool: + return self._open + + def fire(self, delta_t: float) -> None: + pass + + +def test_load_py_camera() -> None: + core = CMMCore() + cam = MinimalCamera(width=64, height=32) + core.loadPyDevice("MyCam", cam, DeviceType.CameraDevice) + core.initializeDevice("MyCam") + + # The device should appear in loaded devices + assert "MyCam" in core.getLoadedDevices() + + # Set as the current camera + core.setCameraDevice("MyCam") + assert core.getCameraDevice() == "MyCam" + + # Snap and retrieve image + core.snapImage() + img = core.getImage() + + assert img.shape == (32, 64) + assert img.dtype == np.uint8 + + # Verify the pixel pattern round-trips + expected = np.arange(64 * 32, dtype=np.uint8).reshape(32, 64) + np.testing.assert_array_equal(img, expected) + + +def test_sequence_acquisition() -> None: + core = CMMCore() + cam = MinimalCamera(width=64, height=32) + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + + # Set up circular buffer + core.setCircularBufferMemoryFootprint(16) # 16 MB + core.initializeCircularBuffer() + + # Start finite acquisition (5 frames) + core.startSequenceAcquisition(5, 0.0, True) + + # Wait for frames to arrive + deadline = time.time() + 5.0 + while core.getRemainingImageCount() < 5 and time.time() < deadline: + time.sleep(0.01) + + core.stopSequenceAcquisition() + + assert core.getRemainingImageCount() >= 5 + + # Pop a frame and verify shape + metadata + img, md = core.popNextImageMD() + assert img.shape == (32, 64) + assert img.dtype == np.uint8 + + # CMMCore auto-adds standard metadata + assert md.HasTag("Width") + assert md.GetSingleTag("Width").GetValue() == "64" + assert md.HasTag("Height") + assert md.GetSingleTag("Height").GetValue() == "32" + assert md.HasTag("Camera") + assert md.GetSingleTag("Camera").GetValue() == "Cam" + + # Our Python device's custom metadata should be present too + assert md.HasTag("frame") + assert md.GetSingleTag("frame").GetValue() == "0" + + +def test_load_py_shutter() -> None: + core = CMMCore() + shutter = MinimalShutter() + core.loadPyDevice("MyShutter", shutter, DeviceType.ShutterDevice) + core.initializeDevice("MyShutter") + + assert "MyShutter" in core.getLoadedDevices() + + core.setShutterDevice("MyShutter") + assert core.getShutterDevice() == "MyShutter" + + core.setShutterOpen(True) + assert shutter._open is True + + core.setShutterOpen(False) + assert shutter._open is False + + +def test_camera_exposure() -> None: + core = CMMCore() + cam = MinimalCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + + core.setExposure(42.0) + assert cam._exposure == 42.0 + assert core.getExposure() == 42.0 + + +def test_unload_py_device() -> None: + core = CMMCore() + cam = MinimalCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + assert "Cam" in core.getLoadedDevices() + + core.unloadDevice("Cam") + assert "Cam" not in core.getLoadedDevices() + + +def test_reload_py_device_after_unload() -> None: + """Reloading the same label after unload should work.""" + core = CMMCore() + + cam1 = MinimalCamera(width=64, height=32) + core.loadPyDevice("Cam", cam1, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + core.snapImage() + assert core.getImage().shape == (32, 64) + + core.unloadDevice("Cam") + + # Reload with a new device using the same label + cam2 = MinimalCamera(width=16, height=8) + core.loadPyDevice("Cam", cam2, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + core.snapImage() + assert core.getImage().shape == (8, 16) + + +def test_adapter_cleanup_on_core_destroy() -> None: + """CMMCore destruction should release bridge adapter references.""" + import gc + import weakref + + cam = MinimalCamera() + cam_ref = weakref.ref(cam) + + core = CMMCore() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + + del cam + # Adapter still holds a reference + assert cam_ref() is not None + + # Destroying the core should clean up everything + del core + gc.collect() + assert cam_ref() is None, "CMMCore destruction leaked bridge adapter" + + +def test_device_properties() -> None: + core = CMMCore() + cam = MinimalCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + + # Properties should be visible through CMMCore + names = core.getDevicePropertyNames("Cam") + # CCameraBase adds Transpose_* properties automatically + assert "Gain" in names + assert "Mode" in names + + # Read property through CMMCore + # FloatProperty stores 4 decimal places (MM convention) + assert float(core.getProperty("Cam", "Gain")) == 1.0 + assert core.getProperty("Cam", "Mode") == "Normal" + + # Write property through CMMCore → Python device updated + core.setProperty("Cam", "Gain", "42.5") + assert cam._gain == 42.5 + assert float(core.getProperty("Cam", "Gain")) == 42.5 + + core.setProperty("Cam", "Mode", "Fast") + assert cam._mode == "Fast" + assert core.getProperty("Cam", "Mode") == "Fast" + + # Limits should be enforced by CDeviceBase + assert core.hasPropertyLimits("Cam", "Gain") + assert core.getPropertyLowerLimit("Cam", "Gain") == 0.0 + assert core.getPropertyUpperLimit("Cam", "Gain") == 100.0 + + # Allowed values + allowed = core.getAllowedPropertyValues("Cam", "Mode") + assert set(allowed) == {"Normal", "Fast", "Slow"} + + +def test_load_py_device_adapter() -> None: + """Test building a DeviceAdapter in Python and registering it.""" + + class MyCam(MinimalCamera): + """A test camera.""" + + _TYPE = DeviceType.CameraDevice + + class MyShutter(MinimalShutter): + """A test shutter.""" + + _TYPE = DeviceType.ShutterDevice + + # Build the adapter in Python — scanning logic is Python's job + # ultimately, pymmcore-plus would likely have conveniences to accept a ModuleType + # and protocols to define device name, type, and description... + # but this is the low level API: + adapter = DeviceAdapter() + adapter.add_device_class(MyCam.__name__, MyCam, MyCam._TYPE, MyCam.__doc__) + adapter.add_device_class( + MyShutter.__name__, MyShutter, MyShutter._TYPE, MyShutter.__doc__ + ) + + core = CMMCore() + core.loadPyDeviceAdapter("MyHardware", adapter) + + # Device discovery should work + available = core.getAvailableDevices("MyHardware") + assert "MyCam" in available + assert "MyShutter" in available + + # Descriptions should work + descs = core.getAvailableDeviceDescriptions("MyHardware") + assert "A test camera." in descs + assert "A test shutter." in descs + + # Load devices through normal CMMCore flow + core.loadDevice("Cam1", "MyHardware", "MyCam") + core.loadDevice("Shutter1", "MyHardware", "MyShutter") + core.initializeDevice("Cam1") + core.initializeDevice("Shutter1") + + assert "Cam1" in core.getLoadedDevices() + assert "Shutter1" in core.getLoadedDevices() + + # Loaded device descriptions should come from the bridge's GetDescription + assert core.getDeviceDescription("Cam1") == "A test camera." + assert core.getDeviceDescription("Shutter1") == "A test shutter." + + # Devices should work normally + core.setCameraDevice("Cam1") + core.snapImage() + img = core.getImage() + assert img.shape == (32, 64) # MinimalCamera defaults + + core.setShutterDevice("Shutter1") + core.setShutterOpen(True) + assert core.getShutterOpen() is True + + # Can load a second instance of the same device class + core.loadDevice("Cam2", "MyHardware", "MyCam") + core.initializeDevice("Cam2") + assert "Cam2" in core.getLoadedDevices() + + +# ============================================================================ +# Minimal device stubs for new device types +# ============================================================================ + + +class MinimalStage(MinimalDevice): + def __init__(self) -> None: + self._pos_um = 0.0 + self._pos_steps = 0 + + def set_position_um(self, pos: float) -> None: + self._pos_um = pos + + def get_position_um(self) -> float: + return self._pos_um + + def set_relative_position_um(self, d: float) -> None: + self._pos_um += d + + def set_position_steps(self, steps: int) -> None: + self._pos_steps = steps + + def get_position_steps(self) -> int: + return self._pos_steps + + def set_adapter_origin_um(self, d: float) -> None: + pass + + def set_origin(self) -> None: + self._pos_um = 0.0 + self._pos_steps = 0 + + def get_limits(self) -> tuple[float, float]: + return (-10000.0, 10000.0) + + def move(self, velocity: float) -> None: + pass + + def stop(self) -> None: + pass + + def home(self) -> None: + self._pos_um = 0.0 + self._pos_steps = 0 + + def get_focus_direction(self) -> int: + return 0 # FocusDirectionUnknown + + def is_continuous_focus_drive(self) -> bool: + return False + + def is_stage_sequenceable(self) -> bool: + return False + + +class MinimalXYStage(MinimalDevice): + def __init__(self) -> None: + self._x_um = 0.0 + self._y_um = 0.0 + self._x_steps = 0 + self._y_steps = 0 + + # position (um) + def set_position_um(self, x: float, y: float) -> None: + self._x_um = x + self._y_um = y + self._x_steps = int(x / 0.1) + self._y_steps = int(y / 0.1) + + def get_position_um(self) -> tuple[float, float]: + return (self._x_um, self._y_um) + + def set_relative_position_um(self, dx: float, dy: float) -> None: + self.set_position_um(self._x_um + dx, self._y_um + dy) + + def set_adapter_origin_um(self, x: float, y: float) -> None: + pass + + # position (steps) + def set_position_steps(self, x: int, y: int) -> None: + self._x_steps = x + self._y_steps = y + self._x_um = x * 0.1 + self._y_um = y * 0.1 + + def get_position_steps(self) -> tuple[int, int]: + return (self._x_steps, self._y_steps) + + def set_relative_position_steps(self, x: int, y: int) -> None: + self.set_position_steps(self._x_steps + x, self._y_steps + y) + + # motion + def home(self) -> None: + self._x_steps = 0 + self._y_steps = 0 + self._x_um = 0.0 + self._y_um = 0.0 + + def stop(self) -> None: + pass + + def move(self, vx: float, vy: float) -> None: + pass + + # origin + def set_origin(self) -> None: + pass + + def set_x_origin(self) -> None: + pass + + def set_y_origin(self) -> None: + pass + + # limits + step size + def get_limits_um(self) -> tuple[float, float, float, float]: + return (-10000.0, 10000.0, -10000.0, 10000.0) + + def get_step_limits(self) -> tuple[int, int, int, int]: + return (-100000, 100000, -100000, 100000) + + def get_step_size_x_um(self) -> float: + return 0.1 + + def get_step_size_y_um(self) -> float: + return 0.1 + + # sequencing + def is_xy_stage_sequenceable(self) -> bool: + return False + + +class MinimalSignalIO(MinimalDevice): + def __init__(self) -> None: + self._gate_open = True + self._volts = 0.0 + self._min_volts = 0.0 + self._max_volts = 5.0 + + def set_gate_open(self, open: bool) -> None: + self._gate_open = open + + def get_gate_open(self) -> bool: + return self._gate_open + + def set_signal(self, volts: float) -> None: + self._volts = volts + + def get_signal(self) -> float: + return self._volts + + def get_limits(self) -> tuple[float, float]: + return (self._min_volts, self._max_volts) + + def is_da_sequenceable(self) -> bool: + return False + + +class MinimalMagnifier(MinimalDevice): + def __init__(self, mag: float = 10.0) -> None: + self._mag = mag + + def get_magnification(self) -> float: + return self._mag + + +class MinimalSerial(MinimalDevice): + def __init__(self) -> None: + self._buf = b"" + + def get_port_type(self) -> int: + return 1 # SerialPort + + def set_command(self, command: str, term: str) -> None: + self._buf = (command + term).encode() + + def get_answer(self, term: str) -> str: + return self._buf.decode() + + def write(self, data: bytes) -> None: + self._buf = data + + def read(self, max_bytes: int) -> bytes: + result = self._buf[:max_bytes] + self._buf = self._buf[max_bytes:] + return result + + def purge(self) -> None: + self._buf = b"" + + +class MinimalGalvo(MinimalDevice): + def __init__(self) -> None: + self._x = 0.0 + self._y = 0.0 + self._illumination = False + self._spot_interval = 0.0 + self._polygons: dict[int, list[tuple[float, float]]] = {} + self._repetitions = 1 + self._sequence_running = False + + def point_and_fire(self, x: float, y: float, time_us: float) -> None: + self._x = x + self._y = y + + def set_spot_interval(self, pulse_interval_us: float) -> None: + self._spot_interval = pulse_interval_us + + def set_position(self, x: float, y: float) -> None: + self._x = x + self._y = y + + def get_position(self) -> tuple[float, float]: + return (self._x, self._y) + + def set_illumination_state(self, on: bool) -> None: + self._illumination = on + + def get_x_range(self) -> float: + return 100.0 + + def get_x_minimum(self) -> float: + return 0.0 + + def get_y_range(self) -> float: + return 100.0 + + def get_y_minimum(self) -> float: + return 0.0 + + def add_polygon_vertex(self, polygon_index: int, x: float, y: float) -> None: + self._polygons.setdefault(polygon_index, []).append((x, y)) + + def delete_polygons(self) -> None: + self._polygons.clear() + + def load_polygons(self) -> None: + pass + + def set_polygon_repetitions(self, repetitions: int) -> None: + self._repetitions = repetitions + + def run_polygons(self) -> None: + pass + + def run_sequence(self) -> None: + self._sequence_running = True + + def stop_sequence(self) -> None: + self._sequence_running = False + + def get_channel(self) -> str: + return "" + + +class MinimalState(MinimalDevice): + def __init__(self, n_positions: int = 4, labels: list[str] | None = None) -> None: + self._n = n_positions + self._labels = labels + + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + super().initialize_bridge(create_property, notify) + if self._labels is not None: + for i, label in enumerate(self._labels): + notify.set_position_label(i, label) + + def get_number_of_positions(self) -> int: + return self._n + + +class MinimalAutoFocus(MinimalDevice): + def __init__(self) -> None: + self._continuous = False + self._offset = 0.0 + + def set_continuous_focusing(self, state: bool) -> None: + self._continuous = state + + def get_continuous_focusing(self) -> bool: + return self._continuous + + def is_continuous_focus_locked(self) -> bool: + return self._continuous + + def full_focus(self) -> None: + pass + + def incremental_focus(self) -> None: + pass + + def get_last_focus_score(self) -> float: + return 1.0 + + def get_current_focus_score(self) -> float: + return 1.0 + + def get_offset(self) -> float: + return self._offset + + def set_offset(self, offset: float) -> None: + self._offset = offset + + +class MinimalGeneric(MinimalDevice): + pass + + +class MinimalHub(MinimalDevice): + """Hub that discovers a camera and shutter as peripherals.""" + + def detect_installed_devices(self): + return [ + ("HubCam", MinimalCamera(), DeviceType.CameraDevice), + ("HubShutter", MinimalShutter(), DeviceType.ShutterDevice), + ] + + +class MinimalSLM(MinimalDevice): + def __init__(self, width: int = 128, height: int = 128) -> None: + self._width = width + self._height = height + self._exposure = 0.0 + self._image: np.ndarray | None = None + + def set_image(self, pixels: np.ndarray) -> None: + self._image = pixels + + def display_image(self) -> None: + pass + + def set_pixels_to(self, intensity: int) -> None: + pass + + def set_pixels_to_rgb(self, r: int, g: int, b: int) -> None: + pass + + def set_exposure(self, interval_ms: float) -> None: + self._exposure = interval_ms + + def get_exposure(self) -> float: + return self._exposure + + def get_width(self) -> int: + return self._width + + def get_height(self) -> int: + return self._height + + def get_number_of_components(self) -> int: + return 1 + + def get_bytes_per_pixel(self) -> int: + return 1 + + def is_slm_sequenceable(self) -> bool: + return False + + +# ============================================================================ +# Tests for new device types +# ============================================================================ + + +def test_load_py_stage() -> None: + core = CMMCore() + stage = MinimalStage() + core.loadPyDevice("Z", stage, DeviceType.StageDevice) + core.initializeDevice("Z") + core.setFocusDevice("Z") + + assert "Z" in core.getLoadedDevices() + assert core.getFocusDevice() == "Z" + + # Position (um) + core.setPosition(42.5) + assert stage._pos_um == 42.5 + assert core.getPosition() == 42.5 + + core.setRelativePosition(10.0) + assert stage._pos_um == 52.5 + + # Origin + core.setOrigin("Z") + assert stage._pos_um == 0.0 + + # Home + core.home("Z") + + # Focus direction + continuous focus drive + assert core.getFocusDirection("Z") == 0 + assert core.isContinuousFocusDrive("Z") is False + + # Sequenceable query + assert not core.isStageSequenceable("Z") + + +def test_load_py_xy_stage() -> None: + core = CMMCore() + xy = MinimalXYStage() + core.loadPyDevice("XY", xy, DeviceType.XYStageDevice) + core.initializeDevice("XY") + core.setXYStageDevice("XY") + + assert "XY" in core.getLoadedDevices() + + # Position (um) + core.setXYPosition(10.0, 20.0) + assert xy._x_um == 10.0 + assert xy._y_um == 20.0 + x, y = core.getXYPosition() + assert x == 10.0 + assert y == 20.0 + + # Relative position + core.setRelativeXYPosition(5.0, -5.0) + assert xy._x_um == 15.0 + assert xy._y_um == 15.0 + + # Home + core.home("XY") + assert xy._x_um == 0.0 + assert xy._y_um == 0.0 + + # Stop (no-op but exercises the bridge) + core.stop("XY") + + # Sequenceable query + assert not core.isXYStageSequenceable("XY") + + +def test_load_py_state() -> None: + core = CMMCore() + labels = ["DAPI", "FITC", "TRITC", "Cy5"] + state = MinimalState(n_positions=4, labels=labels) + core.loadPyDevice("Wheel", state, DeviceType.StateDevice) + core.initializeDevice("Wheel") + + assert "Wheel" in core.getLoadedDevices() + assert core.getNumberOfStates("Wheel") == 4 + + # Labels set during initialize should be accessible + assert core.getStateLabels("Wheel") == ["DAPI", "FITC", "TRITC", "Cy5"] + + +def test_load_py_autofocus() -> None: + core = CMMCore() + af = MinimalAutoFocus() + core.loadPyDevice("AF", af, DeviceType.AutoFocusDevice) + core.initializeDevice("AF") + core.setAutoFocusDevice("AF") + + assert "AF" in core.getLoadedDevices() + + # Offset + core.setAutoFocusOffset(5.0) + assert af._offset == 5.0 + assert core.getAutoFocusOffset() == 5.0 + + # Continuous focusing + core.enableContinuousFocus(True) + assert af._continuous is True + assert core.isContinuousFocusEnabled() is True + assert core.isContinuousFocusLocked() is True + + core.enableContinuousFocus(False) + assert af._continuous is False + + # Focus operations + core.fullFocus() + core.incrementalFocus() + + # Scores + assert core.getLastFocusScore() == 1.0 + assert core.getCurrentFocusScore() == 1.0 + + +def test_load_py_signal_io() -> None: + core = CMMCore() + da = MinimalSignalIO() + core.loadPyDevice("DA", da, DeviceType.SignalIODevice) + core.initializeDevice("DA") + + assert "DA" in core.getLoadedDevices() + assert core.getDeviceType("DA") == DeviceType.SignalIODevice + + +def test_load_py_magnifier() -> None: + core = CMMCore() + mag = MinimalMagnifier(mag=40.0) + core.loadPyDevice("Mag", mag, DeviceType.MagnifierDevice) + core.initializeDevice("Mag") + + assert "Mag" in core.getLoadedDevices() + assert core.getDeviceType("Mag") == DeviceType.MagnifierDevice + assert core.getMagnificationFactor() == 40.0 + + +def test_load_py_serial() -> None: + core = CMMCore() + ser = MinimalSerial() + core.loadPyDevice("COM1", ser, DeviceType.SerialDevice) + core.initializeDevice("COM1") + + assert "COM1" in core.getLoadedDevices() + assert core.getDeviceType("COM1") == DeviceType.SerialDevice + + # Command/answer round-trip + core.setSerialPortCommand("COM1", "HELLO", "\n") + assert core.getSerialPortAnswer("COM1", "\n") == "HELLO\n" + + +def test_load_py_galvo() -> None: + core = CMMCore() + galvo = MinimalGalvo() + core.loadPyDevice("Galvo", galvo, DeviceType.GalvoDevice) + core.initializeDevice("Galvo") + core.setGalvoDevice("Galvo") + + assert "Galvo" in core.getLoadedDevices() + assert core.getDeviceType("Galvo") == DeviceType.GalvoDevice + + # Position + core.setGalvoPosition("Galvo", 10.0, 20.0) + assert galvo._x == 10.0 + assert galvo._y == 20.0 + pos = core.getGalvoPosition("Galvo") + assert pos == (10.0, 20.0) + + # Range + assert core.getGalvoXRange("Galvo") == 100.0 + assert core.getGalvoYRange("Galvo") == 100.0 + + # Illumination + core.setGalvoIlluminationState("Galvo", True) + assert galvo._illumination is True + + # Polygons + core.addGalvoPolygonVertex("Galvo", 0, 1.0, 2.0) + core.addGalvoPolygonVertex("Galvo", 0, 3.0, 4.0) + assert galvo._polygons[0] == [(1.0, 2.0), (3.0, 4.0)] + + core.deleteGalvoPolygons("Galvo") + assert galvo._polygons == {} + + # Point and fire + core.pointGalvoAndFire("Galvo", 5.0, 5.0, 100.0) + assert galvo._x == 5.0 + assert galvo._y == 5.0 + + # Spot interval + core.setGalvoSpotInterval("Galvo", 50.0) + assert galvo._spot_interval == 50.0 + + # Range minimums + assert core.getGalvoXMinimum("Galvo") == 0.0 + assert core.getGalvoYMinimum("Galvo") == 0.0 + + # Polygon load/repetitions/run + core.addGalvoPolygonVertex("Galvo", 0, 0.0, 0.0) + core.addGalvoPolygonVertex("Galvo", 0, 10.0, 10.0) + core.setGalvoPolygonRepetitions("Galvo", 3) + assert galvo._repetitions == 3 + core.loadGalvoPolygons("Galvo") + core.runGalvoPolygons("Galvo") + + # Channel + assert core.getGalvoChannel("Galvo") == "" + + # Sequence + core.runGalvoSequence("Galvo") + assert galvo._sequence_running is True + + +def test_load_py_generic() -> None: + core = CMMCore() + dev = MinimalGeneric() + core.loadPyDevice("Gen", dev, DeviceType.GenericDevice) + core.initializeDevice("Gen") + + assert "Gen" in core.getLoadedDevices() + assert core.getDeviceType("Gen") == DeviceType.GenericDevice + + +def test_load_py_hub() -> None: + """Hub discovers peripherals via detect_installed_devices.""" + core = CMMCore() + hub = MinimalHub() + core.loadPyDevice("Hub", hub, DeviceType.HubDevice) + core.initializeDevice("Hub") + + assert core.getDeviceType("Hub") == DeviceType.HubDevice + + # Hub should discover its peripherals + peripherals = core.getInstalledDevices("Hub") + assert "HubCam" in peripherals + assert "HubShutter" in peripherals + + # Peripheral descriptions should come from the Python class docstring + cam_desc = core.getInstalledDeviceDescription("Hub", "HubCam") + shutter_desc = core.getInstalledDeviceDescription("Hub", "HubShutter") + assert "Minimal Python camera" in cam_desc + assert "Minimal Python shutter" in shutter_desc + + +def test_load_py_slm() -> None: + core = CMMCore() + slm = MinimalSLM(width=64, height=32) + core.loadPyDevice("SLM", slm, DeviceType.SLMDevice) + core.initializeDevice("SLM") + core.setSLMDevice("SLM") + + assert "SLM" in core.getLoadedDevices() + assert core.getSLMWidth("SLM") == 64 + assert core.getSLMHeight("SLM") == 32 + assert core.getSLMNumberOfComponents("SLM") == 1 + assert core.getSLMBytesPerPixel("SLM") == 1 + + core.setSLMExposure("SLM", 100.0) + assert slm._exposure == 100.0 + assert core.getSLMExposure("SLM") == 100.0 + + # Set and verify image data round-trips through the bridge + img = np.arange(64 * 32, dtype=np.uint8).reshape(32, 64) + core.setSLMImage("SLM", img) + assert slm._image is not None + assert slm._image.shape == (32, 64) + np.testing.assert_array_equal(slm._image, img) + + # DisplayImage + core.displaySLMImage("SLM") + + # SetPixelsTo (uniform intensity) + core.setSLMPixelsTo("SLM", 128) + + # SetPixelsTo (RGB) + core.setSLMPixelsTo("SLM", 10, 20, 30) + + # Not sequenceable + assert not slm.is_slm_sequenceable() + + +def test_slm_rgb_image() -> None: + """RGB SLM receives properly shaped (h, w, 3) array.""" + + class RGBSlm(MinimalSLM): + def get_number_of_components(self) -> int: + return 3 + + core = CMMCore() + slm = RGBSlm(width=8, height=4) + core.loadPyDevice("SLM", slm, DeviceType.SLMDevice) + core.initializeDevice("SLM") + + img = np.ones((4, 8, 3), dtype=np.uint8) * 42 + core.setSLMImage("SLM", img) + assert slm._image is not None + assert slm._image.shape == (4, 8, 3) + np.testing.assert_array_equal(slm._image, img) + + +def test_property_sequencing() -> None: + """Property sequencing lifecycle: query, load, start, stop.""" + + class SeqDevice(MinimalDevice): + def __init__(self) -> None: + self._voltage = 0.0 + self._loaded_seq: list[str] = [] + self._seq_started = False + self._seq_stopped = False + self._voltage_handle = None + + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + super().initialize_bridge(create_property, notify) + # Non-sequenceable property + create_property( + "Mode", + "Off", + 1, + False, + getter=lambda: "Off", + ) + # Sequenceable property + self._voltage_handle = create_property( + "Voltage", + "0.0", + 2, + False, + getter=lambda: self._voltage, + setter=lambda v: setattr(self, "_voltage", float(v)), + sequence_max_length=10, + sequence_loader=lambda seq: setattr(self, "_loaded_seq", seq), + sequence_starter=lambda: setattr(self, "_seq_started", True), + sequence_stopper=lambda: setattr(self, "_seq_stopped", True), + ) + + core = CMMCore() + dev = SeqDevice() + core.loadPyDevice("Dev", dev, DeviceType.GenericDevice) + core.initializeDevice("Dev") + + # Non-sequenceable property + assert not core.isPropertySequenceable("Dev", "Mode") + + # Sequenceable property + assert core.isPropertySequenceable("Dev", "Voltage") + assert core.getPropertySequenceMaxLength("Dev", "Voltage") == 10 + + # Load a sequence + core.loadPropertySequence("Dev", "Voltage", ["1.0", "2.0", "3.0"]) + assert dev._loaded_seq == ["1.0", "2.0", "3.0"] + + # Start / stop + core.startPropertySequence("Dev", "Voltage") + assert dev._seq_started + + core.stopPropertySequence("Dev", "Voltage") + assert dev._seq_stopped + + # Dynamic max length update via PropertyHandle + assert core.getPropertySequenceMaxLength("Dev", "Voltage") == 10 + dev._voltage_handle.set_sequence_max_length(20) + assert core.getPropertySequenceMaxLength("Dev", "Voltage") == 20 + + +def test_device_notifications() -> None: + """Test that Python devices can emit notifications via DeviceCallbacks.""" + + received: dict[str, tuple] = {} + + class Listener(pmn.MMEventCallback): + def onPropertyChanged(self, dev: str, name: str, value: str) -> None: + received["onPropertyChanged"] = (dev, name, value) + + def onExposureChanged(self, dev: str, exposure: float) -> None: + received["onExposureChanged"] = (dev, exposure) + + class NotifyingCamera(MinimalCamera): + def set_exposure(self, ms: float) -> None: + self._exposure = ms + # Notify CMMCore that exposure changed + if self._notify is not None: + self._notify.on_exposure_changed(ms) + + core = CMMCore() + cam = NotifyingCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + + cb = Listener() + core.registerCallback(cb) + + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + + # Setting exposure through CMMCore triggers the bridge → Python setter + # → Python calls notify.on_exposure_changed → CMMCore posts notification + # → notification thread delivers to callback (async) + core.setExposure(42.0) + assert cam._exposure == 42.0 + + # Wait for async notification delivery + deadline = time.time() + 2.0 + while "onExposureChanged" not in received and time.time() < deadline: + time.sleep(0.01) + + assert "onExposureChanged" in received + assert received["onExposureChanged"] == ("Cam", 42.0) + + +def test_python_exception_surfaces_as_runtime_error() -> None: + """Python exceptions in device methods should surface with traceback info.""" + + class BrokenStage(MinimalStage): + def get_position_um(self) -> float: + raise ValueError("motor not homed") + + core = CMMCore() + stage = BrokenStage() + core.loadPyDevice("Z", stage, DeviceType.StageDevice) + core.initializeDevice("Z") + core.setFocusDevice("Z") + + try: + core.getPosition() + msg = "" + except RuntimeError as e: + msg = str(e) + + assert "motor not homed" in msg, f"Expected Python error message, got: {msg!r}" + + +def test_python_exception_in_property_getter() -> None: + """Python exceptions in property getters should surface.""" + + class BadCamera(MinimalCamera): + def initialize_bridge( + self, create_property: CreatePropertyFn, notify: DeviceCallbacks + ) -> None: + create_property( + "Bad", + "0", + 2, + False, + getter=lambda: 1 / 0, # ZeroDivisionError + ) + + core = CMMCore() + cam = BadCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + + try: + core.getProperty("Cam", "Bad") + msg = "" + except RuntimeError as e: + msg = str(e) + + assert "ZeroDivisionError" in msg, f"Expected Python error info, got: {msg!r}" + + +# ============================================================================ +# Device-level sequencing tests +# ============================================================================ + + +class SequenceableCamera(MinimalCamera): + """Camera that supports exposure sequencing.""" + + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self._loaded_exposure_seq: list[float] = [] + self._exp_seq_started = False + self._exp_seq_stopped = False + + def is_exposure_sequenceable(self) -> bool: + return True + + def get_exposure_sequence_max_length(self) -> int: + return 5 + + def load_exposure_sequence(self, sequence: list[float]) -> None: + self._loaded_exposure_seq = list(sequence) + + def start_exposure_sequence(self) -> None: + self._exp_seq_started = True + + def stop_exposure_sequence(self) -> None: + self._exp_seq_stopped = True + + +class SequenceableStage(MinimalStage): + """Stage that supports sequencing.""" + + def __init__(self) -> None: + super().__init__() + self._loaded_seq: list[float] = [] + self._seq_started = False + self._seq_stopped = False + + def is_stage_sequenceable(self) -> bool: + return True + + def get_stage_sequence_max_length(self) -> int: + return 10 + + def load_stage_sequence(self, positions: list[float]) -> None: + self._loaded_seq = list(positions) + + def start_stage_sequence(self) -> None: + self._seq_started = True + + def stop_stage_sequence(self) -> None: + self._seq_stopped = True + + +class SequenceableXYStage(MinimalXYStage): + """XY stage that supports sequencing.""" + + def __init__(self) -> None: + super().__init__() + self._loaded_seq: list[tuple[float, float]] = [] + self._seq_started = False + self._seq_stopped = False + + def is_xy_stage_sequenceable(self) -> bool: + return True + + def get_xy_stage_sequence_max_length(self) -> int: + return 8 + + def load_xy_stage_sequence(self, positions: list[tuple[float, float]]) -> None: + self._loaded_seq = [tuple(p) for p in positions] + + def start_xy_stage_sequence(self) -> None: + self._seq_started = True + + def stop_xy_stage_sequence(self) -> None: + self._seq_stopped = True + + +class SequenceableSLM(MinimalSLM): + """SLM that supports sequencing.""" + + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self._loaded_seq: list[np.ndarray] = [] + self._seq_started = False + self._seq_stopped = False + + def is_slm_sequenceable(self) -> bool: + return True + + def get_slm_sequence_max_length(self) -> int: + return 4 + + def load_slm_sequence(self, images: list[np.ndarray]) -> None: + self._loaded_seq = [np.array(img) for img in images] + + def start_slm_sequence(self) -> None: + self._seq_started = True + + def stop_slm_sequence(self) -> None: + self._seq_stopped = True + + +def test_exposure_sequencing() -> None: + """Exposure sequence lifecycle: query, load, start, stop.""" + core = CMMCore() + cam = SequenceableCamera() + core.loadPyDevice("Cam", cam, DeviceType.CameraDevice) + core.initializeDevice("Cam") + core.setCameraDevice("Cam") + + assert core.isExposureSequenceable("Cam") + assert core.getExposureSequenceMaxLength("Cam") == 5 + + core.loadExposureSequence("Cam", [10.0, 20.0, 30.0]) + assert cam._loaded_exposure_seq == [10.0, 20.0, 30.0] + + core.startExposureSequence("Cam") + assert cam._exp_seq_started + + core.stopExposureSequence("Cam") + assert cam._exp_seq_stopped + + +def test_stage_sequencing() -> None: + """Stage sequence lifecycle: query, load, start, stop.""" + core = CMMCore() + stage = SequenceableStage() + core.loadPyDevice("Z", stage, DeviceType.StageDevice) + core.initializeDevice("Z") + core.setFocusDevice("Z") + + assert core.isStageSequenceable("Z") + assert core.getStageSequenceMaxLength("Z") == 10 + + core.loadStageSequence("Z", [0.0, 1.0, 2.0, 3.0]) + assert stage._loaded_seq == [0.0, 1.0, 2.0, 3.0] + + core.startStageSequence("Z") + assert stage._seq_started + + core.stopStageSequence("Z") + assert stage._seq_stopped + + +def test_xy_stage_sequencing() -> None: + """XY stage sequence lifecycle: query, load, start, stop.""" + core = CMMCore() + xy = SequenceableXYStage() + core.loadPyDevice("XY", xy, DeviceType.XYStageDevice) + core.initializeDevice("XY") + core.setXYStageDevice("XY") + + assert core.isXYStageSequenceable("XY") + assert core.getXYStageSequenceMaxLength("XY") == 8 + + core.loadXYStageSequence("XY", [1.0, 3.0, 5.0], [2.0, 4.0, 6.0]) + assert xy._loaded_seq == [(1.0, 2.0), (3.0, 4.0), (5.0, 6.0)] + + core.startXYStageSequence("XY") + assert xy._seq_started + + core.stopXYStageSequence("XY") + assert xy._seq_stopped + + +def test_slm_sequencing() -> None: + """SLM sequence lifecycle: query, load, start, stop.""" + core = CMMCore() + slm = SequenceableSLM(width=8, height=4) + core.loadPyDevice("SLM", slm, DeviceType.SLMDevice) + core.initializeDevice("SLM") + core.setSLMDevice("SLM") + + assert core.getSLMSequenceMaxLength("SLM") == 4 + + img1 = np.ones((4, 8), dtype=np.uint8) * 10 + img2 = np.ones((4, 8), dtype=np.uint8) * 20 + core.loadSLMSequence("SLM", [img1, img2]) + assert len(slm._loaded_seq) == 2 + np.testing.assert_array_equal(slm._loaded_seq[0], img1) + np.testing.assert_array_equal(slm._loaded_seq[1], img2) + + core.startSLMSequence("SLM") + assert slm._seq_started + + core.stopSLMSequence("SLM") + assert slm._seq_stopped diff --git a/tests/test_mmcore_coverage.py b/tests/test_mmcore_coverage.py index f23c3903..62da1ec3 100644 --- a/tests/test_mmcore_coverage.py +++ b/tests/test_mmcore_coverage.py @@ -47,7 +47,7 @@ def test_property_sequenceable(demo_core: pmn.CMMCore) -> None: def test_set_property_numeric_overloads(demo_core: pmn.CMMCore) -> None: demo_core.setProperty("Camera", "Exposure", 25.0) assert demo_core.getProperty("Camera", "Exposure") == "25.0000" - demo_core.setProperty("Camera", "Exposure", float(10.0)) + demo_core.setProperty("Camera", "Exposure", 10.0) assert demo_core.getProperty("Camera", "Exposure") == "10.0000" # bool overload maps True -> "1" demo_core.setProperty("Camera", "TransposeCorrection", True)