[feature][MSD-506] Meteor store annoted data

src/odemis/acq/feature.py

@@ -21,7 +21,6 @@
 """
 
 import copy
-import glob
 import logging
 import os
 import threading
@@ -45,12 +44,13 @@
 from odemis.acq.milling import DEFAULT_MILLING_TASKS_PATH
 from odemis.acq.move import Posture, MicroscopePostureManager
 from odemis.acq.stitching._tiledacq import SAFE_REL_RANGE_DEFAULT
-from odemis.acq.stream import Stream, StaticFluoStream
+from odemis.acq.stream import Stream, StaticFluoStream, StaticSEMStream, StaticFIBStream
 from odemis.dataio import find_fittest_converter
 from odemis.gui.cont.cryo_project import IMG_FILENAME, IMG_IN_FILE_IDS, add_image
 from odemis.model import MD_IN_FILE_INDEX
 from odemis.util import dataio, executeAsyncTask
 from odemis.util.comp import generate_zlevels
+from odemis.util import datacollector
 from odemis.util.dataio import data_to_static_streams, open_acquisition, splitext
 from odemis.util.driver import estimate_stage_movement_time
 from odemis.util.filename import create_filename
@@ -168,13 +168,18 @@ class CryoFeature(object):
     def __init__(self, name: str,
                  stage_position: Dict[str, float],
                  fm_focus_position: Dict[str, float],
-                 milling_tasks: Optional[Dict[str, MillingTaskSettings]] = None, correlation_data=None):
+                 milling_tasks: Optional[Dict[str, MillingTaskSettings]] = None,
+                 correlation_data=None,
+                 collect: bool = False):
         """
         :param name: (string) the feature name
         :param stage_position: (dict) the stage position of the feature (stage-bare)
         :param fm_focus_position: (dict) the focus position of the feature
         :param correlation_data: (Dict[str,FIBFMCorrelationData]) Dictionary mapping the feature status to
         FIBFMCorrelationData, where feature status like Active, Rough Milled or polished is the key.
