Abstract Microscope base class and introduce ThermoDigitalTwin

.github/workflows/test.yaml

@@ -0,0 +1,35 @@
+name: Test
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    name: Run Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+          cache-dependency-glob: "uv.lock"
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libtango-dev
+
+      - name: Set up Python
+        run: uv python install
+
+      - name: Install dependencies
+        run: uv sync --dev
+
+      # - name: Lint with ruff
+      #   run: uv run ruff check .
+
+      - name: Run tests
+        run: uv run pytest

asyncroscopy/Microscope.py

@@ -20,12 +20,20 @@
 
 import json
 import time
+from typing import Optional
+
+from abc import abstractmethod, ABC, ABCMeta
+
+import numpy as np
 import tango
 from tango import AttrWriteType, DevEncoded, DevState
-from tango.server import Device, attribute, command, device_property
+from tango.server import Device, DeviceMeta, attribute, command, device_property
 
+class CombinedMeta(DeviceMeta, ABCMeta):
+    """Combines Tango DeviceMeta and ABCMeta to allow abstract methods in Devices."""
+    pass
 
-class Microscope(Device):
+class Microscope(Device, metaclass=CombinedMeta):
     """
     Top-level TEM microscope device.
     Detector-specific settings (dwell time, resolution) are stored in
@@ -79,20 +87,21 @@ class Microscope(Device):
     # ------------------------------------------------------------------
     # Initialisation
     # ------------------------------------------------------------------
+    @abstractmethod
     def init_device(self) -> None:
-        """Placeholder for more specific device init to be inheritted"""
-        raise NotImplementedError(f"Must define a class-specific init_device() method")
+        print(f"Must define a class-specific init_device() method")
 
+    @abstractmethod
     def _connect(self):
-        raise NotImplementedError(
-            "Subclasses must implement _connect()"
-        )
+        print(f"Must define a class-specific _connect() method")
 
+    @abstractmethod
     def _connect_hardware(self) -> None:
-        raise NotImplementedError(f"Must define a class-specific _connect_hardware() method")
+        print(f"Must define a class-specific _connect_hardware() method")
 
+    @abstractmethod
     def _connect_detector_proxies(self) -> None:
-        raise NotImplementedError(f"Must define a class-specific _connect_detector_proxies() method")  
+        print(f"Must define a class-specific _connect_detector_proxies() method")  
 
     # ------------------------------------------------------------------
     # Attribute read methods

asyncroscopy/ThermoDigitalTwin.py

@@ -0,0 +1,100 @@
+"""
+Digital twin version of ThermoMicroscope for simulated image generation.
+
+Useful for testing and development without requiring AutoScript hardware.
+"""
+
+import json
+import time
+import numpy as np
+from tango import DevState, DeviceProxy, DevFailed
+from tango.server import Device, attribute
+
+from asyncroscopy.Microscope import Microscope
+
+class ThermoDigitalTwin(Microscope):
+    """
+    A digital twin for the ThermoMicroscope.
+    """
+
+    manufacturer = attribute(
+        label="ThermoDigitalTwin",
+        dtype=str,
+        doc="Simulation backend",
+    )
+
+    def init_device(self) -> None:
+        Device.init_device(self)
+        self.set_state(DevState.INIT)
+
+        # Internal state
+        self._stem_mode = True
+        self._detector_proxies = {}
+        self._manufacturer = "UTKTeam"
+
+        self._connect()
+
+    def _connect(self):
+        """Simulate connection by connecting to detector proxies."""
+        self._connect_detector_proxies()
+        self.set_state(DevState.ON)
+
+    def _connect_detector_proxies(self) -> None:
+        """
+        Connect to simulated detector proxies. 
+        """
+        addresses: dict[str, str] = {
+            "haadf": self.haadf_device_address,
+            "AdvancedAcquistion": self.advanced_acquisition_device_address,
+        }
+
+        for name, address in addresses.items():
+            if not address:
+                continue
+            try:
+                self._detector_proxies[name] = DeviceProxy(address)
+                self.info_stream(f"Connected to detector proxy: {name} @ {address}")
+            except DevFailed as e:
+                self.error_stream(f"Failed to connect to {name} proxy at {address}: {e}")
+
+    def read_manufacturer(self) -> str:
+        return self._manufacturer
+
+    def _acquire_stem_image(
+        self,
+        detector_name: str,
+        width: int,
+        height: int,
+        dwell_time: float,
+    ) -> np.ndarray:
+        """Generate a random image to simulate acquisition."""
+        self.info_stream(f"Simulating {detector_name} image: {width}x{height}")
+        rng = np.random.default_rng()
+        return rng.integers(0, 65535, size=(height, width), dtype=np.uint16)
+
+    def _acquire_stem_image_advanced(
+        self,
+        detector_names: list[str],
+        base_resolution: int,
+        scan_region: list[float],
+        dwell_time: float,
+        auto_beam_blank: bool,
+    ) -> list[np.ndarray]:
+        """Generate multiple random images to simulate simultaneous acquisition."""
+        self.info_stream(f"Simulating advanced acquisition for detectors: {detector_names}")
+        rng = np.random.default_rng()
+
+        # Calculate dimensions from scan_region [left, top, width, height]
+        width = int(base_resolution * scan_region[2])
+        height = int(base_resolution * scan_region[3])
+
+        return [rng.integers(0, 65535, size=(height, width), dtype=np.uint16) 
+                for _ in detector_names]
+
+
+# ----------------------------------------------------------------------
+# Server entry point
+# ----------------------------------------------------------------------
+
+if __name__ == "__main__":
+    ThermoDigitalTwin.run_server()

pyproject.toml

@@ -18,7 +18,6 @@ dependencies = [
     "numpy>=2.3.5",
     "opencv-python>=4.13.0.92",
     "pillow>=12.1.0",
-    "pyside6>=6.10.2",
     "pytango==10.1.2",
     "pytest>=9.0.2",
     "ruff>=0.15.2",

tests/conftest.py

@@ -17,20 +17,12 @@
 
 # Import device classes to test
 from asyncroscopy.detectors.HAADF import HAADF
-from asyncroscopy.Microscope import Microscope as RealMicroscope
+from asyncroscopy.ThermoDigitalTwin import ThermoDigitalTwin
 
 
-# ---- Test-only microscope subclass ----
-# Your Microscope._connect_autoscript() currently references self.autoscript_host
-# but you only defined autoscript_host_ip/autoscript_host_port.
-# In CI you don't want AutoScript anyway, so we bypass it here.
-class TestMicroscope(RealMicroscope):
-    def _connect_autoscript(self) -> None:
-        self.warn_stream("AutoScript disabled in tests (forcing simulation mode)")
-        self._microscope = None
+# We use ThermoDigitalTwin as our simulated microscope for all tests.
 
-
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="session")
 def tango_ctx():
     """
     One Tango device server hosting HAADF + Microscope together.
