Merge pull request #126 from pycroscopy/jeol-cm

Towards JEOL.py(TEM)

Author: utkarshp1161

README.md

@@ -50,18 +50,15 @@ See - docs/dev_guide.md
 
 ### Core installation (simulation mode)
 
-```bash
-pip install --find-links ./stubs -e .
-```
-
-or with `uv`:
-
 ```bash
 uv sync
 ```
 
-This installs `asyncroscopy` and all core dependencies. AutoScript is not required—the
-framework will fall back to simulated acquisition automatically.
+This installs `asyncroscopy` and all core dependencies. The vendor wheels are
+local, version-pinned, and resolved via `[tool.uv.sources]` in `pyproject.toml`
+(AutoScript under `stubs/AutoScript_v_1.17/`, PyJEM under
+`stubs/PyJEM_v_1.3.0.3564/`). AutoScript/PyJEM hardware is not required—the
+framework falls back to simulated acquisition automatically.
 
 ### Hardware installation (Thermo Fisher AutoScript)
 

asyncroscopy/data/data_writer.py

@@ -9,6 +9,7 @@
 
 import h5py
 import numpy as np
+import tifffile
 
 DEFAULT_ACQUISITION_DIR = "outputs/tiled_acquisitions"
 
@@ -35,6 +36,21 @@ def acquisition_filename(
     return directory / f"{acquisition_type}_{detector}_{stamp}.{extension.lower().lstrip('.')}"
 
 
+def _write_tiff(source, path: Path, attrs: dict | None = None) -> None:
+    """Write one image to ``path`` as TIFF.
+
+    AutoScript adorned images save themselves natively (embedding their own
+    metadata); anything else is a raw array (JEOL/twin) written via tifffile,
+    with ``attrs`` (e.g. JEOL's get_detectorsetting() dict) json-encoded into the
+    TIFF ImageDescription tag so the provenance survives the format.
+    """
+    if hasattr(source, "save"): # adorned images from AutoScript have a save method
+        source.save(str(path))
+    else:
+        array = source.data if hasattr(source, "data") and not isinstance(source, np.ndarray) else source
+        tifffile.imwrite(str(path), np.asarray(array), description=json.dumps(attrs) if attrs else None)
+
+
 def save_acquisition(
     device,
     data_server,
@@ -44,11 +60,31 @@ def save_acquisition(
     dataset_name: str = "image",
     dataset_attrs: dict | list[dict] | None = None,
     file_attrs: dict | None = None,
+    output_format: str = ".h5",
 ) -> str:
-    """Save one acquisition to one HDF5 file and return its DATA/Tiled key."""
+    """Save one acquisition and return its DATA/Tiled key.
+
+    ``.h5`` writes all detectors into one stacked file; ``.tiff`` writes one
+    file per detector sharing a timestamp (TIFF cannot stack), registering each.
+    """
+    if output_format not in (".h5", ".tiff"):
+        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]
     attrs_list = dataset_attrs if isinstance(dataset_attrs, list) else [dataset_attrs] * len(data_list)
+
+    if output_format == ".tiff":
+        save_dir = data_server.save_path if data_server is not None else DEFAULT_ACQUISITION_DIR
+        directory = Path(save_dir).expanduser()
+        directory.mkdir(parents=True, exist_ok=True)
+        stem = f"{acquisition_type}_{datetime.now().strftime('%Y%m%dT%H%M%S%f')}"
+        for index, (source, detector) in enumerate(zip(data_list, detector_list)):
+            path = directory / f"{stem}_{detector}.tiff"
+            _write_tiff(source, path, attrs_list[index])
+            if data_server is not None:
+                data_server.register_path(str(path))
+        return stem
+
     has_labeled_datasets = len(detector_list) > 1 or len(data_list) > 1
     detector_label = "_".join([str(detector) for detector in detector_list])
     path = acquisition_filename(device, acquisition_type, detector_label, data_server)
@@ -98,3 +134,6 @@ def save_acquisition_hdf5(path: str | Path, datasets: list[dict], file_attrs: di
                         if key in dset.attrs:
                             key = f"{key}_{len(dset.attrs)}"
                         dset.attrs[key] = elem.text.strip()
+            elif isinstance(metadata, dict):
+                for key, value in metadata.items():
+                    dset.attrs[key] = value if isinstance(value, (str, int, float, bool, np.number)) else json.dumps(value)

asyncroscopy/instruments/electron_microscope/auto_script.py

@@ -18,8 +18,6 @@
 
 import math
 import time
-from datetime import datetime
-from pathlib import Path
 
 import numpy as np
 import tango
@@ -250,63 +248,35 @@ def register_stage(self):
     # ------------------------------------------------------------------
     # Internal acquisition helpers
     # ------------------------------------------------------------------
-    def _persist(self, adorned, acquisition_type, detector, data_server, dataset_name="image"):
-        """Save acquired images in the format requested by the SCAN device.
-        """
-        scan = self._detector_proxies.get("scan")
-        fmt = scan.output_format if scan is not None else ".h5"  # ".h5" default
-        if fmt == ".h5":
-            return save_acquisition(self, data_server, acquisition_type, detector, adorned, dataset_name=dataset_name)
-        if fmt != ".tiff":
-            raise ValueError(f"Unsupported output_format {fmt!r}; expected '.h5' or '.tiff'")
-
-        # .tiff → AutoScript native save, one file per detector sharing one stamp
-        images = list(adorned) if isinstance(adorned, (list, tuple)) else [adorned]
-        detectors = list(detector) if isinstance(detector, (list, tuple)) else [detector]
-        if len(images) != len(detectors):
-            raise ValueError(f"Got {len(images)} images for {len(detectors)} detector(s) {detectors}")
-
-        save_dir = data_server.save_path if data_server is not None else DEFAULT_ACQUISITION_DIR
-        directory = Path(save_dir).expanduser()
-        directory.mkdir(parents=True, exist_ok=True)
-        stamp = datetime.now().strftime("%Y%m%dT%H%M%S%f")
-        stem = f"{acquisition_type}_{stamp}"
-        # AutoScript returns images in the requested detector order (assumed; verify on hardware)
-        for img, det in zip(images, detectors):
-            path = directory / f"{stem}_{det}.tiff"
-            img.save(str(path))
-            if data_server is not None:
-                data_server.register_path(str(path))
-        return stem
-
     def _acquire_scanned_image(
         self,
         imsize: int,
         dwell_time: float,
         detector_list: list[str] = ["haadf"],
         scan_region: list[float] = [0.0, 0.0, 1.0, 1.0],
+        output_format: str = ".h5",
     ) -> str:
         """
-        Call AutoScript scanned image acquisition, save one HDF5 file, and return its DATA/Tiled key.
+        Call AutoScript scanned image acquisition, save it, and return its DATA/Tiled key.
         """
         detector_list = [d.upper() for d in detector_list]
         settings = StemAcquisitionSettings(dwell_time=dwell_time, detector_types=detector_list, size=imsize, region=Region(RegionCoordinateSystem.RELATIVE, Rectangle(*scan_region)))
         adorned = self._microscope.acquisition.acquire_stem_images_advanced(settings)
         if not isinstance(adorned, list):
             adorned = [adorned]
         data_server = self._detector_proxies.get("data")
-        return self._persist(adorned, "stem_image", detector_list, data_server)
+        return save_acquisition(self, data_server, "stem_image", detector_list, adorned, output_format=output_format)
 
 
     def _acquire_camera_image(self, imsize: int, exposure_time: float, detector: str, readout_area: str) -> str:
         """
-        Call AutoScript acquisition, save the adorned image, and return its path.
+        Call AutoScript acquisition, save the adorned image, and return its DATA/Tiled key.
         this is the advanced version
         """
         settings = CameraAcquisitionSettings(camera_detector=detector, size=imsize, exposure_time=exposure_time, fixed_readout_area=readout_area, frame_combining=1)
         adorned = self._microscope.acquisition.acquire_camera_image_advanced(settings)
         data_server = self._detector_proxies.get("data")
-        return self._persist(adorned, "camera_image", str(detector), data_server)
+        return save_acquisition(self, data_server, "camera_image", str(detector), adorned)
 
     def _acquire_scanned_data_advanced(self, imsize: int, dwell_time: float, detector: str, scan_region: list[float]) -> str:
         """

asyncroscopy/instruments/electron_microscope/digital_twin.py

@@ -518,15 +518,16 @@ def _acquire_scanned_image(
         dwell_time: float,
         detector_list: list[str] = ["haadf"],
         scan_region: list[float] = [0.0, 0.0, 1.0, 1.0],
+        output_format: str = ".h5",
     ) -> str:
-        """Simulate STEM acquisition, save HDF5 data with metadata, and return its DATA/Tiled key."""
+        """Simulate STEM acquisition, save the data with metadata, and return its DATA/Tiled key."""
         detector_list = [detector.upper() for detector in detector_list]
         data_server = self._detector_proxies.get("data")
         images = []
         for detector in detector_list:
             image = self._render_stem_image(int(imsize), float(dwell_time), [detector])
             images.append(image)
-        return save_acquisition(self, data_server, "stem_image", detector_list, images)
+        return save_acquisition(self, data_server, "stem_image", detector_list, images, output_format=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."""

asyncroscopy/instruments/electron_microscope/digital_twin_diffraction.py

@@ -356,6 +356,7 @@ def _acquire_scanned_image(
         dwell_time: float,
         detector_list: list[str] = ['haadf'],
         scan_region: list[float] = [0.0, 0.0, 1.0, 1.0],
+        output_format: str = '.h5',
     ) -> str:
         image_size = int(self.overview_image_size)
         detector_list = [detector.upper() for detector in detector_list]
@@ -369,7 +370,7 @@ def _acquire_scanned_image(
             'sample_pixel_size_nm': float(FOV_NM / int(self.map_size)),
         }
         images = [self._render_stem_image(image_size, float(dwell_time), [detector]) for detector in detector_list]
-        return save_acquisition(self, data_server, 'stem_image', detector_list, images, dataset_attrs=attrs)
+        return save_acquisition(self, data_server, 'stem_image', detector_list, images, dataset_attrs=attrs, output_format=output_format)
 
     def _set_fov(self, fov) -> None:
         self.warn_stream('DigitalTwinDiffraction uses a fixed 500 nm field of view.')

asyncroscopy/instruments/electron_microscope/electron_microscope.py

@@ -123,7 +123,7 @@ def acquire_spectrum(self, detector_name: str) -> str:
     def acquire_scanned_image(self, detector_list: list[str] = ['haadf']) -> str:
         """Acquire an image with scanning detectors and return its DATA/Tiled key."""
         scan = self._detector_proxies.get('scan')
-        return self._acquire_scanned_image(scan.imsize, scan.dwell_time, detector_list, list(scan.scan_region))
+        return self._acquire_scanned_image(scan.imsize, scan.dwell_time, detector_list, list(scan.scan_region), scan.output_format)
 
     @command(dtype_out=str)
     def acquire_scanned_data_advanced(self) -> str:
@@ -245,6 +245,7 @@ def _acquire_scanned_image(
         dwell_time: float,
         detector_list: list[str] = ['haadf'],
         scan_region: list[float] = [0.0, 0.0, 1.0, 1.0],
+        output_format: str = '.h5',
     ) -> str:
         """Vendor-specific scanned image acquisition implementation."""
         pass