+        :param collect: (bool) Whether this feature is eligible for data collection.
+            Defaults to False. The GUI sets this based on the per-project sampling
+            decision made when a project is opened or created.
         """
         self.name = model.StringVA(name)
         # FIXME: The 'position' parameter should eventually contain the SampleStage coordinates and not stage bare from the stage_position!
@@ -205,6 +210,9 @@ def __init__(self, name: str,
             correlation_data = {}
         self.correlation_data = correlation_data
 
+        # Whether this feature is eligible for data collection
+        self.collect: bool = collect
+
         # attributes for automated milling
         self.path: str = None  # TODO:support path creation here, rather than on milling data save
         self.reference_image: model.DataArray = None
@@ -292,9 +300,11 @@ def feature_decoder(feature_raw: Dict) -> CryoFeature:
     fm_focus_position = feature_raw['fm_focus_position']
     posture_positions = feature_raw.get('posture_positions', {})
     milling_task_json = feature_raw.get('milling_tasks', {})
+    collect = feature_raw.get('collect', False)
     feature = CryoFeature(name=feature_raw['name'],
                           stage_position=stage_position,
-                          fm_focus_position=fm_focus_position
+                          fm_focus_position=fm_focus_position,
+                          collect=collect
                           )
     feature.correlation_data = FIBFMCorrelationData.from_dict(correlation_data) if correlation_data else None
     feature.status.value = feature_raw['status']
@@ -396,6 +406,159 @@ def _create_fibsem_filename(filename: str, acq_type: str) -> str:
 
     return create_filename(path, ptn, ext, count="001")
 
+def _stream_overlaps_position(stream: Stream, x: float, y: float) -> bool:
+    """
+    True if the stage position (x, y) falls within the stream's field of view.
+
+    :param stream: A static stream with a getBoundingBox() method.
+    :param x: Stage x position in metres.
+    :param y: Stage y position in metres.
+    :return: True when the position is inside the bounding box, False otherwise.
+    """
+    try:
+        bbox = stream.getBoundingBox()  # (left, top, right, bottom) in metres
+    except Exception:
+        return False
+    left, top, right, bottom = bbox
+    return left <= x <= right and top <= y <= bottom
+
+
+def collect_feature_data(
+    feature: CryoFeature,
+    overview_streams: Optional[List[Stream]] = None,
+    project_dir: Optional[str] = None) -> None:
+    """
+    Collect anonymized data for a feature and submit it to the data collector. It skips immediately if feature.collect
+    is False or if data collection consent has not been granted.  Never raises — all errors are logged as warnings.
+
+    The payload contains:
+    - Acquired z-stack per FM channel (or first FM image if no z-stack) before moving to SEM/FIB
+    - FM and SEM overview images that spatially overlap the feature's position.
+    - Feature status, stage position, and FM focus position.
+
+    Privacy rules enforced:
+    - Feature name is never included.
+    - Image payload keys are generic (channel_0, overview_fm_0, etc.).
+    - Original filenames are not included in the payload.
+
+    After collection feature.collect is set to False to prevent re-collection.
+
+    :param feature: The feature for which the data is collected. Must have feature.collect == True.
+    :param overview_streams: Optional list of overview static streams. Used to find FM / SEM overviews
+     that overlap the feature position.
+    :param project_dir: Optional project directory path. When provided and
+     feature.streams is empty, streams are loaded from disk first.
+    """
+    if not feature.collect:
+        return
+
+    try:
+        _dc = datacollector.DataCollector()
+        if not _dc.get_consent():
+            return
+    except Exception:
+        logging.warning("Failed to access DataCollector; skipping collection", exc_info=True)
+        return
+
+    try:
+        # Load feature streams from disk when not yet in memory.
+        if not feature.streams.value and project_dir:
+            try:
+                load_feature_streams_from_disk(feature, project_dir)
+            except Exception:
+                logging.warning("Failed to load streams from disk; skipping collection", exc_info=True)
+                return
+
+        feature_streams = list(feature.streams.value)
+
+        # Prioritize collecting z stack as it is more insightful than a single image
+        fm_zstacks: List = []
+        fm_images: List = []
+        for s in feature_streams:
+            if isinstance(s, StaticFluoStream):
+                if hasattr(s, "zIndex"):
+                    fm_zstacks.append(s)
+                else:
+                    fm_images.append(s)
+
+        def _get_raw(stream: Stream) -> Optional[model.DataArray]:
+            """Return image data for a static stream from raw or image VA."""
+            try:
+                if stream.raw:
+                    return stream.raw[0]
+            except Exception:
+                pass
+            try:
+                image_va = getattr(stream, "image", None)
+                if image_va is not None:
+                    return image_va.value
+            except Exception:
+                pass
+            return None
+
+        # Per channel: prefer z-stack, then single FM image.
+        # Channels are delineated by MD_OUT_WL; use index as fallback.
+        selected_fm: List = []
+        seen_channels: set = set()
+        for s in fm_zstacks + fm_images:
+            da = _get_raw(s)
+            channel_key = da.metadata.get(model.MD_OUT_WL) if da is not None else None
+            if channel_key not in seen_channels:
+                seen_channels.add(channel_key)
+                selected_fm.append(s)
+
+        # Collect spatially overlapping overview streams.
+        stage_pos = feature.stage_position.value
+        feat_x = stage_pos.get("x", 0.0)
+        feat_y = stage_pos.get("y", 0.0)
+
+        overview_fm: List = []
+        overview_sem: List = []
+        for s in (overview_streams or []):
+            if not _stream_overlaps_position(s, feat_x, feat_y):
+                continue
+            if isinstance(s, StaticFluoStream):
+                overview_fm.append(s)
+            elif isinstance(s, (StaticSEMStream, StaticFIBStream)):
+                overview_sem.append(s)
+
+        # Build privacy-preserving payload — generic keys, no names or filenames.
+        payload: dict = {
+            "status": feature.status.value,
+            "stage_position": dict(stage_pos),
+            "fm_focus_position": dict(feature.fm_focus_position.value),
+        }
+
+
+        for idx, s in enumerate(selected_fm):
+            da = _get_raw(s)
+            if da is not None:
+                payload[f"channel_{idx}"] = da
+
+        for idx, s in enumerate(overview_fm):
+            da = _get_raw(s)
+            if da is not None:
+                payload[f"overview_fm_{idx}"] = da
+
+        for idx, s in enumerate(overview_sem):
+            da = _get_raw(s)
+            if da is not None:
+                payload[f"overview_sem_{idx}"] = da
+
+        image_keys = [k for k in payload if k.startswith(("channel_", "overview_fm_", "overview_sem_"))]
+        if not image_keys:
+            logging.debug( "No images found for the given feature; skipping collection")
+            return
+
+        _dc.record("feature_collected", "1.0", payload)
+
+        feature.collect = False
+        logging.debug("Data collected successfully")
+
+    except Exception:
+        logging.warning("Failed to collect data; skipping collection", exc_info=True)
+
+
 # To handle the timeout error when the stage is not able to move to the desired position
 # It logs the message and raises the MoveError exception
 class MoveError(Exception):