@@ -51,16 +43,12 @@ def tango_ctx():
             ],
         },
         {
-            "class": TestMicroscope,
+            "class": ThermoDigitalTwin,
             "devices": [
                 {
-                    "name": "test/nodb/microscope",
+                    "name": "test/nodb/twin",
                     "properties": {
-                        # IMPORTANT: address must match the HAADF device name above
                         "haadf_device_address": "test/nodb/haadf",
-                        # you can also set these if you later fix autoscript_host usage
-                        "autoscript_host_ip": "localhost",
-                        "autoscript_host_port": "9090",
                     },
                 }
             ],
@@ -75,11 +63,16 @@ def tango_ctx():
         yield ctx
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="session")
 def haadf_proxy(tango_ctx) -> tango.DeviceProxy:
     return tango.DeviceProxy("test/nodb/haadf")
 
 
-@pytest.fixture(scope="module")
-def microscope_proxy(tango_ctx) -> tango.DeviceProxy:
-    return tango.DeviceProxy("test/nodb/microscope")
+@pytest.fixture(scope="session")
+def twin_proxy(tango_ctx) -> tango.DeviceProxy:
+    return tango.DeviceProxy("test/nodb/twin")
+
+
+@pytest.fixture(scope="session")
+def microscope_proxy(twin_proxy):
+    return twin_proxy

tests/test_digital_twin.py

@@ -0,0 +1,42 @@
+"""
+Tests for the ThermoDigitalTwin Tango device.
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+import json
+import numpy as np
+import pytest
+import tango
+from tango.test_context import MultiDeviceTestContext
+
+from asyncroscopy.detectors.HAADF import HAADF
+from asyncroscopy.ThermoDigitalTwin import ThermoDigitalTwin
+
+# Using shared twin_proxy from conftest.py
+
+class TestThermoDigitalTwin:
+
+    def test_state_is_on(self, twin_proxy):
+        assert twin_proxy.state() == tango.DevState.ON
+
+    def test_manufacturer_is_digital_twin(self, twin_proxy):
+        assert twin_proxy.manufacturer == "UTKTeam"
+
+    def test_get_image_returns_valid_data(self, twin_proxy):
+        json_meta, raw_bytes = twin_proxy.get_image("haadf")
+        meta = json.loads(json_meta)
+
+        assert meta["detector"] == "haadf"
+        assert "shape" in meta
+        assert "dtype" in meta
+
+        image = np.frombuffer(raw_bytes, dtype=meta["dtype"]).reshape(meta["shape"])
+        assert image.shape == tuple(meta["shape"])
+        assert image.dtype == np.uint16
+
+    def test_unknown_detector_raises(self, twin_proxy):
+        with pytest.raises(tango.DevFailed):
+            twin_proxy.get_image("void")