Feat: GUI updated and using process manager

asyncroscopy/data/data_reader.py

@@ -0,0 +1,79 @@
+"""Shared logic for reading Tiled dataset metadata and small previews.
+
+Both the MCP bridge (``get_data_from_key``) and the electron microscope's
+legacy byte-over-Tango command (``get_image_data_cached``) need the same
+thing: given an already-resolved Tiled node, describe its shape/dtype/attrs
+and a small flattened preview. This module is the one place that logic
+lives; callers are responsible for resolving the Tiled client/node
+themselves, since that involves a DeviceProxy in one case and an in-process
+DeviceProxy in the other.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+
+def numpy_to_python(obj: Any) -> Any:
+    """Recursively convert numpy types to Python types for JSON serialization."""
+    if isinstance(obj, np.ndarray):
+        return numpy_to_python(obj.tolist())
+    if isinstance(obj, np.generic):
+        return obj.item()
+    if isinstance(obj, dict):
+        return {k: numpy_to_python(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        conv = [numpy_to_python(v) for v in obj]
+        return tuple(conv) if isinstance(obj, tuple) else conv
+    return obj
+
+
+def describe_tiled_node(key: str, uri: str, node: Any, max_values: int = 64) -> dict[str, Any]:
+    """Build shape/dtype/attrs metadata plus a small flattened preview for one Tiled node."""
+    limit = max(0, int(max_values))
+    suffix = key.rsplit(".", 1)[-1].lower() if "." in key else "unknown"
+    result: dict[str, Any] = {
+        "key": key,
+        "uri": uri,
+        "format": "hdf5" if suffix in {"h5", "hdf5"} else suffix,
+        "attrs": numpy_to_python(dict(getattr(node, "metadata", {}) or {})),
+    }
+    datasets: list[dict[str, Any]] = []
+
+    def visit(current: Any, name: str = "") -> None:
+        read = getattr(current, "read", None)
+        if callable(read):
+            shape = tuple(getattr(current, "shape", ()) or ())
+            if limit == 0:
+                array = np.asarray([], dtype=getattr(current, "dtype", float))
+            elif shape:
+                remaining = limit
+                slices = []
+                for size in reversed(shape):
+                    take = min(int(size), max(1, remaining))
+                    slices.append(slice(0, take))
+                    remaining = (remaining + take - 1) // take
+                array = np.asarray(read(tuple(reversed(slices))))
+            else:
+                array = np.asarray(read())
+            item: dict[str, Any] = {
+                "name": name,
+                "shape": list(shape or array.shape),
+                "dtype": str(getattr(current, "dtype", array.dtype)),
+                "attrs": numpy_to_python(dict(getattr(current, "metadata", {}) or {})),
+                "preview": numpy_to_python(array.reshape(-1)[:limit]),
+            }
+            datasets.append(item)
+            return
+
+        keys = getattr(current, "keys", None)
+        if callable(keys):
+            for child_name in keys():
+                child_path = f"{name}/{child_name}" if name else str(child_name)
+                visit(current[child_name], child_path)
+
+    visit(node)
+    result["datasets"] = datasets
+    return result

asyncroscopy/data/data_writer.py

@@ -71,6 +71,8 @@ def save_acquisition(
         raise ValueError(f"Unsupported output_format {output_format!r}; expected '.h5' or '.tiff'")
     detector_list = list(detectors) if isinstance(detectors, (list, tuple)) else [detectors]
     data_list = list(data) if isinstance(data, (list, tuple)) else [data]
+    if not data_list:
+        raise ValueError("save_acquisition called with no data to save (empty detector/data list)")
     attrs_list = dataset_attrs if isinstance(dataset_attrs, list) else [dataset_attrs] * len(data_list)
 
     if output_format == ".tiff":

asyncroscopy/instruments/electron_microscope/digital_twin.py

@@ -16,6 +16,7 @@
 from tango.server import Device, attribute, device_property
 
 from asyncroscopy.instruments.electron_microscope.electron_microscope import ElectronMicroscope
+from asyncroscopy.instruments.electron_microscope.detectors.camera import CAMERA
 from asyncroscopy.data.data_writer import save_acquisition
 
 DEFAULT_ACQUISITION_DIR = "outputs/tiled_acquisitions"
@@ -129,6 +130,7 @@ def _connect(self):
     def _connect_detector_proxies(self) -> None:
         """Build DeviceProxy objects for each configured detector device."""
         addresses: dict[str, str] = {
+            "camera": self.camera_device_address,
             "eds": self.eds_device_address,
             "stage": self.stage_device_address,
             "scan": self.scan_device_address,
@@ -529,6 +531,25 @@ def _acquire_scanned_image(
             images.append(image)
         return save_acquisition(self, data_server, "stem_image", detector_list, images, output_format=output_format)
 
+    def _acquire_camera_image(
+        self,
+        imsize: int,
+        exposure_time: float,
+        detector: str,
+        readout_area: str,
+        frame_combining: int = 1,
+        electron_counting: bool = True,
+        output_format: str = ".h5",
+    ) -> str:
+        """Simulate a single-shot camera acquisition.
+
+        DigitalTwin only models STEM-probe imaging (see _render_stem_image), not a
+        separate TEM-mode camera; reuse the same HAADF renderer as
+        acquire_scanned_image so acquire_camera_image returns a usable fake image
+        instead of the "unsupported" error from the base ElectronMicroscope class.
+        """
+        return self._acquire_scanned_image(int(imsize), float(exposure_time), ["haadf"], [0.0, 0.0, 1.0, 1.0], output_format)
+
     def _simulate_spectrum(self, detector_name: str, exposure_time: float) -> dict[str, float]:
         """Simulate EDS spectrum acquisition at the current beam position weighted by surrounding particles."""
         self._sync_stage_from_proxy()
@@ -587,7 +608,15 @@ def _acquire_spectrum(self, detector_name: str, exposure_time: float) -> str:
         spectrum = self._simulate_spectrum(detector_name, exposure_time)
         data_server = self._detector_proxies.get("data")
         spectrum_array = np.array(list(spectrum.values()), dtype=np.float64)
-        return save_acquisition(self, data_server, "spectrum", detector_name, spectrum_array, dataset_name="spectrum")
+        return save_acquisition(
+            self,
+            data_server,
+            "spectrum",
+            detector_name,
+            spectrum_array,
+            dataset_name="spectrum",
+            dataset_attrs={"elements": list(spectrum.keys())},
+        )
 
     def _place_beam(self, position) -> None:
         """Place the electron beam at the specified [x, y] coordinates."""
@@ -634,6 +663,35 @@ def _move_stage(self, position):
         self._stage_position = target
         self._update_view_cache(force=False)
 
+    def _get_parameters(self) -> str:
+        """Return all simulated status parameters as a JSON string.
+
+        The base class exposes this through the get_parameters Tango command,
+        which is typed DevString — returning a dict (or the inherited abstract
+        stub's None) makes the command fail with a Tango translation error.
+
+        The detector keys are deliberately split by what they mean: device_proxies
+        are the twin's connected settings devices (scan, stage, data, ... — not
+        detectors), scan_detectors is what the twin's STEM renderer actually
+        produces (a simulated HAADF signal, whatever label is requested),
+        spectrum_detectors is what acquire_spectrum accepts, and camera_detectors
+        are the names the CAMERA device validates writes against. An earlier
+        draft published the proxy keys under a "detectors" key, which told an
+        agent that "stage" and "data" were detectors it could scan with.
+        """
+        self._sync_stage_from_proxy()
+        parameters = {
+            "manufacturer": self._manufacturer,
+            "stem_mode": bool(self._stem_mode),
+            "defocus_m": float(self._defocus),
+            "device_proxies": sorted(self._detector_proxies.keys()),
+            "scan_detectors": ["haadf"],
+            "spectrum_detectors": ["eds"],
+            "camera_detectors": sorted(CAMERA._CAMERA_DETECTORS),
+            **self._viewport_metadata(),
+        }
+        return json.dumps(parameters)
+
     def get_viewport_metadata(self) -> str:
         """Return JSON-formatted metadata regarding the current simulation viewport and environment state."""
         self._sync_stage_from_proxy()