Controllers

optari.controllers.base

Base class for task-specific controllers.

TaskControllerBase

TaskControllerBase(parent_controller)

Minimal interface for task controllers.

Each task controller (Analysis, Unmixing, etc) inherits from this.

A controller that runs heavy work implements three methods and gets the dock's Run button (run_from_ui) for free:

• _params_from_ui() reads the dock into a params object, raising ValueError with a user-facing message when the run cannot be set up
• prepare(params) validates and returns a BackgroundStep, raising ValueError
• publish(result, params) adds the layers and returns the text shown after "Finished:"

Batch mode builds params from a preset instead and calls the same prepare and publish, which is what keeps its numbers equal to the GUI's.

Initialize with reference to parent controller.

Parameters:

Name	Type	Description	Default
parent_controller	OptariController	OptariController instance with viewer and session state.	required

Source code in src/optari/controllers/base.py

def __init__(self, parent_controller: OptariController):
    """Initialize with reference to parent controller.

    Args:
        parent_controller: OptariController instance with viewer and session state.
    """
    self.optari_controller = parent_controller
    self.viewer = parent_controller.viewer

bind_events

bind_events()

Connect this controller's dock signals to their handlers.

Source code in src/optari/controllers/base.py

def bind_events(self) -> None:
    """Connect this controller's dock signals to their handlers."""
    for signal, handler in self._signal_bindings():
        signal.connect(handler)

initialize_ui

initialize_ui()

Initialize UI controls (populate lists, combos, etc). Override if needed.

Source code in src/optari/controllers/base.py

def initialize_ui(self) -> None:
    """Initialize UI controls (populate lists, combos, etc). Override if needed."""
    pass

refresh_ui

refresh_ui()

Refresh controller-owned UI after application state changes.

Source code in src/optari/controllers/base.py

def refresh_ui(self) -> None:
    """Refresh controller-owned UI after application state changes."""
    pass

run_from_ui

run_from_ui(dock)

The dock's Run button: params from widgets, fetch weights if needed, run, publish.

Source code in src/optari/controllers/base.py

def run_from_ui(self, dock) -> None:
    """The dock's Run button: params from widgets, fetch weights if needed, run, publish."""
    try:
        params = self._params_from_ui()
    except ValueError as exc:
        dock.status_label.setText(str(exc))
        return

    def run() -> None:
        try:
            step = self.prepare(params)
        except ValueError as exc:
            dock.status_label.setText(str(exc))
            return
        dock.status_label.setText(f"{step.desc}…")
        start_task(
            self.optari_controller,
            step,
            on_result=lambda result: dock.status_label.setText(
                f"Finished: {self.publish(result, params)}"
            ),
            status_label=dock.status_label,
        )

    weights = self.weights_to_fetch(params)
    if weights is None:
        run()
        return
    download_then(self.optari_controller, *weights, dock.status_label, run)

teardown

teardown()

Release resources (models, threads). Override if needed.

Source code in src/optari/controllers/base.py

def teardown(self) -> None:
    """Release resources (models, threads). Override if needed."""
    pass

unbind_events

unbind_events()

Disconnect everything bind_events connected.

Source code in src/optari/controllers/base.py

def unbind_events(self) -> None:
    """Disconnect everything bind_events connected."""
    for signal, handler in self._signal_bindings():
        signal.disconnect(handler)

weights_to_fetch

weights_to_fetch(params)

(local path, download url) a run needs on disk before it can start, or None.

Source code in src/optari/controllers/base.py

def weights_to_fetch(self, params) -> tuple[Path, str | None] | None:
    """(local path, download url) a run needs on disk before it can start, or None."""
    return None

optari.controllers.scan_controller

Scan and study discovery, and loading a scan into the viewer.

ScanController

ScanController(parent_controller)

Bases: TaskControllerBase

Scan/session lifecycle and data-loading helpers for OPTARI.

Initialize with reference to parent controller.

Parameters:

Name	Type	Description	Default
parent_controller	OptariController	OptariController instance with viewer and session state.	required

Source code in src/optari/controllers/scan_controller.py

def __init__(self, parent_controller: OptariController):
    """Initialize with reference to parent controller.

    Args:
        parent_controller: OptariController instance with viewer and session state.
    """
    super().__init__(parent_controller)
    # The Scan Browser's study. A batch run loads scans from elsewhere.
    self.scans: dict[Path, ScanInfo] = {}
    self._export_view_handler = (
        lambda: ViewerExportController.on_export_clicked(
            self.optari_controller
        )
    )

close_current_scan

close_current_scan()

Close the HDF5 handle for the current scan.

Source code in src/optari/controllers/scan_controller.py

def close_current_scan(self) -> None:
    """Close the HDF5 handle for the current scan."""
    if self.optari_controller.pa_data is not None:
        try:
            self.optari_controller.pa_data.close()
        except Exception:
            logger.info(
                "failed to close current scan handle", exc_info=True
            )
    self.optari_controller.pa_data = None
    self.optari_controller.scan_info = None
    self.optari_controller.patato_objects = {}
    self.optari_controller.derived_patato_objects = {}
    self.optari_controller.clinical_metadata_edits = None

get_fov

get_fov()

Return (fov_x_m, fov_y_m) from stored PATATO objects, or None.

Source code in src/optari/controllers/scan_controller.py

def get_fov(self) -> "tuple[float, float] | None":
    """Return ``(fov_x_m, fov_y_m)`` from stored PATATO objects, or ``None``."""
    return fov_from_objects(self.optari_controller.patato_objects)

go_to_frame

go_to_frame(selector)

Show the frame selector means for the loaded scan, and return it.

"spectral_motion" and "frame_motion" are the frame with the lowest score of that kind in motion_scores(). A scan without spectral motion (one image per wav set) uses frame-to-frame motion instead. A scan with no ultrasound (a raw time series) has nothing to score and falls back to frame 0. "acquisition_start" is the ithera recording start frame "roi" is the first frame holding one of the scan's ROIs, frame 0 without any. It needs the scan's ROIs restored first. Both the viewer and a batch run go through this, so a selector means the same frame in either.

Raises ValueError for a frame number the scan does not have.

Source code in src/optari/controllers/scan_controller.py

def go_to_frame(self, selector: int | str) -> int:
    """Show the frame *selector* means for the loaded scan, and return it.

    ``"spectral_motion"`` and ``"frame_motion"`` are the frame with the lowest score
    of that kind in ``motion_scores()``. A scan without spectral motion (one image per
    wav set) uses frame-to-frame motion instead. A scan with no ultrasound (a raw time
    series) has nothing to score and falls back to frame 0.
    ``"acquisition_start"`` is the ithera recording start frame
    ``"roi"`` is the first frame holding one of the scan's ROIs, frame 0 without any.
    It needs the scan's ROIs restored first.
    Both the viewer and a batch run go through this, so a selector means the same
    frame in either.

    Raises ValueError for a frame number the scan does not have.
    """
    if selector in ("spectral_motion", "frame_motion"):
        frame_id = self._lowest_motion_frame(selector)
    elif selector == "acquisition_start":
        frame_id = self._acquisition_start_frame()
    elif selector == "roi":
        frame_id = self._first_roi_frame()
    else:
        frame_id = int(selector)
        n_frames = (
            int(self.viewer.dims.nsteps[0]) if self.viewer.dims.ndim else 0
        )
        if not 0 <= frame_id < n_frames:
            raise ValueError(
                f"frame {frame_id} is outside this scan ({n_frames} frames)"
            )
    self.viewer.dims.set_point(0, frame_id)
    return frame_id

init_shapes_from_scan

init_shapes_from_scan()

Fill the freshly created ROIs layer with the ROIs stored in the scan.

Source code in src/optari/controllers/scan_controller.py

def init_shapes_from_scan(self) -> None:
    """Fill the freshly created ROIs layer with the ROIs stored in the scan."""
    roi_ctrl = self.optari_controller.roi_ctrl
    fov = self.get_fov()
    if fov is not None:
        try:
            # Projects the current frame into the layer, with ids and properties.
            roi_ctrl.restore_roi_records(
                roi_records_from_scan_rois(
                    self.optari_controller.pa_data, *fov
                )
            )
            logger.info(
                "restored %s stored ROI record(s)",
                len(roi_ctrl.roi_records),
            )
        except ValueError as exc:
            logger.warning(
                "could not restore stored ROIs from %s: %s",
                self.optari_controller.path,
                exc,
                exc_info=True,
            )
            show_warning(
                f"Scan loaded without saved ROI annotations: {exc}"
            )

    # Select the restored ROIs, which also enables the Save button.
    shapes_layer = self.optari_controller.shapes_layer
    shapes_layer.selected_data = set(range(len(shapes_layer.data)))
    roi_ctrl.refresh_ui()

layers_from_pa_data

layers_from_pa_data()

Build napari LayerData tuples from the open controller.pa_data handle.

Source code in src/optari/controllers/scan_controller.py

def layers_from_pa_data(self) -> list[tuple]:
    """Build napari LayerData tuples from the open ``controller.pa_data`` handle."""
    layers, self.optari_controller.patato_objects = build_napari_layers(
        self.optari_controller.pa_data
    )
    return layers

load_scan

load_scan(scan_path, scan_info)

Open scan_path and build its layers. False if the scan could not be loaded.

Failures fall back to the startup logo rather than raising, which is what the GUI wants. The return value is what lets an unattended caller (batch mode) tell a loaded scan from an empty viewer. scan_info comes from discovery, so the Scan Browser's list and a batch plan can each load their own scans.

Source code in src/optari/controllers/scan_controller.py

def load_scan(self, scan_path: Path, scan_info: ScanInfo) -> bool:
    """Open *scan_path* and build its layers. False if the scan could not be loaded.

    Failures fall back to the startup logo rather than raising, which is what the
    GUI wants. The return value is what lets an unattended caller (batch mode) tell
    a loaded scan from an empty viewer. *scan_info* comes from discovery, so the
    Scan Browser's list and a batch plan can each load their own scans.
    """
    logger.info("loading scan: %s", scan_path)

    self.optari_controller.path = scan_path
    self.reset_scan_state(restore_startup_logo=False)

    if not scan_path.exists():
        logger.warning("scan not found: %s", scan_path)
        load_startup_logo(self.viewer)
        self.optari_controller.refresh_all()
        return False

    try:
        if scan_info.kind == "ithera":
            self.optari_controller.pa_data = pat.PAData(
                iTheraMSOT(str(scan_path))
            )
        else:
            # PATATO's reader factory tells the PATATO and IPASC layouts apart itself.
            self.optari_controller.pa_data = pat.PAData.from_hdf5(
                str(scan_path), mode="r"
            )
    except Exception:
        logger.exception("failed to open scan '%s'", scan_path)
        load_startup_logo(self.viewer)
        self.optari_controller.refresh_all()
        return False
    self.optari_controller.scan_info = scan_info

    try:
        layers = self.layers_from_pa_data()
    except Exception:
        logger.exception("failed to load '%s'", scan_path)
        self.close_current_scan()
        load_startup_logo(self.viewer)
        self.optari_controller.refresh_all()
        return False

    for data, kw in layers:
        kw["metadata"] = {
            "filepath": str(scan_path),
            "scan_name": scan_info.internal_name,
            **kw["metadata"],
        }
        self.viewer.add_image(
            data, units=self.image_units, **kw
        ).colorbar.visible = True

    # Create the ROIs layer after image layers so it stays on top.
    self.optari_controller.shapes_layer = self.viewer.add_shapes(
        name="ROIs",
        edge_color=roi_color_for_index(0),
        face_color="transparent",
        edge_width=0.1,
        ndim=2,
        metadata={"type": "roi"},
        units=("mm", "mm"),
    )
    self.optari_controller.shapes_layer.locked = True
    self.optari_controller.connect_shapes_layer_events()

    # After adding layers, pick a sensible default selected layer.
    try:
        self.optari_controller.select_default_pa_layer()
    except RuntimeError as e:
        logger.info("Default PA layer not found", exc_info=True)
        show_warning(str(e))

    # Set the active US layer if available.
    self.optari_controller.active_us_layer = next(
        (
            l
            for l in self.viewer.layers
            if isinstance(l, Image) and l.metadata.get("type") == "us"
        ),
        None,
    )
    self.optari_controller.resolve_active_recon_layer()
    self.optari_controller.segmentation_ctrl.restore_from_scan(
        self.optari_controller.pa_data
    )
    self.optari_controller.refresh_controller_uis()

    # Initialize viewer position to DEFAULT_FRAME_INDEX and DEFAULT_CHANNEL_INDEX
    frame_selector = settings.general.DEFAULT_FRAME_INDEX
    try:
        # "roi" needs the scan's ROIs, which are restored below
        self.go_to_frame(0 if frame_selector == "roi" else frame_selector)
    except ValueError as exc:  # a configured frame this scan does not have
        logger.warning("%s, starting at frame 0 instead", exc)
        self.viewer.dims.set_point(0, 0)
    self.viewer.dims.set_point(1, settings.general.DEFAULT_CHANNEL_INDEX)

    # Populate ROIs after dims are initialized to avoid computing stats before the viewer is ready.
    self.init_shapes_from_scan()
    if frame_selector == "roi":
        self.go_to_frame("roi")
    # Free when a motion-based default frame selection already scored the scan.
    self.optari_controller.analysis_ctrl.show_motion_scores(compute=False)

    # Fit view to the newly loaded data (prevents "zoomed out" state).
    self.viewer.reset_view()

    return True

motion_scores

motion_scores(*, compute=True)

Motion scores per frame of the loaded scan (0 = stillest, 1 = most motion), keyed "spectral_motion" and "frame_motion", or None without ultrasound.

Spectral motion is None for a scan with one image per wav set. Both kinds are computed together on first use and cached on the US layer. With compute=False only cached scores are returned, which lets a caller show them without paying for scoring.

Source code in src/optari/controllers/scan_controller.py

def motion_scores(
    self, *, compute: bool = True
) -> dict[str, np.ndarray | None] | None:
    """Motion scores per frame of the loaded scan (0 = stillest, 1 = most motion),
    keyed ``"spectral_motion"`` and ``"frame_motion"``, or None without ultrasound.

    Spectral motion is None for a scan with one image per wav set. Both kinds are
    computed together on first use and cached on the US layer.
    With ``compute=False`` only cached scores are returned, which lets a
    caller show them without paying for scoring.
    """
    us_layer = self.optari_controller.active_us_layer
    if us_layer is None:
        return None
    if "motion_scores" not in us_layer.metadata:
        if not compute:
            return None
        us_data = np.asarray(us_layer.data)
        us_layer.metadata["motion_scores"] = {
            "spectral_motion": spectral_motion_scores(us_data),
            "frame_motion": frame_motion_scores(us_data),
        }
    return us_layer.metadata["motion_scores"]

on_browse_study_clicked

on_browse_study_clicked()

Prompt for a study folder, the folder that contains the scans.

Source code in src/optari/controllers/scan_controller.py

def on_browse_study_clicked(self) -> None:
    """Prompt for a study folder, the folder that contains the scans."""
    folder = QFileDialog.getExistingDirectory(
        None, "Select study folder", str(Path.cwd())
    )
    if folder:
        self.set_scan_folder(Path(folder))

on_hdf5_export_clicked

on_hdf5_export_clicked(event=None)

Prompt for a destination and export the current scan to HDF5.

Source code in src/optari/controllers/scan_controller.py

def on_hdf5_export_clicked(self, event=None) -> None:
    """Prompt for a destination and export the current scan to HDF5."""
    destination = self._choose_export_path()
    if destination is None:
        return

    if export_scan_to_hdf5(self.optari_controller, destination):
        show_info(f"Exported scan to {destination.name}")

on_ipasc_export_clicked

on_ipasc_export_clicked(event=None)

Prompt for a destination and export the current scan's raw time series as IPASC.

Source code in src/optari/controllers/scan_controller.py

def on_ipasc_export_clicked(self, event=None) -> None:
    """Prompt for a destination and export the current scan's raw time series as IPASC."""
    destination = self._choose_export_path(
        title="Export raw time series as IPASC", suffix="_ipasc"
    )
    if destination is None:
        return

    if not export_scan_to_ipasc(self.optari_controller, destination):
        return
    show_info(
        f"Exported raw time series to {destination.name}. "
        + ipasc_export_report(destination)
    )

on_scan_selected

on_scan_selected(row)

Load the scan at row in the scan list, ignoring an out-of-range selection.

Source code in src/optari/controllers/scan_controller.py

def on_scan_selected(self, row: int) -> None:
    """Load the scan at *row* in the scan list, ignoring an out-of-range selection."""
    scans = list(self.scans.items())
    if row < 0 or row >= len(scans):
        return
    self.load_scan(*scans[row])

refresh_ui

refresh_ui()

Lock the scan browser while a task runs, and gate HDF5 export on a scan being loaded.

Source code in src/optari/controllers/scan_controller.py

def refresh_ui(self) -> None:
    """Lock the scan browser while a task runs, and gate HDF5 export on a scan being loaded."""
    self.optari_controller.scan_browser.widget.setEnabled(
        not self.optari_controller.task_running
    )
    self.optari_controller.scan_browser.hdf5_button.setEnabled(
        self.optari_controller.pa_data is not None
    )
    self.optari_controller.scan_browser.ipasc_button.setEnabled(
        self.optari_controller.pa_data is not None
    )

reset_scan_state

reset_scan_state(restore_startup_logo=True)

Clear current scan state and remove all viewer layers.

Source code in src/optari/controllers/scan_controller.py

def reset_scan_state(self, restore_startup_logo: bool = True) -> None:
    """Clear current scan state and remove all viewer layers."""
    self.close_current_scan()
    self.optari_controller.segmentation_ctrl.teardown()
    self.optari_controller.active_recon_layer = None
    self.optari_controller.active_us_layer = None
    self.optari_controller.shapes_layer = None
    self.optari_controller.roi_ctrl.clear_roi_records()
    self.optari_controller._last_frame_idx = None

    for layer in list(self.viewer.layers):
        self.viewer.layers.remove(layer)
    self.optari_controller.refresh_controller_uis()
    self.optari_controller.analysis_ctrl.show_motion_scores(compute=False)
    if restore_startup_logo:
        load_startup_logo(self.viewer)

scan_name

scan_name()

The scan's internal (vendor) name, for stamping onto layers OPTARI creates.

Source code in src/optari/controllers/scan_controller.py

def scan_name(self) -> "str | None":
    """The scan's internal (vendor) name, for stamping onto layers OPTARI creates."""
    return self.optari_controller.scan_info.internal_name

set_scan_folder

set_scan_folder(folder)

Discover scans in folder and select the first one.

Source code in src/optari/controllers/scan_controller.py

def set_scan_folder(self, folder: Path) -> None:
    """Discover scans in *folder* and select the first one."""
    self.scans = discover_scans(folder)

    self.optari_controller.scan_browser.set_folder(folder)
    self.optari_controller.scan_browser.set_scans(list(self.scans.items()))

    if self.scans:
        self.optari_controller.scan_browser.scans_list.setCurrentRow(0)
    else:
        self.reset_scan_state()

optari.controllers.roi_controller

Owns the ROI shapes layer, the Live and Saved Analysis tables, and ROI presets and placement.

RoiController

RoiController(parent_controller)

Bases: TaskControllerBase

ROI visualization, table management, and ROI export helpers.

Initialize with reference to parent controller.

Sets up the ROI record store, the session's autosaved Saved Analysis table, and the ROI preset store.

Parameters:

Name	Type	Description	Default
parent_controller	OptariController	OptariController instance with viewer and session state.	required

Source code in src/optari/controllers/roi_controller.py

def __init__(self, parent_controller: OptariController):
    """Initialize with reference to parent controller.

    Sets up the ROI record store, the session's autosaved Saved Analysis table,
    and the ROI preset store.

    Args:
        parent_controller: OptariController instance with viewer and session state.
    """
    super().__init__(parent_controller)
    self.saved_table = SavedRoiTable(
        new_session_file(get_user_autosave_dir(), "roi_table", ".xlsx")
    )
    self._syncing = False
    self._roi_records: dict[int, ROIRecord] = {}
    self._projection_ids: list[int] = []
    self._next_roi_id = 0
    self._next_track_id = 0
    self._projecting = False
    # Whether the ROIs stored in the scan are in the session. Export only replaces
    # stored OPTARI ROIs if they are, so a failed restore cannot drop them.
    self.stored_rois_restored = False
    self.intensity_clamp = NO_CLAMP
    self._roi_preset_store = RoiPresetStore(get_user_roi_presets_dir())

roi_records property

roi_records

All ROI records currently held, across every frame.

add_shape

add_shape(verts, shape_type, tissue_class='undefined')

Add an ROI to the current frame and select it, which also enables Save.

tissue_class labels the ROI's location, e.g. the class it was placed in from a segmentation mask, and is exported with it.

Source code in src/optari/controllers/roi_controller.py

def add_shape(
    self,
    verts: np.ndarray,
    shape_type: str,
    tissue_class: str = "undefined",
) -> None:
    """Add an ROI to the current frame and select it, which also enables Save.

    *tissue_class* labels the ROI's location, e.g. the class it was placed in from a
    segmentation mask, and is exported with it.
    """
    shapes = self.optari_controller.shapes_layer
    shapes.add(verts, shape_type=shape_type)
    self.sync_records_from_shapes()
    self._roi_records[self._projection_ids[-1]].tissue_class = tissue_class
    shapes.selected_data = {len(shapes.data) - 1}

clear_roi_records

clear_roi_records()

Discard all ROI records and the rasterized-mask cache, e.g. when switching scans.

Source code in src/optari/controllers/roi_controller.py

def clear_roi_records(self) -> None:
    """Discard all ROI records and the rasterized-mask cache, e.g. when switching scans."""
    self.stored_rois_restored = False
    self._roi_records.clear()
    self._projection_ids = []
    clear_mask_cache()

current_records

current_records()

Records for the ROIs currently displayed, in display order.

Source code in src/optari/controllers/roi_controller.py

def current_records(self) -> list[ROIRecord]:
    """Records for the ROIs currently displayed, in display order."""
    return [self._roi_records[roi_id] for roi_id in self._projection_ids]

expand_current_projection_to_all_frames

expand_current_projection_to_all_frames(verts_for_frame=None)

Duplicate the last-placed ROI onto every other frame, as one group.

verts_for_frame(frame_id, source_verts) lets a caller resolve each frame's own geometry instead of reusing source_verts unchanged e.g. auto-placement re-anchoring to that frame's own segmentation mask. Returning None skips that frame entirely

Source code in src/optari/controllers/roi_controller.py

def expand_current_projection_to_all_frames(
    self,
    verts_for_frame: (
        Callable[[int, np.ndarray], np.ndarray | None] | None
    ) = None,
) -> None:
    """Duplicate the last-placed ROI onto every other frame, as one group.

    `verts_for_frame(frame_id, source_verts)` lets a caller resolve each
    frame's own geometry instead of reusing `source_verts` unchanged e.g.
    auto-placement re-anchoring to that frame's own segmentation mask.
    Returning None skips that frame entirely
    """
    self.sync_records_from_shapes()
    if not self._projection_ids:
        return
    source = self._roi_records[self._projection_ids[-1]]
    resolve = verts_for_frame or (lambda frame_id, verts: verts)
    for frame_id in range(self._n_frames()):
        if frame_id == source.frame_id:
            continue
        verts = resolve(frame_id, source.verts)
        if verts is None:
            continue
        record = ROIRecord(
            roi_id=self._next_roi_id,
            track_id=source.track_id,
            frame_id=frame_id,
            verts=verts,
            kind=source.kind,
            source=source.source,
            tissue_class=source.tissue_class,
            roi_group_uid=source.roi_group_uid,
        )
        self._roi_records[record.roi_id] = record
        self._next_roi_id += 1
    self.project_current_frame()

initialize_ui

initialize_ui()

Initialize ROI preset controls and current ROI display state.

Source code in src/optari/controllers/roi_controller.py

def initialize_ui(self) -> None:
    """Initialize ROI preset controls and current ROI display state."""
    self._refresh_roi_presets_ui()
    self._set_saved_table_view()
    self.refresh_ui()

measure_selected_rois

measure_selected_rois(selected_ids, *, scope=None, layers=None, clamp=None)

Measure the selected ROIs over every layer, frame and channel in scope.

scope defaults to the annotation dock's checkboxes, layers to whatever that scope resolves to and clamp to the dock's intensity filter. Batch mode passes all three, so its numbers depend on the plan alone, not on the dock.

Source code in src/optari/controllers/roi_controller.py

def measure_selected_rois(
    self,
    selected_ids: set[int],
    *,
    scope: MeasureScope | None = None,
    layers: list | None = None,
    clamp: IntensityClamp | None = None,
) -> pd.DataFrame:
    """Measure the selected ROIs over every layer, frame and channel in scope.

    *scope* defaults to the annotation dock's checkboxes, *layers* to whatever
    that scope resolves to and *clamp* to the dock's intensity filter. Batch mode
    passes all three, so its numbers depend on the plan alone, not on the dock.
    """
    active_layer = self.optari_controller.active_recon_layer
    if active_layer is None:
        logger.info("Select an image layer to save ROI stats")
        return pd.DataFrame()

    frame_channel = selected_frame_and_channel(self.viewer)
    if frame_channel is None:
        logger.info("Active layer has no frame/channel dimensions")
        return pd.DataFrame()
    frame_idx, channel_idx = frame_channel

    scope = scope if scope is not None else self._save_scope()
    clamp = clamp if clamp is not None else self.intensity_clamp
    follow_track = scope.all_frames and scope.follow_track
    if layers is not None:
        target_layers = layers
    elif scope.all_layers:
        target_layers = [
            layer
            for layer in self.viewer.layers
            if isinstance(layer, Image)
            and layer.metadata.get("type") == "pa"
        ]
    else:
        target_layers = [active_layer]

    tracks = self._track_histories_for_selection(selected_ids)

    collected: list[pd.DataFrame] = []
    for layer in target_layers:
        layer_data = layer.data
        frames_meta = layer.metadata.get("frames")
        frame_indices = (
            [int(f) for f in frames_meta or range(layer_data.shape[0])]
            if scope.all_frames
            else [frame_idx]
        )
        # The viewer's channel index is a wavelength of the reconstruction. On a derived
        # layer (unmixed, sO2, THb) it would pick an arbitrary chromophore, so those are
        # always measured on every channel.
        every_channel = (
            scope.all_layers
            or scope.all_channels
            or layer.metadata["pa_kind"] != "recon"
        )
        channel_indices = (
            list(range(layer_data.shape[1]))
            if every_channel
            else [channel_idx]
        )

        for f_idx in frame_indices:
            records_here = self._records_at_frame(
                tracks, f_idx, follow_track=follow_track
            )
            for c_idx in channel_indices:
                measured = compute_roi_stats(
                    records_here,
                    layer,
                    f_idx,
                    c_idx,
                    clamp=clamp,
                )
                if not measured.empty:
                    collected.append(measured)

    return (
        pd.concat(collected, ignore_index=True)
        if collected
        else pd.DataFrame()
    )

on_delete_saved_clicked

on_delete_saved_clicked(event=None)

Delete the selected rows from the Saved Analysis table.

Source code in src/optari/controllers/roi_controller.py

def on_delete_saved_clicked(self, event=None) -> None:
    """Delete the selected rows from the Saved Analysis table."""
    selection_model = (
        self.optari_controller.roi.saved_table.native.selectionModel()
    )
    selected_rows = selection_model.selectedRows()
    selected_indices = [idx.row() for idx in selected_rows]
    if not selected_indices:
        logger.warning("No row selected to delete.")
        return

    deleted = self.saved_table.delete_at(selected_indices)
    self._set_saved_table_view()
    logger.info("Deleted %s saved rows", deleted)

on_import_clicked

on_import_clicked(event=None)

Append a previously exported table to the saved table.

Source code in src/optari/controllers/roi_controller.py

def on_import_clicked(self, event=None) -> None:
    """Append a previously exported table to the saved table."""
    try:
        imported = import_roi_table_from_xlsx()
    except ValueError as exc:
        QMessageBox.warning(None, "Import failed", str(exc))
        return
    if imported is None:
        return

    rows, filename = imported
    added, skipped = self.saved_table.merge_imported(rows)
    self._set_saved_table_view()

    already = f", {skipped} already present" if skipped else ""
    show_info(
        f"Imported {added} row(s) from {Path(filename).name}{already}"
    )
    logger.info(
        "Imported %s of %s row(s) from %s", added, len(rows), filename
    )

on_intensity_settings_changed

on_intensity_settings_changed(_=None)

Read the intensity filter and measure.

Source code in src/optari/controllers/roi_controller.py

def on_intensity_settings_changed(self, _=None) -> None:
    """Read the intensity filter and measure."""
    annotation = self.optari_controller.annotation
    if annotation.roi_exclusion_box.isChecked():
        self.intensity_clamp = IntensityClamp(
            minimum=parse_float_input(
                annotation.roi_exclude_min_edit.text()
            ),
            maximum=parse_float_input(
                annotation.roi_exclude_max_edit.text()
            ),
            mode="exclude",
        )
    elif annotation.roi_clipping_box.isChecked():
        self.intensity_clamp = IntensityClamp(
            minimum=parse_float_input(annotation.roi_clip_min_edit.text()),
            maximum=parse_float_input(annotation.roi_clip_max_edit.text()),
        )
    else:
        # Both boxes off must mean no filtering, not the last range that was set.
        self.intensity_clamp = NO_CLAMP

    self.update_live_table()

on_live_table_delete_key

on_live_table_delete_key()

Delete selected shapes when Delete is pressed in the live table. on MacOS the actual Delete (forward-delete) key (Fn+Backspace) has to be pressed, not just the backspace key

Source code in src/optari/controllers/roi_controller.py

def on_live_table_delete_key(self) -> None:
    """Delete selected shapes when Delete is pressed in the live table.
    on MacOS the actual Delete (forward-delete) key (Fn+Backspace) has to be pressed, not just the backspace key
    """
    if self.optari_controller.shapes_layer is None:
        return
    if self.optari_controller.shapes_layer.selected_data:
        self.optari_controller.shapes_layer.remove_selected()

on_live_table_selection_changed

on_live_table_selection_changed()

Sync live table row selection → shapes layer selection.

Source code in src/optari/controllers/roi_controller.py

def on_live_table_selection_changed(self) -> None:
    """Sync live table row selection → shapes layer selection."""
    shapes = self.optari_controller.shapes_layer
    if self._syncing or shapes is None:
        return
    # Never write selection back into napari during shape drag/select interactions.
    if getattr(shapes, "_is_moving", False) or getattr(
        shapes, "_is_selecting", False
    ):
        return
    rows = {
        idx.row()
        for idx in self.optari_controller.roi.live_table.native.selectedIndexes()
    }
    self._syncing = True
    try:
        if rows and self.viewer.layers.selection.active is not shapes:
            self.viewer.layers.selection.active = shapes
        shapes.selected_data = rows
    finally:
        self._syncing = False

    # The selection callback is suppressed while _syncing is True,
    # so update the Save button state explicitly after table -> viewer sync.
    self._update_save_button_state()

on_place_roi_clicked

on_place_roi_clicked(event=None)

Place the preset currently selected in the presets list, using the dock's placement and scope.

Source code in src/optari/controllers/roi_controller.py

def on_place_roi_clicked(self, event=None) -> None:
    """Place the preset currently selected in the presets list, using the dock's placement and scope."""
    self._place_selected_roi_preset()

on_remove_roi_preset_clicked

on_remove_roi_preset_clicked(event=None)

Delete the preset currently selected in the presets list.

Source code in src/optari/controllers/roi_controller.py

def on_remove_roi_preset_clicked(self, event=None) -> None:
    """Delete the preset currently selected in the presets list."""
    item = self.optari_controller.annotation.roi_presets_list.currentItem()
    if item is None:
        return
    preset_name = item.text()
    if not preset_name:
        return

    try:
        removed = self._roi_preset_store.delete(preset_name)
    except (TypeError, ValueError, OSError) as exc:
        self.optari_controller.annotation.roi_presets_description_label.setText(
            f"Could not remove ROI preset: {exc}"
        )
        return
    if removed:
        self._refresh_roi_presets_ui()
        logger.info("Removed ROI preset '%s'", preset_name)

on_restore_selected_rows

on_restore_selected_rows(default_row=None)

Put the selected saved rows back into the viewer, one ROI per row. Restoring is per row, not per ROI group

Source code in src/optari/controllers/roi_controller.py

def on_restore_selected_rows(self, default_row: int | None = None) -> None:
    """Put the selected saved rows back into the viewer, one ROI per row.
    Restoring is per row, not per ROI group
    """
    selection = (
        self.optari_controller.roi.saved_table.native.selectionModel()
    )
    positions = [index.row() for index in selection.selectedRows()]
    if not positions and default_row is not None:
        positions = [default_row]
    if not positions:
        logger.info("No row selected to restore.")
        return

    rows = self.saved_table.rows_at(positions)
    try:
        problem = self._saved_rows_match_current_context(rows)
        if problem:
            show_error(problem)
            return
        records = self._records_from_saved_rows(rows)
    except (ValueError, TypeError) as exc:
        show_error(f"Could not restore ROI: {exc}")
        return

    if not records:
        show_info("Already in the viewer.")
        return

    for record in records:
        self._roi_records[record.roi_id] = record

    self._go_to_frame(int(rows["frame"].min()))
    self.project_current_frame()
    self.update_live_table()
    restored_ids = {r.roi_id for r in records}
    self.optari_controller.shapes_layer.selected_data = {
        index
        for index, roi_id in enumerate(self._projection_ids)
        if roi_id in restored_ids
    }
    show_info(f"Restored {len(records)} ROI(s)")

on_roi_preset_item_clicked

on_roi_preset_item_clicked(item)

Place the double-clicked preset.

Source code in src/optari/controllers/roi_controller.py

def on_roi_preset_item_clicked(self, item) -> None:
    """Place the double-clicked preset."""
    self._place_selected_roi_preset()

on_roi_preset_item_selected

on_roi_preset_item_selected(item, previous=None)

Preselect the placement for item's preset, and show its description and location.

Runs only when the selection changes, so a placement the user picked by hand survives clicking or double-clicking the same preset again.

Source code in src/optari/controllers/roi_controller.py

def on_roi_preset_item_selected(self, item, previous=None) -> None:
    """Preselect the placement for *item*'s preset, and show its description and location.

    Runs only when the selection changes, so a placement the user picked by hand
    survives clicking or double-clicking the same preset again.
    """
    combo = self.optari_controller.annotation.roi_placement_mode_combo
    location_item = combo.model().item(combo.findData("auto"))
    if item is None:
        location_item.setText("Location")
        location_item.setEnabled(False)
        combo.setCurrentIndex(combo.findData("static"))
        return
    try:
        preset = self._roi_preset_store.get(item.text())
    except (TypeError, ValueError, OSError) as exc:
        self.optari_controller.annotation.roi_presets_description_label.setText(
            f"Could not load ROI preset: {exc}"
        )
        return

    location = preset.geometry.tissue_class
    has_location = location != "undefined"
    location_item.setText(location if has_location else "Location")
    location_item.setEnabled(has_location)
    # Anchored placement without a matching segmentation then fails with a
    # message instead of quietly going static.
    combo.setCurrentIndex(combo.findData(preset.placement))

    parts = [preset.description]
    if has_location:
        parts.append(
            f"(location: {location}, top margin {preset.top_margin_mm:.2f} mm)"
        )
    self.optari_controller.annotation.roi_presets_description_label.setText(
        " ".join(filter(None, parts))
    )

on_save_clicked

on_save_clicked(event=None)

Measure the ROIs selected in the viewer and add to the Saved Analysis table.

Source code in src/optari/controllers/roi_controller.py

def on_save_clicked(self, event=None) -> None:
    """Measure the ROIs selected in the viewer and add to the Saved Analysis table."""
    selected_ids = self.selected_roi_ids()
    if not selected_ids:
        return

    rows = self.measure_selected_rois(selected_ids)
    if rows.empty:
        logger.info("Nothing to save")
        return

    self._append_saved_rows(rows, n_rois=len(selected_ids))

on_save_roi_preset_clicked

on_save_roi_preset_clicked(event=None)

Save the first selected ROI as a named preset.

Prompts for a name, description, location and top margin. A location makes the preset placed in that segmentation class, horizontally centred, with its top the given margin below the class's top edge.

Source code in src/optari/controllers/roi_controller.py

def on_save_roi_preset_clicked(self, event=None) -> None:
    """Save the first selected ROI as a named preset.

    Prompts for a name, description, location and top margin. A location makes
    the preset placed in that segmentation class, horizontally centred, with its top
    the given margin below the class's top edge.
    """
    if self.optari_controller.shapes_layer is None:
        return

    selected = list(self.optari_controller.shapes_layer.selected_data)
    if len(selected) == 0:
        logger.info("Save ROI preset clicked with no selected ROI")
        self.optari_controller.annotation.roi_presets_description_label.setText(
            "No ROI selected in viewer."
        )
        return

    self.sync_records_from_shapes()
    record = self.current_records()[int(selected[0])]
    verts = record.verts[:, -2:]

    # Offer the classes this frame's mask can anchor to (present at the image
    # centre), each with the ROI's current margin below its top edge.
    # Background (id 0) is no location.
    margins_by_location = {}
    mask_result = (
        self.optari_controller.segmentation_ctrl.active_seg_mask_2d()
    )
    if mask_result is not None:
        seg, seg_layer = mask_result
        class_names = (seg_layer.metadata or {}).get("class_names", {})
        for class_id in set(map(int, np.unique(seg))) - {0}:
            try:
                anchor = self._class_anchor_mm(seg == class_id, seg_layer)
            except ValueError:
                continue
            margins_by_location[str(class_names.get(class_id, class_id))] = (
                float(verts[:, 0].min() - anchor[0])
            )

    metadata = _roi_name_popup(margins_by_location, record.tissue_class)
    if metadata is None:
        return
    roi_id, description, location, top_margin_mm = metadata

    fov_m = self.optari_controller.scan_ctrl.get_fov()
    if fov_m is None:
        self.optari_controller.annotation.roi_presets_description_label.setText(
            "Could not save ROI preset: current scan has no valid FOV."
        )
        return
    geometry = RoiGeometry(
        verts_m=napari_to_patato(verts, *fov_m),
        kind=record.kind,
        tissue_class=location or "undefined",
    )
    try:
        preset_path = self._roi_preset_store.save_preset(
            name=roi_id,
            description=description,
            geometry=geometry,
            source_fov_m=fov_m,
            placement="static" if location is None else "auto",
            top_margin_mm=top_margin_mm,
        )
    except (TypeError, ValueError, OSError) as exc:
        self.optari_controller.annotation.roi_presets_description_label.setText(
            f"Could not save ROI preset: {exc}"
        )
        return

    self._refresh_roi_presets_ui()
    logger.info("Saved ROI preset '%s' to %s", roi_id, preset_path)

on_saved_row_double_clicked

on_saved_row_double_clicked(row, _column=0)

Restore the ROI of the double-clicked row onto that row's frame.

Source code in src/optari/controllers/roi_controller.py

def on_saved_row_double_clicked(self, row: int, _column: int = 0) -> None:
    """Restore the ROI of the double-clicked row onto that row's frame."""
    if row >= len(self.saved_table):
        return
    self.on_restore_selected_rows(default_row=row)

on_shapes_data_changed

on_shapes_data_changed(event=None)

Sync ROI records after a completed Shapes-layer edit, then refresh the display and live table.

Runs on every mouse-move of a drag. Ignores the intent event napari fires before a mutation (only the completion event is acted on), and skips the shape-display refresh unless the set of projected ids actually changed.

Source code in src/optari/controllers/roi_controller.py

def on_shapes_data_changed(self, event=None) -> None:
    """Sync ROI records after a completed Shapes-layer edit, then refresh the display and live table.

    Runs on every mouse-move of a drag. Ignores the intent event napari fires
    before a mutation (only the completion event is acted on), and skips the
    shape-display refresh unless the set of projected ids actually changed.
    """
    # napari emits events.data in pairs: an intent event before the layer is
    # mutated (adding/changing/removing) and a completion event after.
    # We only want to refresh the ROI table on the completion event, so ignore the intent event.
    action = getattr(event, "action", None)
    if action is not None and str(action) not in _COMPLETED_DATA_ACTIONS:
        return
    if self.optari_controller.shapes_layer is None or self._projecting:
        return

    # Several napari signals can report the same edit. only refresh for real ones.
    projected_before = list(self._projection_ids)
    if not self.sync_records_from_shapes():
        return

    # Shape labels and colours come from ids, sources and positions, none of which
    # is affected by geometry edits. So only refresh the display if the set of projected ids changed
    if self._projection_ids != projected_before:
        self._refresh_shape_display()

    self._update_save_button_state()
    self.update_live_table()

on_shapes_selection_changed

on_shapes_selection_changed(event=None)

Sync shapes selection -> live table selection on selection changes only.

Source code in src/optari/controllers/roi_controller.py

def on_shapes_selection_changed(self, event=None) -> None:
    """Sync shapes selection -> live table selection on selection changes only."""
    shapes = self.optari_controller.shapes_layer
    if self._syncing or shapes is None:
        return
    selected = list(shapes.selected_data)
    self._syncing = True
    try:
        self._set_live_table_selection(selected)
    finally:
        self._syncing = False

    self._update_save_button_state()

on_shapes_set_data

on_shapes_set_data(event=None)

Catch shape additions that never emit events.data.

copy/pasting shapes (e.g. from one frame to another) appends to the layer and emits only set_data, so a pasted ROI would stay invisible. set_data also fires on every redraw, hence the count check before doing any real work.

Source code in src/optari/controllers/roi_controller.py

def on_shapes_set_data(self, event=None) -> None:
    """Catch shape additions that never emit ``events.data``.

    copy/pasting shapes (e.g. from one frame to another) appends to the layer and emits
    only ``set_data``, so a pasted ROI would stay invisible. ``set_data`` also fires on every redraw,
    hence the count check before doing any real work.
    """
    shapes = self.optari_controller.shapes_layer
    if shapes is None or self._projecting:
        return
    if len(shapes.data) != len(self._projection_ids):
        self.on_shapes_data_changed()

on_xlsx_export_clicked

on_xlsx_export_clicked(event=None)

Prompt for a destination and export the Saved Analysis table to XLSX.

Source code in src/optari/controllers/roi_controller.py

def on_xlsx_export_clicked(self, event=None) -> None:
    """Prompt for a destination and export the Saved Analysis table to XLSX."""
    filename = export_roi_table_to_xlsx(self.saved_table.rows)
    if filename is None:
        return

    show_info(f"Exported ROI table to {Path(filename).name}")
    logger.info("Saved ROI table to %s", filename)

place_preset

place_preset(preset, *, placement, all_frames)

Place preset on the current frame, and across every frame if asked.

Auto placement re-anchors onto each frame's own segmentation when expanding, so every frame gets an ROI that belongs to it. The dock and a batch run both come through here, which is what makes "measure all frames" mean the same thing in both. Raises ValueError with a user-facing message when placement fails.

Source code in src/optari/controllers/roi_controller.py

def place_preset(
    self, preset, *, placement: str, all_frames: bool
) -> None:
    """Place *preset* on the current frame, and across every frame if asked.

    Auto placement re-anchors onto each frame's own segmentation when expanding,
    so every frame gets an ROI that belongs to it. The dock and a batch run both
    come through here, which is what makes "measure all frames" mean the same thing
    in both. Raises ValueError with a user-facing message when placement fails.
    """
    resolver = None
    if placement == "auto":
        resolver = self._per_frame_auto_resolver(
            *self._place_roi_preset_auto(self.optari_controller, preset),
            preset.top_margin_mm,
        )
    else:
        self._place_roi_preset_static(self.optari_controller, preset)
    # The GUI relies on napari's data events to fold the new shape into the record
    # store; do it explicitly so a synchronous caller sees it immediately too.
    self.sync_records_from_shapes()
    if all_frames:
        self.expand_current_projection_to_all_frames(resolver)

project_current_frame

project_current_frame()

Write the current frame's ROI records into the Shapes layer.

self._roi_records is the source of truth. This is the only place that should assign shapes.data directly. Guarded by self._projecting so the resulting data-change event doesn't sync back into the records it was generated from.

Source code in src/optari/controllers/roi_controller.py

def project_current_frame(self) -> None:
    """Write the current frame's ROI records into the Shapes layer.

    ``self._roi_records`` is the source of truth. This is the only place that
    should assign ``shapes.data`` directly. Guarded by ``self._projecting`` so the
    resulting data-change event doesn't sync back into the records it was
    generated from.
    """
    shapes = self.optari_controller.shapes_layer
    if shapes is None or self._projecting:
        return
    frame_id = selected_frame_idx(self.viewer, self._n_frames())
    visible = sorted(
        (r for r in self._roi_records.values() if r.frame_id == frame_id),
        key=lambda r: r.roi_id,
    )
    self._projecting = True
    try:
        # pass (verts, kind) pairs so napari never rebuilds shapes against a stale shape_type
        shapes.data = [(r.verts, r.kind) for r in visible]
        if visible:
            shapes.edge_width = [ROI_EDGE_WIDTH] * len(visible)
        self._projection_ids = [r.roi_id for r in visible]
        shapes.selected_data = set()
    finally:
        self._projecting = False
    self._refresh_shape_display()

refresh_ui

refresh_ui()

Refresh ROI controls after scan or active-layer state changes.

Source code in src/optari/controllers/roi_controller.py

def refresh_ui(self) -> None:
    """Refresh ROI controls after scan or active-layer state changes."""
    if self.optari_controller.shapes_layer is None:
        self.optari_controller.roi.save_button.setEnabled(False)
        self.update_live_table()
        return
    self._update_save_button_state()
    self.update_live_table()

restore_roi_records

restore_roi_records(records)

Replace all ROI records with the ones stored in the scan, resuming the id counters above their highest values.

Also projects the current frame, so the Shapes layer reflects the new records.

Source code in src/optari/controllers/roi_controller.py

def restore_roi_records(self, records: list[ROIRecord]) -> None:
    """Replace all ROI records with the ones stored in the scan, resuming the id
    counters above their highest values.

    Also projects the current frame, so the Shapes layer reflects the new records.
    """
    self.stored_rois_restored = True
    self._roi_records = {record.roi_id: record for record in records}
    self._next_roi_id = (
        max((record.roi_id for record in records), default=-1) + 1
    )
    self._next_track_id = (
        max((record.track_id for record in records), default=-1) + 1
    )
    self.project_current_frame()

selected_roi_ids

selected_roi_ids()

roi_ids of the ROIs selected in the viewer, empty (with a reason) if none.

Source code in src/optari/controllers/roi_controller.py

def selected_roi_ids(self) -> set[int]:
    """roi_ids of the ROIs selected in the viewer, empty (with a reason) if none."""
    if self.optari_controller.shapes_layer is None:
        return set()

    selected = list(self.optari_controller.shapes_layer.selected_data)
    if not selected:
        logger.info("Select at least one ROI to save")
        return set()
    return {
        self._projection_ids[index]
        for index in selected
        if index < len(self._projection_ids)
    }

sync_records_from_shapes

sync_records_from_shapes()

Keep frame-owned ROI records in sync with the projected Shapes layer.

Same/grown shape count: ids are kept positionally (napari always appends new shapes at the end. in-place edits keep their position). Smaller shape count: removal preserves the relative order of remaining shapes, so ids are recovered by matching remaining geometry to the old records.

Returns whether anything actually changed, so callers can skip the refresh when the layer already agrees with the records.

Source code in src/optari/controllers/roi_controller.py

def sync_records_from_shapes(self) -> bool:
    """Keep frame-owned ROI records in sync with the projected Shapes layer.

    Same/grown shape count: ids are kept positionally (napari always
    appends new shapes at the end. in-place edits keep their position).
    Smaller shape count: removal preserves the relative order of remaining shapes,
    so ids are recovered by matching remaining geometry to the old records.

    Returns whether anything actually changed, so callers can skip the
    refresh when the layer already agrees with the records.
    """
    shapes = self.optari_controller.shapes_layer
    if shapes is None or self._projecting:
        return False

    data = list(shapes.data)
    kinds = [str(k) for k in shapes.shape_type]
    old_ids = self._projection_ids

    if len(data) < len(old_ids):
        new_ids = self._match_ids_after_removal(data, kinds)
    else:
        new_ids = old_ids + [None] * (len(data) - len(old_ids))

    changed = len(data) != len(old_ids)
    for i, (verts, kind) in enumerate(zip(data, kinds)):
        if new_ids[i] is None:
            record = self._new_record(verts, kind)
            new_ids[i] = record.roi_id
            self._roi_records[record.roi_id] = record
            changed = True
        else:
            record = self._roi_records[new_ids[i]]
            if _layer_holds(record, verts, kind):
                continue
            record.verts = np.array(verts, dtype=float)
            record.kind = kind
            # An edited annotation is OPTARI's, but also the stored original stays as it was.
            record.source = OPTARI_SOURCE_TAG
            changed = True

    for stale_id in set(old_ids) - set(new_ids):
        del self._roi_records[stale_id]
        changed = True
    self._projection_ids = new_ids
    return changed

update_live_table

update_live_table(event=None)

Recompute the Live Analysis table for the ROIs on the current frame, active layer and channel.

Falls back to an empty table when there is no shapes layer or no active reconstruction layer to measure against.

Source code in src/optari/controllers/roi_controller.py

def update_live_table(self, event=None) -> None:
    """Recompute the Live Analysis table for the ROIs on the current frame, active layer and channel.

    Falls back to an empty table when there is no shapes layer or no active
    reconstruction layer to measure against.
    """
    live_cols = visible_feature_columns()
    table = self.optari_controller.roi.live_table
    shapes = self.optari_controller.shapes_layer
    layer = self.optari_controller.active_recon_layer
    if shapes is None or layer is None:
        table.value = pd.DataFrame(columns=live_cols)
        return

    frame_channel = selected_frame_and_channel(self.viewer)
    if frame_channel is None:
        return
    frame_idx, channel_idx = frame_channel
    df_live = compute_roi_stats(
        self.current_records(),
        layer,
        frame_idx,
        channel_idx,
        clamp=self.intensity_clamp,
        feature_ids=live_cols,
    )

    self._syncing = True
    try:
        table.value = df_live
        self._set_live_table_selection(list(shapes.selected_data))
    finally:
        self._syncing = False

    # Color first table column to match each ROI color. A shape whose geometry
    # cannot be measured has no row.
    for row_idx in range(len(shapes.data)):
        item = table.native.item(row_idx, 0)
        if item is not None:
            item.setBackground(QColor(roi_color_for_index(row_idx)))

optari.controllers.segmentation_controller

Segmentation controller: the ONNX model registry, running inference, the segmentation Labels layer, and automatic ROI-from-mask placement.

SegmentParams dataclass

SegmentParams(model_id, us_data, us_shape, class_ids, output_frames, us_scale, us_translate, current_frame_id=None)

Everything one segmentation run needs, resolved against a loaded scan.

frame_mode property

frame_mode

Whether the run covers the current frame or all frames.

n_channels property

n_channels

Number of channels in the source US layer.

build classmethod

build(controller, model_id, class_ids, *, frame_id)

Resolve a segmentation setup against the loaded scan's US layer.

Raises ValueError, message safe to show the user, when it cannot run here.

Source code in src/optari/controllers/segmentation_controller.py

@classmethod
def build(
    cls,
    controller,
    model_id: str,
    class_ids: set[int],
    *,
    frame_id: int | None,
) -> "SegmentParams":
    """Resolve a segmentation setup against the loaded scan's US layer.

    Raises ValueError, message safe to show the user, when it cannot run here.
    """
    us_layer = controller.active_us_layer
    if us_layer is None:
        raise ValueError("No US layer found.")
    if not class_ids:
        raise ValueError("No classes selected.")

    us_raw = np.asarray(us_layer.data)
    us_data = us_raw[:, 0]
    n_frames = int(us_raw.shape[0])
    output_frames = list(range(n_frames))
    if frame_id is not None:
        if not (0 <= frame_id < n_frames):
            raise ValueError("Selected frame is not available.")
        us_data = us_data[frame_id : frame_id + 1]
        output_frames = [frame_id]

    return cls(
        model_id=model_id,
        us_data=us_data,
        us_shape=us_raw.shape,
        class_ids=set(class_ids),
        output_frames=output_frames,
        us_scale=tuple(us_layer.scale),
        us_translate=tuple(us_layer.translate),
        current_frame_id=frame_id,
    )

SegmentationController

SegmentationController(parent_controller)

Bases: TaskControllerBase

Segmentation-related UI actions and geometry helpers. Segmentation masks are generated either for a single frame or all frames, depending on the checkbox state in the UI. For single frame, the mask will be shown only on the selected frame. In any case, the mask is padded to the full shape of the US data and repeated across the channel dimension for correct napari display.

Load the model registry and initialize segmentation state.

Source code in src/optari/controllers/segmentation_controller.py

def __init__(self, parent_controller):
    """Load the model registry and initialize segmentation state."""
    super().__init__(parent_controller)
    self._segmentation_model_registry = load_model_registry()
    default_model_id = settings.segmentation.default_model
    if default_model_id not in self._segmentation_model_registry:
        logger.warning(
            f"Default segmentation model '{default_model_id}' not found in registry. "
            f"Available models: {list(self._segmentation_model_registry.keys())}. "
            f"Falling back to first available model."
        )
        default_model_id = next(iter(self._segmentation_model_registry))
    self._active_segmentation_model_id = default_model_id
    self._segmenter = None
    self._segmenter_model_id = None
    self._seg_layer: Labels | None = None
    self._pending_roi_class_id: int | None = None
    self.preset_store = PresetStore(get_user_segmentation_presets_dir())

active_segmentation_model_id property

active_segmentation_model_id

ID of the currently selected segmentation model.

seg_layer property

seg_layer

The current segmentation layer, generated or restored, or None.

active_seg_mask_2d

active_seg_mask_2d()

Return (H, W) segmentation mask for the current viewer frame, or None.

Source code in src/optari/controllers/segmentation_controller.py

def active_seg_mask_2d(self) -> tuple[np.ndarray, "Labels"] | None:
    """Return (H, W) segmentation mask for the current viewer frame, or None."""
    seg_layer = self.seg_layer
    if seg_layer is None:
        return None
    seg = np.asarray(seg_layer.data)
    frame_idx = selected_frame_idx(self.viewer, seg.shape[0])
    return seg[frame_idx, 0], seg_layer

active_segmentation_class_items

active_segmentation_class_items()

Return sorted (class_id, class_name) entries for the active model.

Source code in src/optari/controllers/segmentation_controller.py

def active_segmentation_class_items(self) -> list[tuple[int, str]]:
    """Return sorted ``(class_id, class_name)`` entries for the active model."""
    return sorted(
        self._active_segmentation_model_config().class_names.items()
    )

get_segmenter

get_segmenter(model_id=None)

The segmenter for model_id, defaulting to the dock's current selection.

A caller that already decided which model to run (a batch plan naming one) must get that model, not whichever the dock happens to be showing.

Source code in src/optari/controllers/segmentation_controller.py

def get_segmenter(self, model_id: str | None = None):
    """The segmenter for *model_id*, defaulting to the dock's current selection.

    A caller that already decided which model to run (a batch plan naming one)
    must get that model, not whichever the dock happens to be showing.
    """
    model_id = model_id or self._active_segmentation_model_id
    if self._segmenter is None or self._segmenter_model_id != model_id:
        self._segmenter = create_segmenter(
            self._segmentation_model_registry[model_id]
        )
        self._segmenter_model_id = model_id
    return self._segmenter

initialize_ui

initialize_ui()

Initialize model combo and populate default classes once.

Source code in src/optari/controllers/segmentation_controller.py

def initialize_ui(self) -> None:
    """Initialize model combo and populate default classes once."""

    dock = self.optari_controller.segmentation

    if dock.segmentation_model_combo.count() == 0:
        # Populate available models once on first dock init.
        for model_id in self.segmentation_model_options():
            dock.segmentation_model_combo.addItem(
                model_id, userData=model_id
            )

        # Set to active model (default is first in registry).
        for i in range(dock.segmentation_model_combo.count()):
            if (
                dock.segmentation_model_combo.itemData(i)
                == self.active_segmentation_model_id
            ):
                dock.segmentation_model_combo.setCurrentIndex(i)
                break

    self.on_segmentation_model_changed()

    populate_preset_combo(dock.preset_combo, self.preset_store)

    if dock.preset_combo.count() > 0:
        self.on_preset_changed()

    self.refresh_ui()

model_weights_path

model_weights_path(model_id)

Where the given model's ONNX weights are expected on disk.

Source code in src/optari/controllers/segmentation_controller.py

def model_weights_path(self, model_id: str) -> Path:
    """Where the given model's ONNX weights are expected on disk."""
    if model_id not in self._segmentation_model_registry:
        raise ValueError(f"Unknown segmentation model: {model_id}")
    return (
        get_user_models_dir()
        / self._segmentation_model_registry[model_id].filename
    )

on_generate_roi_from_mask_clicked

on_generate_roi_from_mask_clicked()

Generate one ROI from the selected frame of the segmentation mask.

Source code in src/optari/controllers/segmentation_controller.py

def on_generate_roi_from_mask_clicked(self) -> None:
    """Generate one ROI from the selected frame of the segmentation mask."""
    seg_dock = self.optari_controller.segmentation
    result = self.active_seg_mask_2d()
    if result is None:
        seg_dock.status_label.setText("No segmentation mask found")
        return
    seg_2d, seg_layer = result

    class_id = seg_dock.roi_class_id_combo.currentData()
    if class_id is None:
        seg_dock.status_label.setText("Select a class id")
        return

    sy, sx = float(seg_layer.scale[-2]), float(seg_layer.scale[-1])
    ty, tx = float(seg_layer.translate[-2]), float(seg_layer.translate[-1])

    shape_type = str(seg_dock.roi_shape_combo.currentData() or "ellipse")
    try:
        top_margin_mm = self._parse_roi_value(
            seg_dock.roi_top_margin_edit.text(), "Top margin"
        )
        width_mm = self._parse_roi_value(
            seg_dock.roi_width_edit.text(), "Width"
        )
        height_mm = self._parse_roi_value(
            seg_dock.roi_height_edit.text(), "Height"
        )
    except ValueError as exc:
        seg_dock.status_label.setText(str(exc))
        return
    if (width_mm is None or height_mm is None) and shape_type in (
        "rectangle",
        "ellipse",
    ):
        seg_dock.status_label.setText("Width and height are required")
        return

    config = ROIPlacementConfig(
        width_mm=width_mm,
        height_mm=height_mm,
        depth_mm=top_margin_mm or 0.0,
    )

    frame_idx = selected_frame_idx(
        self.viewer, np.asarray(seg_layer.data).shape[0]
    )
    class_mask = seg_2d == int(class_id)
    if not np.any(class_mask):
        seg_dock.status_label.setText(
            "Selected class is not present in the selected frame"
        )
        return

    try:
        verts = roi_verts_from_mask(
            shape_type, class_mask, config, sy=sy, sx=sx, ty=ty, tx=tx
        )
    except ValueError as exc:
        seg_dock.status_label.setText(str(exc))
        return

    class_names = (seg_layer.metadata or {}).get("class_names", {})
    self.optari_controller.roi_ctrl.add_shape(
        verts, shape_type, str(class_names.get(int(class_id), class_id))
    )
    seg_dock.status_label.setText(
        f"ROI generated from class {class_id} in frame {frame_idx}"
    )

on_generate_tissue_segmentation_clicked

on_generate_tissue_segmentation_clicked()

Run segmentation from the dock's Run button.

Source code in src/optari/controllers/segmentation_controller.py

def on_generate_tissue_segmentation_clicked(self) -> None:
    """Run segmentation from the dock's Run button."""
    self.run_from_ui(self.optari_controller.segmentation)

on_preset_changed

on_preset_changed()

Load and apply the selected segmentation preset.

Source code in src/optari/controllers/segmentation_controller.py

def on_preset_changed(self) -> None:
    """Load and apply the selected segmentation preset."""
    dock = self.optari_controller.segmentation
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        return

    try:
        settings = self.preset_store.load(preset_path)
        settings = validate_segmentation_settings(settings)
        model_config = self._segmentation_model_registry.get(
            settings["model_id"]
        )
        if model_config is None:
            raise ValueError(
                f"Unknown segmentation model: {settings['model_id']}"
            )
        settings = validate_segmentation_settings(
            settings, model_config.class_names
        )
        self._apply_preset(settings)
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not apply preset: {exc}")

on_remove_preset_clicked

on_remove_preset_clicked()

Remove the selected segmentation preset.

Source code in src/optari/controllers/segmentation_controller.py

def on_remove_preset_clicked(self) -> None:
    """Remove the selected segmentation preset."""
    dock = self.optari_controller.segmentation
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        dock.status_label.setText("Select a preset to remove.")
        return

    try:
        removed_path, removed = remove_selected_preset(
            dock.preset_combo, self.preset_store
        )
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not remove preset: {exc}")
        return
    if not removed:
        dock.status_label.setText(f"Preset not found: {preset_path.name}")
        return

    dock.status_label.setText(f"Removed preset: {removed_path.name}")

on_save_preset_clicked

on_save_preset_clicked()

Save the current segmentation and ROI settings as a new user preset.

Source code in src/optari/controllers/segmentation_controller.py

def on_save_preset_clicked(self) -> None:
    """Save the current segmentation and ROI settings as a new user preset."""
    dock = self.optari_controller.segmentation
    selected_class_ids = sorted(self.selected_segmentation_class_ids())
    if not selected_class_ids:
        dock.status_label.setText("Select at least one class.")
        return

    try:
        preset_values = {
            "model_id": self.active_segmentation_model_id,
            "selected_class_ids": selected_class_ids,
            "roi_class_id": dock.roi_class_id_combo.currentData(),
            "roi_shape": dock.roi_shape_combo.currentData() or "ellipse",
            "roi_width_mm": self._parse_roi_value(
                dock.roi_width_edit.text(), "ROI width"
            ),
            "roi_height_mm": self._parse_roi_value(
                dock.roi_height_edit.text(), "ROI height"
            ),
            "roi_top_margin_mm": self._parse_roi_value(
                dock.roi_top_margin_edit.text(), "ROI top margin"
            ),
        }
        preset_name = prompt_preset_name(
            dock.widget, dock.preset_combo, "segmentation_preset"
        )
        if preset_name is None:
            return
        preset_values = validate_segmentation_settings(
            preset_values,
            self._segmentation_model_registry[
                self.active_segmentation_model_id
            ].class_names,
        )
        preset_path = self.preset_store.save(preset_name, preset_values)
    except (KeyError, ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not save preset: {exc}")
        return

    add_preset_to_combo(dock.preset_combo, preset_path)
    dock.status_label.setText(f"Saved preset: {preset_path.name}")

on_segmentation_clear_classes_clicked

on_segmentation_clear_classes_clicked()

Uncheck all segmentation classes.

Source code in src/optari/controllers/segmentation_controller.py

def on_segmentation_clear_classes_clicked(self) -> None:
    """Uncheck all segmentation classes."""
    set_checked(
        self.optari_controller.segmentation.segmentation_classes_list,
        lambda item: False,
    )

on_segmentation_model_changed

on_segmentation_model_changed()

Apply the model picked in the combo and repopulate the class list.

Source code in src/optari/controllers/segmentation_controller.py

def on_segmentation_model_changed(self) -> None:
    """Apply the model picked in the combo and repopulate the class list."""
    dock = self.optari_controller.segmentation
    model_id = dock.segmentation_model_combo.currentData()
    if model_id is None:
        return
    self.set_active_segmentation_model(str(model_id))
    self.populate_segmentation_controls()
    self._pending_roi_class_id = None
    dock.status_label.setText(
        f"Model: {dock.segmentation_model_combo.currentText()}"
    )

on_segmentation_select_all_classes_clicked

on_segmentation_select_all_classes_clicked()

Check all segmentation classes.

Source code in src/optari/controllers/segmentation_controller.py

def on_segmentation_select_all_classes_clicked(self) -> None:
    """Check all segmentation classes."""
    set_checked(
        self.optari_controller.segmentation.segmentation_classes_list,
        lambda item: True,
    )

populate_segmentation_controls

populate_segmentation_controls(selected_class_ids=None)

Rebuild the classes list for the active model.

Checks selected_class_ids if given, otherwise falls back to the model's own default class.

Source code in src/optari/controllers/segmentation_controller.py

def populate_segmentation_controls(
    self, selected_class_ids: set[int] | None = None
) -> None:
    """Rebuild the classes list for the active model.

    Checks *selected_class_ids* if given, otherwise falls back to the model's
    own default class.
    """
    seg_dock = self.optari_controller.segmentation
    default_class = self._active_segmentation_model_config().default_class
    seg_dock.segmentation_classes_list.clear()

    for class_id, class_name in self.active_segmentation_class_items():
        add_checkable_item(
            seg_dock.segmentation_classes_list,
            f"{class_id}: {class_name}",
            class_id,
            checked=(
                class_id in selected_class_ids
                if selected_class_ids is not None
                else default_class == class_name
            ),
        )

prepare

prepare(params)

Validate params and return the work to run. Raises ValueError if it can't run.

Model weights are required to already be on disk: fetching them is the caller's job, so an unattended run can download once up front instead of per scan.

Source code in src/optari/controllers/segmentation_controller.py

def prepare(self, params: SegmentParams) -> BackgroundStep:
    """Validate *params* and return the work to run. Raises ValueError if it can't run.

    Model weights are required to already be on disk: fetching them is the caller's
    job, so an unattended run can download once up front instead of per scan.
    """
    weights = self.model_weights_path(params.model_id)
    if not weights.is_file():
        raise ValueError(
            f"Segmentation model weights not found: {weights}"
        )

    logger.info(
        "running segmentation with model '%s' on %s frame(s)",
        params.model_id,
        len(params.output_frames),
    )
    return BackgroundStep(
        func=partial(
            _segment_frames,
            self.get_segmenter(params.model_id),
            params.us_data,
            params.class_ids,
            params.n_channels,
            params.current_frame_id,
            params.us_shape,
        ),
        total=len(params.output_frames),
        desc="Segmenting",
    )

publish

publish(result, params)

Add the finished segmentation as a labels layer. Runs on the main thread.

Source code in src/optari/controllers/segmentation_controller.py

def publish(self, result, params: SegmentParams) -> str:
    """Add the finished segmentation as a labels layer. Runs on the main thread."""
    mask, class_names, blank_positions = result

    if (
        self._seg_layer is not None
        and self._seg_layer in self.viewer.layers
    ):
        self.viewer.layers.remove(self._seg_layer)
    self._seg_layer = self.viewer.add_labels(
        mask,
        name="Segmentation",
        scale=params.us_scale,
        translate=params.us_translate,
        opacity=0.5,
        metadata={
            "type": "segmentation",
            "frame_mode": params.frame_mode,
            "frames": params.output_frames,
            "class_names": class_names,
            "source_model_id": params.model_id,
            "settings": {
                "model_id": params.model_id,
                "selected_class_ids": sorted(params.class_ids),
                "class_names": class_names,
                "frame_mode": params.frame_mode,
                "frames": params.output_frames,
            },
        },
        units=self.image_units,
    )
    self._populate_roi_class_combo(mask, class_names)
    if not blank_positions:
        return f"{len(params.output_frames)} frame(s) segmented"
    blank_frames = [params.output_frames[i] for i in blank_positions]
    logger.warning(
        "%d frame(s) contain no image data and were left unsegmented: %s",
        len(blank_frames),
        blank_frames,
    )
    return (
        f"{len(params.output_frames)} frame(s) segmented, "
        f"{len(blank_frames)} without image data left empty"
    )

refresh_ui

refresh_ui()

Refresh controls that require scan or segmentation output.

Source code in src/optari/controllers/segmentation_controller.py

def refresh_ui(self) -> None:
    """Refresh controls that require scan or segmentation output."""
    dock = self.optari_controller.segmentation
    has_us_layer = (
        self.optari_controller.pa_data is not None
        and self.optari_controller.active_us_layer is not None
    )
    dock.generate_tissue_segmentation_button.setEnabled(
        has_us_layer and not self.optari_controller.task_running
    )
    dock.generate_roi_button.setEnabled(
        self.active_seg_mask_2d() is not None
        and not self.optari_controller.task_running
    )

restore_from_scan

restore_from_scan(pa_data)

Rebuild the segmentation layer from scan data.

Source code in src/optari/controllers/segmentation_controller.py

def restore_from_scan(self, pa_data) -> None:
    """Rebuild the segmentation layer from scan data."""
    us_layer = self.optari_controller.active_us_layer
    if us_layer is None:
        return
    restored = segmentation_from_scan(pa_data)
    if restored is None:
        return

    n_channels = int(np.asarray(us_layer.data).shape[1])
    mask = np.repeat(restored["mask"][:, np.newaxis], n_channels, axis=1)
    class_names = restored["class_names"]
    self._seg_layer = self.viewer.add_labels(
        mask,
        name="Segmentation",
        scale=us_layer.scale,
        translate=us_layer.translate,
        opacity=0.5,
        metadata={
            "type": "segmentation",
            "frame_mode": restored["frame_mode"],
            "frames": restored["frames"],
            "class_names": class_names,
            "source_model_id": restored["source_model_id"],
        },
        units=self.image_units,
    )
    self._populate_roi_class_combo(mask, class_names)

segmentation_model_options

segmentation_model_options()

Return model IDs for the model combo.

Source code in src/optari/controllers/segmentation_controller.py

def segmentation_model_options(self) -> list[str]:
    """Return model IDs for the model combo."""
    return list(self._segmentation_model_registry.keys())

selected_segmentation_class_ids

selected_segmentation_class_ids()

Return the class IDs checked in the segmentation classes list.

Source code in src/optari/controllers/segmentation_controller.py

def selected_segmentation_class_ids(self) -> set[int]:
    """Return the class IDs checked in the segmentation classes list."""
    return {
        item.data(Qt.UserRole)
        for item in checked_items(
            self.optari_controller.segmentation.segmentation_classes_list
        )
    }

set_active_segmentation_model

set_active_segmentation_model(model_id)

Set the active model. Raises ValueError if model_id is not in the registry.

Source code in src/optari/controllers/segmentation_controller.py

def set_active_segmentation_model(self, model_id: str) -> None:
    """Set the active model. Raises ValueError if *model_id* is not in the registry."""
    if model_id not in self._segmentation_model_registry:
        raise ValueError(f"Unknown segmentation model: {model_id}")
    self._active_segmentation_model_id = model_id

teardown

teardown()

Release cached segmentation model to free memory.

Source code in src/optari/controllers/segmentation_controller.py

def teardown(self) -> None:
    """Release cached segmentation model to free memory."""
    self._segmenter = None
    self._segmenter_model_id = None
    self._seg_layer = None

weights_to_fetch

weights_to_fetch(params)

Local path and download URL for the model params requires.

Source code in src/optari/controllers/segmentation_controller.py

def weights_to_fetch(
    self, params: SegmentParams
) -> tuple[Path, str | None]:
    """Local path and download URL for the model *params* requires."""
    return (
        self.model_weights_path(params.model_id),
        self._segmentation_model_registry[params.model_id].url,
    )

optari.controllers.reconstruction_controller

Reconstruction controller: presets and running PATATO's backprojection/DeepMB pipelines.

ReconParams dataclass

ReconParams(settings, pa_data_for_run, algorithm_name, speed_of_sound, suffix, output_frames, source_frame_count, scan_path, scan_name=None, current_frame_id=None, offset_x_mm=0.0, offset_z_mm=0.0, timestamps=None, acquisition_start=None, n_wavelengths=1)

Everything one reconstruction run needs, resolved against a loaded scan.

The dock fills this from its widgets and batch mode from a preset, so neither prepare nor publish has to reach back into ambient viewer or widget state.

chunk_frames property

chunk_frames

Frames per reconstruction chunk, capped so frames x wavelengths stays within PATATO's batch size limit.

layer_name property

layer_name

Output layer name, encoding the algorithm, suffix and frame.

from_settings classmethod

from_settings(settings_dict, controller, *, speed_of_sound=None, suffix='', frame_id=None)

Resolve a reconstruction preset against the currently loaded scan.

speed_of_sound defaults to the preset's own value. The dock passes its slider instead, which is what lets the slider override the preset.

Raises ValueError, message safe to show the user, when the preset cannot be run on this scan.

Source code in src/optari/controllers/reconstruction_controller.py

@classmethod
def from_settings(
    cls,
    settings_dict: dict,
    controller,
    *,
    speed_of_sound: float | None = None,
    suffix: str = "",
    frame_id: int | None = None,
) -> "ReconParams":
    """Resolve a reconstruction preset against the currently loaded scan.

    *speed_of_sound* defaults to the preset's own value. The dock passes its
    slider instead, which is what lets the slider override the preset.

    Raises ValueError, message safe to show the user, when the preset cannot be
    run on this scan.
    """
    pa_data = controller.pa_data
    if pa_data is None:
        raise ValueError("Load a scan first.")

    settings_dict = dict(settings_dict)
    if speed_of_sound is None:
        if ReconAttributeTags.SPEED_OF_SOUND not in settings_dict:
            raise ValueError(
                "Reconstruction preset has no speed of sound."
            )
        speed_of_sound = settings_dict[ReconAttributeTags.SPEED_OF_SOUND]
    offset_x_mm = float(settings_dict.pop("OFFSET_X", 0.0))
    offset_z_mm = float(settings_dict.pop("OFFSET_Z", 0.0))

    source_frame_count = int(pa_data.shape[0])
    pa_data_for_run = pa_data
    output_frames = list(range(source_frame_count))
    if frame_id is not None:
        if not (0 <= frame_id < source_frame_count):
            raise ValueError("Selected frame is not available.")
        pa_data_for_run = pa_data[frame_id : frame_id + 1]
        output_frames = [frame_id]

    return cls(
        settings=settings_dict,
        pa_data_for_run=pa_data_for_run,
        algorithm_name=settings_dict.get(
            ReconAttributeTags.RECONSTRUCTION_ALGORITHM, "Reconstruction"
        ),
        speed_of_sound=float(speed_of_sound),
        suffix=suffix.strip(),
        output_frames=output_frames,
        source_frame_count=source_frame_count,
        scan_path=Path(controller.path),
        scan_name=controller.scan_ctrl.scan_name(),
        current_frame_id=frame_id,
        offset_x_mm=offset_x_mm,
        offset_z_mm=offset_z_mm,
        timestamps=np.asarray(pa_data.get_timestamps()),
        acquisition_start=acquisition_start(pa_data),
        n_wavelengths=max(1, int(pa_data.shape[1])),
    )

ReconstructionController

ReconstructionController(parent_controller)

Bases: TaskControllerBase

Run PATATO reconstruction presets on the loaded scan and add as layers.

Initialize preset state and the reconstruction preset store.

Source code in src/optari/controllers/reconstruction_controller.py

def __init__(self, parent_controller):
    """Initialize preset state and the reconstruction preset store."""
    super().__init__(parent_controller)
    self._updating_settings = False
    self._settings_dirty = False
    self._applied_settings: dict = {}
    self.preset_store = PresetStore(get_user_reconstruction_presets_dir())

initialize_ui

initialize_ui()

Populate the preset combo once.

Source code in src/optari/controllers/reconstruction_controller.py

def initialize_ui(self) -> None:
    """Populate the preset combo once."""
    dock = self.optari_controller.reconstruction

    populate_preset_combo(dock.preset_combo, self.preset_store)

    self.on_preset_changed()
    self.refresh_ui()

on_all_settings_toggled

on_all_settings_toggled(checked)

Show or hide the in-memory JSON settings editor.

Source code in src/optari/controllers/reconstruction_controller.py

def on_all_settings_toggled(self, checked: bool) -> None:
    """Show or hide the in-memory JSON settings editor."""
    dock = self.optari_controller.reconstruction
    dock.settings_edit.setVisible(checked)
    dock.apply_preset_button.setVisible(checked)
    dock.all_settings_button.setArrowType(
        Qt.DownArrow if checked else Qt.RightArrow
    )

on_apply_preset_clicked

on_apply_preset_clicked()

Validate and apply the edited JSON settings in memory.

Source code in src/optari/controllers/reconstruction_controller.py

def on_apply_preset_clicked(self) -> None:
    """Validate and apply the edited JSON settings in memory."""
    dock = self.optari_controller.reconstruction
    try:
        settings_dict = json.loads(dock.settings_edit.toPlainText())
    except json.JSONDecodeError as exc:
        dock.status_label.setText(f"Invalid JSON settings: {exc.msg}")
        return
    if not isinstance(settings_dict, dict):
        dock.status_label.setText("JSON settings must be an object.")
        return

    self._applied_settings = settings_dict
    self._settings_dirty = False
    speed_of_sound = settings_dict.get(
        ReconAttributeTags.SPEED_OF_SOUND, SPEED_OF_SOUND_DEFAULT
    )
    warning = self._set_speed_of_sound_slider(speed_of_sound)
    dock.status_label.setText(f"Preset applied. {warning}".strip())

on_preset_changed

on_preset_changed()

Load the selected preset into the in-memory settings editor.

Source code in src/optari/controllers/reconstruction_controller.py

def on_preset_changed(self) -> None:
    """Load the selected preset into the in-memory settings editor."""
    dock = self.optari_controller.reconstruction
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        self._set_settings_editor({})
        return

    try:
        preset_settings = self.preset_store.load(preset_path)
        self._set_settings_editor(preset_settings)
        self._applied_settings = preset_settings
        self._settings_dirty = False
        warning = self._set_speed_of_sound_slider(
            preset_settings.get(
                ReconAttributeTags.SPEED_OF_SOUND, SPEED_OF_SOUND_DEFAULT
            )
        )
    except (ValueError, OSError) as e:
        dock.status_label.setText(f"Could not load preset: {e}")
        return
    if warning:
        dock.status_label.setText(warning)

on_remove_preset_clicked

on_remove_preset_clicked()

Remove the selected reconstruction preset.

Source code in src/optari/controllers/reconstruction_controller.py

def on_remove_preset_clicked(self) -> None:
    """Remove the selected reconstruction preset."""
    dock = self.optari_controller.reconstruction
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        dock.status_label.setText("Select a preset to remove.")
        return

    try:
        removed_path, removed = remove_selected_preset(
            dock.preset_combo, self.preset_store
        )
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not remove preset: {exc}")
        return
    if not removed:
        dock.status_label.setText(f"Preset not found: {preset_path.name}")
        return

    dock.status_label.setText(f"Removed preset: {removed_path.name}")

on_run_reconstruction_clicked

on_run_reconstruction_clicked()

Run reconstruction from the dock's Run button.

Source code in src/optari/controllers/reconstruction_controller.py

def on_run_reconstruction_clicked(self) -> None:
    """Run reconstruction from the dock's Run button."""
    self.run_from_ui(self.optari_controller.reconstruction)

on_save_preset_clicked

on_save_preset_clicked()

Save the current JSON editor contents as a new user preset.

Source code in src/optari/controllers/reconstruction_controller.py

def on_save_preset_clicked(self) -> None:
    """Save the current JSON editor contents as a new user preset."""
    dock = self.optari_controller.reconstruction
    try:
        settings_dict = json.loads(dock.settings_edit.toPlainText())
    except json.JSONDecodeError as exc:
        dock.status_label.setText(f"Invalid JSON settings: {exc.msg}")
        return
    if not isinstance(settings_dict, dict):
        dock.status_label.setText("JSON settings must be an object.")
        return

    preset_name = prompt_preset_name(
        dock.widget, dock.preset_combo, "reconstruction_preset"
    )
    if preset_name is None:
        return

    try:
        preset_path = self.preset_store.save(preset_name, settings_dict)
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not save preset: {exc}")
        return

    add_preset_to_combo(dock.preset_combo, preset_path)
    dock.status_label.setText(f"Saved preset: {preset_path.name}")

on_settings_text_changed

on_settings_text_changed()

Mark the editor contents as pending without rewriting the document.

Source code in src/optari/controllers/reconstruction_controller.py

def on_settings_text_changed(self) -> None:
    """Mark the editor contents as pending without rewriting the document."""
    if self._updating_settings:
        return
    self._settings_dirty = True

on_speed_of_sound_changed

on_speed_of_sound_changed(value)

Keep the speed of sound label in sync with the slider.

Source code in src/optari/controllers/reconstruction_controller.py

def on_speed_of_sound_changed(self, value: int) -> None:
    """Keep the speed of sound label in sync with the slider."""
    dock = self.optari_controller.reconstruction
    dock.speed_of_sound_value_label.setText(f"{value} m/s")

prepare

prepare(params)

Validate params and return the work to run. Raises ValueError if it can't run.

Model weights are required to already be on disk: fetching them is the caller's job, so an unattended run can download once up front instead of per scan.

Source code in src/optari/controllers/reconstruction_controller.py

def prepare(self, params: ReconParams) -> BackgroundStep:
    """Validate *params* and return the work to run. Raises ValueError if it can't run.

    Model weights are required to already be on disk: fetching them is the caller's
    job, so an unattended run can download once up front instead of per scan.
    """
    weights = self.weights_to_fetch(params)
    if weights is not None:
        model_path, _ = weights
        if not model_path.is_file():
            raise ValueError(
                f"DeepMB model weights not found: {model_path}"
            )
        _resolve_deepmb_model(params.settings, model_path)

    n_chunks = ceil(len(params.output_frames) / params.chunk_frames)
    logger.info(
        "running reconstruction preset '%s' on %s frame(s) in %s chunk(s) of %s, "
        "speed of sound=%s m/s",
        params.algorithm_name,
        len(params.output_frames),
        n_chunks,
        params.chunk_frames,
        params.speed_of_sound,
    )
    return BackgroundStep(
        func=partial(
            _reconstruct_frames,
            params.settings,
            params.pa_data_for_run,
            params.speed_of_sound,
            params.chunk_frames,
        ),
        total=n_chunks + 1,
        desc="Reconstructing",
    )

publish

publish(result, params)

Add the finished reconstruction as a layer. Runs on the main thread.

Source code in src/optari/controllers/reconstruction_controller.py

def publish(self, result, params: ReconParams) -> str:
    """Add the finished reconstruction as a layer. Runs on the main thread."""
    reconstruction, new_settings = result

    wavelengths = [int(w) for w in reconstruction.ax_1_labels]
    settings = dict(params.settings)
    settings.update(new_settings)
    settings.update(
        {
            ReconAttributeTags.RECONSTRUCTION_ALGORITHM: params.algorithm_name,
            ReconAttributeTags.SPEED_OF_SOUND: params.speed_of_sound,
            "OFFSET_X": params.offset_x_mm,
            "OFFSET_Z": params.offset_z_mm,
            "frame_mode": (
                "current" if params.current_frame_id is not None else "all"
            ),
            "frames": params.output_frames,
            "suffix": params.suffix,
        }
    )
    layer_metadata = {
        "type": "pa",
        "pa_kind": "recon",
        "wavelengths": wavelengths,
        "axis1_name": "Channel",
        "axis1_labels": wavelengths,
        "filepath": str(params.scan_path),
        "scan_name": params.scan_name,
        "timestamps": params.timestamps,
        "acquisition_start": params.acquisition_start,
        "frames": params.output_frames,
        "settings": settings,
    }

    # reopened single-frame recon lands on its frame
    reconstruction.attributes["frames"] = params.output_frames
    data = expand_to_acquisition_frames(
        display_data_from_patato_obj(reconstruction),
        params.output_frames,
        params.source_frame_count,
    )

    # absolute placement: a recon comes from the raw time series, so it must not
    # inherit (and re-apply) the offset of whichever layer is currently active
    translate = (0.0,) * (data.ndim - 2) + (
        params.offset_z_mm,
        params.offset_x_mm,
    )

    layer_name = params.layer_name
    self._add_or_update_image_layer(
        layer_name,
        data,
        layer_metadata,
        reconstruction,
        colormap=layer_colormap(HDF5Tags.RECONSTRUCTION),
        translate=translate,
    )
    self.optari_controller.patato_objects[layer_name] = reconstruction
    self.optari_controller.derived_patato_objects[layer_name] = (
        reconstruction
    )
    self.optari_controller.ensure_shapes_layer_on_top()

    logger.info("reconstruction complete: %s", layer_name)
    return layer_name

refresh_ui

refresh_ui()

Refresh scan-dependent controls. Reconstruction runs off the loaded scan, not a layer. There is no "Source" to pick, so the only thing to reflect here is whether a scan is loaded at all.

Source code in src/optari/controllers/reconstruction_controller.py

def refresh_ui(self) -> None:
    """Refresh scan-dependent controls. Reconstruction runs off the loaded scan, not a
    layer. There is no "Source" to pick, so the only thing to reflect here is whether a
    scan is loaded at all."""
    dock = self.optari_controller.reconstruction
    pa_data = self.optari_controller.pa_data

    if pa_data is None:
        dock.status_label.setText(self._NO_SCAN_MESSAGE)
        dock.run_button.setEnabled(False)
        return

    if dock.status_label.text() == self._NO_SCAN_MESSAGE:
        dock.status_label.setText("Select a preset and run.")
    dock.run_button.setEnabled(not self.optari_controller.task_running)

weights_to_fetch

weights_to_fetch(params)

DeepMB needs its ONNX weights on disk before the preset can be built.

Source code in src/optari/controllers/reconstruction_controller.py

def weights_to_fetch(
    self, params: ReconParams
) -> tuple[Path, str | None] | None:
    """DeepMB needs its ONNX weights on disk before the preset can be built."""
    if params.algorithm_name != DEEPMB_ALGORITHM:
        return None
    extra = params.settings[ReconAttributeTags.ADDITIONAL_PARAMETERS]
    return resolve_model_path(extra["model_path"]), extra.get("model_url")

optari.controllers.unmixing_controller

Unmixing controller: presets and running PATATO's spectral unmixing (+ THb/sO2).

UnmixParams dataclass

UnmixParams(recon_for_run, source_layer_name, source_layer_metadata, source_frame_count, wavelengths, chromophores, output_frames, translate, reduce_factor=1, suffix='', generate_thb=False, generate_so2=False, current_frame_id=None)

Everything one unmixing run needs, resolved against a loaded scan.

frame_mode property

frame_mode

Whether the run covers the current frame or all frames.

name_stem property

name_stem

Base name for output layers, derived from the source recon layer, suffix and frame.

settings property

settings

Settings dict stored as metadata on output layers, e.g. for presets and export.

build classmethod

build(controller, *, wavelengths, chromophores, reduce_factor=1, suffix='', generate_thb=False, generate_so2=False, frame_id=None)

Resolve an unmixing setup against the active reconstruction layer.

Raises ValueError, message safe to show the user, when it cannot run here.

Source code in src/optari/controllers/unmixing_controller.py

@classmethod
def build(
    cls,
    controller,
    *,
    wavelengths: list[int],
    chromophores: list[str],
    reduce_factor: int = 1,
    suffix: str = "",
    generate_thb: bool = False,
    generate_so2: bool = False,
    frame_id: int | None = None,
) -> "UnmixParams":
    """Resolve an unmixing setup against the active reconstruction layer.

    Raises ValueError, message safe to show the user, when it cannot run here.
    """
    active_layer = controller.active_recon_layer
    if active_layer is None:
        raise ValueError("Select a PA reconstruction layer.")
    recon = controller.patato_objects.get(active_layer.name)
    if recon is None:
        raise ValueError("Source layer must be a reconstruction.")
    if not wavelengths:
        raise ValueError("Select at least one wavelength.")
    if not chromophores:
        raise ValueError("Select at least one chromophore.")
    validate_unmixing_spectra(wavelengths, chromophores)

    frame_numbers = list(
        active_layer.metadata.get("frames") or range(recon.shape[0])
    )
    recon_for_run = recon
    output_frames = frame_numbers
    if frame_id is not None:
        if frame_id not in frame_numbers:
            raise ValueError("Selected frame is not reconstructed.")
        # PATATO expects a contiguous slice, then we remap to acquisition frame ids.
        recon_idx = frame_numbers.index(frame_id)
        recon_for_run = recon[recon_idx : recon_idx + 1]
        output_frames = [frame_id]

    return cls(
        recon_for_run=recon_for_run,
        source_layer_name=active_layer.name,
        source_layer_metadata=dict(active_layer.metadata),
        source_frame_count=int(np.asarray(active_layer.data).shape[0]),
        # A coarse pixel averages a reduce_factor-wide block of the source's pixels
        # and belongs at the block's centre, (reduce_factor - 1) / 2 source pixels
        # past its first pixel.
        translate=tuple(active_layer.translate[:-2])
        + tuple(
            np.asarray(active_layer.translate[-2:])
            + (reduce_factor - 1) / 2 * np.asarray(active_layer.scale[-2:])
        ),
        wavelengths=wavelengths,
        chromophores=chromophores,
        output_frames=output_frames,
        reduce_factor=int(reduce_factor),
        suffix=suffix.strip(),
        generate_thb=bool(generate_thb),
        generate_so2=bool(generate_so2),
        current_frame_id=frame_id,
    )

from_preset classmethod

from_preset(preset, controller, *, frame_id=None)

Resolve an unmixing preset against the loaded scan's own wavelengths.

Source code in src/optari/controllers/unmixing_controller.py

@classmethod
def from_preset(
    cls, preset: dict, controller, *, frame_id: int | None = None
):
    """Resolve an unmixing preset against the loaded scan's own wavelengths."""
    active_layer = controller.active_recon_layer
    if active_layer is None:
        raise ValueError("Select a PA reconstruction layer.")
    available = [
        int(w) for w in active_layer.metadata.get("wavelengths") or []
    ]
    wavelengths = resolve_unmixing_wavelengths(preset, available)
    if not wavelengths:
        raise ValueError(
            "None of the preset's wavelengths were acquired in this scan "
            f"(scan has {available or 'none'})."
        )
    chromophores = list(preset.get(UnmixingAttributeTags.SPECTRA, []))
    # THb/sO2 need the haemoglobin pair, same rule the dock enforces.
    hb_pair = {"Hb", "HbO2"} <= set(chromophores)
    return cls.build(
        controller,
        wavelengths=wavelengths,
        chromophores=chromophores,
        reduce_factor=int(
            preset.get(UnmixingAttributeTags.RESOLUTION_REDUCE, 1)
        ),
        suffix=str(preset.get(UnmixingAttributeTags.SUFFIX, "")),
        generate_thb=hb_pair
        and bool(preset.get(UnmixingAttributeTags.COMPUTE_THB, True)),
        generate_so2=hb_pair
        and bool(preset.get(UnmixingAttributeTags.COMPUTE_SO2, True)),
        frame_id=frame_id,
    )

UnmixingController

UnmixingController(parent_controller)

Bases: TaskControllerBase

Run spectral unmixing and add as layers.

Initialize the unmixing preset store.

Source code in src/optari/controllers/unmixing_controller.py

def __init__(self, parent_controller):
    """Initialize the unmixing preset store."""
    super().__init__(parent_controller)
    self.preset_store = PresetStore(get_user_unmixing_presets_dir())

initialize_ui

initialize_ui()

Initialize preset, chromophore, and wavelength controls once.

Source code in src/optari/controllers/unmixing_controller.py

def initialize_ui(self) -> None:
    """Initialize preset, chromophore, and wavelength controls once."""
    dock = self.optari_controller.unmixing

    populate_preset_combo(dock.preset_combo, self.preset_store)

    if dock.chromophores_list.count() == 0:
        for name in sorted(SPECTRA_NAMES.keys()):
            add_checkable_item(dock.chromophores_list, name)

    self.refresh_ui()
    self.on_chromophores_changed()

on_chromophores_changed

on_chromophores_changed()

Enable THb and sO2 options only when Hb and HbO2 are selected. so2 is activated by default

Source code in src/optari/controllers/unmixing_controller.py

def on_chromophores_changed(self) -> None:
    """
    Enable THb and sO2 options only when Hb and HbO2 are selected.
    so2 is activated by default
    """
    dock = self.optari_controller.unmixing
    hb_pair_available = {"Hb", "HbO2"} <= set(self._checked_chromophores())

    dock.generate_so2_checkbox.setEnabled(hb_pair_available)
    dock.generate_thb_checkbox.setEnabled(hb_pair_available)
    if not hb_pair_available:
        dock.generate_so2_checkbox.setChecked(False)
        dock.generate_thb_checkbox.setChecked(False)

on_clear_wavelengths_clicked

on_clear_wavelengths_clicked()

Clear all source wavelength selections in the list widget.

Source code in src/optari/controllers/unmixing_controller.py

def on_clear_wavelengths_clicked(self) -> None:
    """Clear all source wavelength selections in the list widget."""
    set_checked(
        self.optari_controller.unmixing.wavelengths_list,
        lambda item: False,
    )

on_preset_changed

on_preset_changed()

Load selected preset values into the unmixing controls.

Source code in src/optari/controllers/unmixing_controller.py

def on_preset_changed(self) -> None:
    """Load selected preset values into the unmixing controls."""
    dock = self.optari_controller.unmixing
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        return

    try:
        settings = self.preset_store.load(preset_path)
        reduce_factor = int(
            settings.get(UnmixingAttributeTags.RESOLUTION_REDUCE, 1)
        )
        spectra = set(settings.get(UnmixingAttributeTags.SPECTRA, []))
        # Same resolution as a batch run, so the dock shows what a run would use.
        selected_wavelengths = set(
            resolve_unmixing_wavelengths(
                settings,
                [
                    item.data(Qt.UserRole)
                    for item in list_items(dock.wavelengths_list)
                ],
            )
        )

        compute_so2 = settings.get(UnmixingAttributeTags.COMPUTE_SO2, True)
        compute_thb = settings.get(UnmixingAttributeTags.COMPUTE_THB, True)
        suffix = str(settings.get(UnmixingAttributeTags.SUFFIX, ""))
    except (TypeError, ValueError, KeyError, OSError) as exc:
        dock.status_label.setText(f"Could not apply preset: {exc}")
        return

    dock.resolution_reduction_factor.setValue(max(1, reduce_factor))
    dock.suffix_edit.setText(suffix)
    set_checked(
        dock.chromophores_list, lambda item: item.text() in spectra
    )
    set_checked(
        dock.wavelengths_list,
        lambda item: item.data(Qt.UserRole) in selected_wavelengths,
    )
    dock.generate_so2_checkbox.setChecked(bool(compute_so2))
    dock.generate_thb_checkbox.setChecked(bool(compute_thb))
    self.on_chromophores_changed()

on_remove_preset_clicked

on_remove_preset_clicked()

Remove the selected unmixing preset.

Source code in src/optari/controllers/unmixing_controller.py

def on_remove_preset_clicked(self) -> None:
    """Remove the selected unmixing preset."""
    dock = self.optari_controller.unmixing
    preset_path = dock.preset_combo.currentData()
    if preset_path is None:
        dock.status_label.setText("Select a preset to remove.")
        return

    try:
        removed_path, removed = remove_selected_preset(
            dock.preset_combo, self.preset_store
        )
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not remove preset: {exc}")
        return
    if not removed:
        dock.status_label.setText(f"Preset not found: {preset_path.name}")
        return

    dock.status_label.setText(f"Removed preset: {removed_path.name}")

on_run_unmixing_clicked

on_run_unmixing_clicked()

Run unmixing from the dock's Run button.

Source code in src/optari/controllers/unmixing_controller.py

def on_run_unmixing_clicked(self) -> None:
    """Run unmixing from the dock's Run button."""
    self.run_from_ui(self.optari_controller.unmixing)

on_save_preset_clicked

on_save_preset_clicked()

Save the current unmixing controls as a new user preset.

Source code in src/optari/controllers/unmixing_controller.py

def on_save_preset_clicked(self) -> None:
    """Save the current unmixing controls as a new user preset."""
    dock = self.optari_controller.unmixing
    preset_settings = {
        UnmixingAttributeTags.RESOLUTION_REDUCE: dock.resolution_reduction_factor.value(),
        UnmixingAttributeTags.UNMIXING_WAVELENGTHS: self._checked_wavelengths(),
        UnmixingAttributeTags.SPECTRA: self._checked_chromophores(),
        UnmixingAttributeTags.COMPUTE_THB: dock.generate_thb_checkbox.isChecked(),
        UnmixingAttributeTags.COMPUTE_SO2: dock.generate_so2_checkbox.isChecked(),
        UnmixingAttributeTags.SUFFIX: dock.suffix_edit.text().strip(),
    }
    preset_name = prompt_preset_name(
        dock.widget, dock.preset_combo, "unmixing_preset"
    )
    if preset_name is None:
        return

    try:
        preset_path = self.preset_store.save(preset_name, preset_settings)
    except (ValueError, OSError) as exc:
        dock.status_label.setText(f"Could not save preset: {exc}")
        return

    add_preset_to_combo(dock.preset_combo, preset_path)
    dock.status_label.setText(f"Saved preset: {preset_path.name}")
    self.on_preset_changed()

on_select_all_wavelengths_clicked

on_select_all_wavelengths_clicked()

Select all source wavelengths in the list widget.

Source code in src/optari/controllers/unmixing_controller.py

def on_select_all_wavelengths_clicked(self) -> None:
    """Select all source wavelengths in the list widget."""
    set_checked(
        self.optari_controller.unmixing.wavelengths_list, lambda item: True
    )

prepare

prepare(params)

Return the work to run for params.

Source code in src/optari/controllers/unmixing_controller.py

def prepare(self, params: UnmixParams) -> BackgroundStep:
    """Return the work to run for *params*."""
    n_chunks = ceil(len(params.output_frames) / UNMIXING_CHUNK_FRAMES)
    logger.info(
        "running unmixing for %s with %s wavelength(s), %s chromophore(s), reduce=%s, "
        "in %s chunk(s) of %s",
        params.source_layer_name,
        len(params.wavelengths),
        len(params.chromophores),
        params.reduce_factor,
        n_chunks,
        UNMIXING_CHUNK_FRAMES,
    )
    return BackgroundStep(
        func=partial(
            _unmix_frames,
            params.recon_for_run,
            self.optari_controller.pa_data,
            params.wavelengths,
            params.chromophores,
            params.reduce_factor,
            params.suffix,
            params.generate_thb,
            params.generate_so2,
            UNMIXING_CHUNK_FRAMES,
        ),
        total=n_chunks,
        desc="Unmixing",
    )

publish

publish(result, params)

Add the finished unmixed/THb/sO2 layers. Runs on the main thread.

Source code in src/optari/controllers/unmixing_controller.py

def publish(self, result, params: UnmixParams) -> str:
    """Add the finished unmixed/THb/sO2 layers. Runs on the main thread."""
    unmixed, thb, so2 = result

    names = [
        self._publish_one(
            unmixed,
            params,
            prefix="Unmixed",
            # Channel labels are used by downstream spectrum displays.
            axis1_labels=list(map(str, unmixed.ax_1_labels)),
            pa_kind="unmixed",
            colormap=layer_colormap(HDF5Tags.UNMIXED),
            include_chromophores=True,
        )
    ]
    if thb is not None:
        names.append(
            self._publish_one(
                thb,
                params,
                prefix="THb",
                axis1_labels=["thb"],
                pa_kind="unmixed_param",
                colormap=layer_colormap(HDF5Tags.THB),
                parameter="thb",
            )
        )
    if so2 is not None:
        names.append(
            self._publish_one(
                so2,
                params,
                prefix="sO2",
                axis1_labels=["so2"],
                pa_kind="unmixed_param",
                colormap=layer_colormap(HDF5Tags.SO2),
                parameter="so2",
            )
        )

    # Reassert ROI visibility priority after adding multiple result layers.
    self.optari_controller.ensure_shapes_layer_on_top()
    logger.info("unmixing complete: %s", ", ".join(names))
    return ", ".join(names)

refresh_ui

refresh_ui()

Refresh source-dependent controls from the active layer.

Source code in src/optari/controllers/unmixing_controller.py

def refresh_ui(self) -> None:
    """Refresh source-dependent controls from the active layer."""
    dock = self.optari_controller.unmixing
    active_recon_layer = self.optari_controller.active_recon_layer

    if (
        active_recon_layer is None
        or active_recon_layer.metadata["pa_kind"] != "recon"
    ):
        dock.source_layer_label.setText("Select a PA reconstruction layer")
        dock.wavelengths_list.clear()
        # disable unmixing button when no valid source is active
        dock.run_button.setEnabled(False)
        return

    dock.source_layer_label.setText(active_recon_layer.name)

    # Enable unmixing button
    dock.run_button.setEnabled(not self.optari_controller.task_running)

    wavelengths = active_recon_layer.metadata.get("wavelengths") or []
    source_name = active_recon_layer.name
    last_source = dock.widget.property("_unmixing_source_name")

    # Keep manual wavelength selections while the same source stays active.
    if last_source == source_name and dock.wavelengths_list.count() > 0:
        return

    dock.widget.setProperty("_unmixing_source_name", source_name)
    dock.wavelengths_list.clear()
    for w in wavelengths:
        add_checkable_item(
            dock.wavelengths_list, f"{int(w)} nm", int(w), checked=True
        )

    self.on_preset_changed()

resolve_unmixing_wavelengths

resolve_unmixing_wavelengths(preset, available)

Which of a scan's available wavelengths a preset selects.

Presets name either an explicit list or an inclusive range, and both are resolved per scan rather than per preset: two scans in the same study can be acquired at different wavelengths, and a range must not invent one the scan does not have.

Source code in src/optari/controllers/unmixing_controller.py

def resolve_unmixing_wavelengths(
    preset: dict, available: list[int]
) -> list[int]:
    """Which of a scan's *available* wavelengths a preset selects.

    Presets name either an explicit list or an inclusive range, and both are resolved
    per scan rather than per preset: two scans in the same study can be acquired at
    different wavelengths, and a range must not invent one the scan does not have.
    """
    explicit = preset.get(UnmixingAttributeTags.UNMIXING_WAVELENGTHS)
    if explicit is not None:
        wanted = {int(w) for w in explicit}
        return sorted(w for w in available if w in wanted)

    wavelength_range = preset.get(UnmixingAttributeTags.WAVELENGTH_RANGE)
    if wavelength_range is not None and len(wavelength_range) == 2:
        start, end = int(wavelength_range[0]), int(wavelength_range[1])
        return sorted(w for w in available if start <= w <= end)

    return []

validate_unmixing_spectra

validate_unmixing_spectra(wavelengths, chromophores)

Raise a user-facing error when a spectrum has no values at a wavelength.

Source code in src/optari/controllers/unmixing_controller.py

def validate_unmixing_spectra(
    wavelengths: list[int], chromophores: list[str]
) -> None:
    """Raise a user-facing error when a spectrum has no values at a wavelength."""
    wavelength_array = np.asarray(wavelengths, dtype=float)
    missing = {
        chromophore: [
            wavelength
            for wavelength, value in zip(
                wavelengths,
                SPECTRA_NAMES[chromophore].get_spectrum(wavelength_array),
            )
            if not np.isfinite(value)
        ]
        for chromophore in chromophores
    }
    missing = {
        chromophore: values
        for chromophore, values in missing.items()
        if values
    }
    if missing:
        details = "; ".join(
            f"{chromophore}: {', '.join(f'{wavelength} nm' for wavelength in wavelengths)}"
            for chromophore, wavelengths in missing.items()
        )
        raise ValueError(f"No spectrum data is available for {details}.")

optari.controllers.analysis_controller

Analysis controller: time-series, histogram and spectral plots (pyqtgraph) over selected ROIs, and the scan's motion score per frame.

AnalysisController

AnalysisController(parent_controller)

Bases: TaskControllerBase

Temporal analysis, histograms, and spectral plotting helpers.

Source code in src/optari/controllers/base.py

def __init__(self, parent_controller: OptariController):
    """Initialize with reference to parent controller.

    Args:
        parent_controller: OptariController instance with viewer and session state.
    """
    self.optari_controller = parent_controller
    self.viewer = parent_controller.viewer

on_generate_time_analysis_clicked

on_generate_time_analysis_clicked(event=None)

Compute and plot the time series for the current Time Analysis scope.

Validates that ROIs and a multi-frame PA layer are selected, lazily creates the plot widget on first use, then redraws all series for either the selected ROIs or a single tracked ID.

Source code in src/optari/controllers/analysis_controller.py

def on_generate_time_analysis_clicked(self, event=None) -> None:
    """Compute and plot the time series for the current Time Analysis scope.

    Validates that ROIs and a multi-frame PA layer are selected, lazily creates the
    plot widget on first use, then redraws all series for either the selected ROIs
    or a single tracked ID.
    """
    annotation = self.optari_controller.annotation
    track_scope = annotation.time_analysis_track_radio.isChecked()

    error_msg = ""

    if self.optari_controller.shapes_layer is None:
        error_msg = "No ROIs layer"
    elif self.optari_controller.active_recon_layer is None:
        error_msg = "Select a PA image layer"
    elif (
        not track_scope
        and len(self.optari_controller.shapes_layer.data) == 0
    ):
        error_msg = "No ROIs defined"
    elif track_scope and not self.optari_controller.roi_ctrl.roi_records:
        error_msg = "No tracked ROIs available"
    elif (
        len(self.optari_controller.active_recon_layer.metadata["frames"])
        < 2
    ):
        error_msg = "PA image layer has less than 2 frames"

    if error_msg:
        self.optari_controller.time_analysis.status_label.setText(
            f'<span style="color:red">{error_msg}</span>'
        )
        if self.optari_controller.time_analysis.plot_widget is not None:
            self.optari_controller.time_analysis.plot_widget.clear()
        return

    if self.optari_controller.time_analysis.plot_widget is None:
        plot = pg.PlotWidget()
        plot.showGrid(x=True, y=True)
        plot.addLegend()
        self.optari_controller.time_analysis.plot_widget = plot
        layout = (
            self.optari_controller.time_analysis.plot_container.layout()
        )
        if layout is not None:
            layout.addWidget(plot)

    plot = self.optari_controller.time_analysis.plot_widget
    assert plot is not None

    _, channel_idx = self._current_frame_channel()
    feature_id = (
        self.optari_controller.annotation.time_analysis_feature_combo.currentData()
    )

    self.optari_controller.time_analysis.status_label.setText(
        "Computing time series…"
    )
    x, series = self._time_series_for_current_scope(
        channel_idx, feature_id
    )
    key_label = "Track" if track_scope else "ROI"

    if not series:
        self.optari_controller.time_analysis.status_label.setText(
            f'<span style="color:red">No data for the selected {key_label.lower()}.</span>'
        )
        if self.optari_controller.time_analysis.plot_widget is not None:
            self.optari_controller.time_analysis.plot_widget.clear()
        return

    plot.clear()
    plot.addLegend()
    plot.addItem(
        self.optari_controller.time_analysis.frame_line, ignoreBounds=True
    )

    for position, (key, y) in enumerate(series.items()):
        color = roi_color_for_index(position)
        plot.plot(
            x,
            y,
            pen=pg.mkPen(color=color, width=2),
            symbol="o",
            symbolSize=6,
            symbolBrush=pg.mkBrush(color),
            symbolPen=pg.mkPen(color=color, width=1),
            name=f"{key_label} {key}",
            connect="finite",  # NaN gaps (Track ID scope) must not be drawn over
        )

    vb = plot.getViewBox()
    vb.enableAutoRange(axis=vb.YAxis, enable=True)
    vb.autoRange(padding=0.02)

    (
        self.optari_controller.time_analysis.plotted_frames,
        self.optari_controller.time_analysis.plotted_x,
    ) = time_axis(self.optari_controller.active_recon_layer, channel_idx)
    self.update_frame_lines()

    # Same source as the x values themselves (roi_utils.time_axis).
    layer_metadata = self.optari_controller.active_recon_layer.metadata
    xlabel = (
        "Time (s)"
        if layer_metadata.get("timestamps") is not None
        else "Frame"
    )

    axis1_value = str(
        self.optari_controller.active_recon_layer.metadata.get(
            "axis1_labels"
        )[channel_idx]
    )

    plot.setLabel("bottom", xlabel)
    plot.setLabel("left", f"{feature_id} ({axis1_value})")

    self.optari_controller.time_analysis.status_label.setText(
        f"Plotted {len(series)} {key_label.lower()}(s) over {len(x)} frame(s) using {feature_id}."
    )

on_motion_frame_line_dragged

on_motion_frame_line_dragged(line)

Show the frame the user dragged the line to.

Source code in src/optari/controllers/analysis_controller.py

def on_motion_frame_line_dragged(self, line) -> None:
    """Show the frame the user dragged the line to."""
    last_frame = int(self.viewer.dims.nsteps[0]) - 1
    self.viewer.dims.set_point(
        0, int(np.clip(round(line.value()), 0, last_frame))
    )
    # Snap the line onto the frame actually shown, also when it did not change.
    self.update_frame_lines()

on_refresh_histograms_clicked

on_refresh_histograms_clicked(event=None)

Recompute and redraw per-ROI intensity histograms for the current frame and channel.

Source code in src/optari/controllers/analysis_controller.py

def on_refresh_histograms_clicked(self, event=None) -> None:
    """Recompute and redraw per-ROI intensity histograms for the current frame and channel."""
    if self.optari_controller.shapes_layer is None:
        self.optari_controller.histograms.status_label.setText(
            "No ROIs layer"
        )
        return
    if len(self.optari_controller.shapes_layer.data) == 0:
        self.optari_controller.histograms.status_label.setText(
            '<span style="color:red">No ROIs defined</span>'
        )
        self._clear_plot_layout(
            self.optari_controller.histograms.plots_container
        )
        return
    if self.optari_controller.active_recon_layer is None:
        self.optari_controller.histograms.status_label.setText(
            "Select a PA image layer"
        )
        return

    frame_idx, channel_idx = self._current_frame_channel()

    self.optari_controller.histograms.status_label.setText(
        "Computing histograms…"
    )

    roi_vals = extract_roi_pixels_for_slice(
        self.optari_controller.roi_ctrl.current_records(),
        self.optari_controller.active_recon_layer,
        frame_idx,
        channel_idx,
        clamp=self.optari_controller.roi_ctrl.intensity_clamp,
    )

    layout = self._clear_plot_layout(
        self.optari_controller.histograms.plots_container
    )

    n_plotted = 0
    for position, (roi_id, vals) in enumerate(roi_vals.items()):
        vals = np.asarray(vals)
        # Remove NaN values before computing histogram
        vals = vals[np.isfinite(vals)]
        if vals.size == 0:
            continue

        counts, edges = np.histogram(
            vals, bins=settings.analysis.histogram_bins
        )

        if counts.size == 0 or edges.size < 2:
            continue

        x = (edges[:-1] + edges[1:]) / 2.0
        width = float(edges[1] - edges[0])

        color = roi_color_for_index(position)
        brush = pg.mkBrush(color)
        pen = pg.mkPen(color)

        plot = pg.PlotWidget()
        plot.setTitle(f"ROI {roi_id}")
        plot.showGrid(x=True, y=True)
        bar = pg.BarGraphItem(
            x=x,
            height=counts,
            width=width,
            brush=brush,
            pen=pen,
        )
        plot.addItem(bar)

        if layout is not None:
            layout.addWidget(plot)
        n_plotted += 1

    self.optari_controller.histograms.status_label.setText(
        f"Plotted {n_plotted} histogram(s) for frame {frame_idx}, channel {channel_idx}."
    )

on_refresh_motion_clicked

on_refresh_motion_clicked()

Score the loaded scan's motion (once, then cached) and plot it.

Source code in src/optari/controllers/analysis_controller.py

def on_refresh_motion_clicked(self) -> None:
    """Score the loaded scan's motion (once, then cached) and plot it."""
    self.show_motion_scores(compute=True)

on_refresh_spectrum_clicked

on_refresh_spectrum_clicked(event=None)

Recompute and redraw per-ROI spectra across channels for the current frame.

Source code in src/optari/controllers/analysis_controller.py

def on_refresh_spectrum_clicked(self, event=None) -> None:
    """Recompute and redraw per-ROI spectra across channels for the current frame."""
    if self.optari_controller.shapes_layer is None:
        self.optari_controller.spectrum.status_label.setText(
            "No ROIs layer"
        )
        return
    if len(self.optari_controller.shapes_layer.data) == 0:
        self.optari_controller.spectrum.status_label.setText(
            '<span style="color:red">No ROIs defined</span>'
        )
        self._clear_plot_layout(
            self.optari_controller.spectrum.plots_container
        )
        return
    if self.optari_controller.active_recon_layer is None:
        self.optari_controller.spectrum.status_label.setText(
            "Select a PA image layer"
        )
        return

    frame_idx, _ = self._current_frame_channel()

    self.optari_controller.spectrum.status_label.setText(
        "Computing spectra…"
    )

    x, series, x_tick_labels = compute_roi_spectra(
        self.optari_controller.roi_ctrl.current_records(),
        self.optari_controller.active_recon_layer,
        frame_idx,
        clamp=self.optari_controller.roi_ctrl.intensity_clamp,
    )

    layout = self._clear_plot_layout(
        self.optari_controller.spectrum.plots_container
    )

    n_plotted = 0
    axis1_name = str(
        self.optari_controller.active_recon_layer.metadata.get(
            "axis1_name", "Channel"
        )
    )
    x_label = "Channel" if axis1_name.lower() == "channel" else axis1_name
    for position, (roi_id, y) in enumerate(series.items()):
        if y.size == 0 or np.all(np.isnan(y)):
            continue

        color = roi_color_for_index(position)

        plot = pg.PlotWidget()
        plot.setTitle(f"ROI {roi_id}")
        plot.showGrid(x=True, y=True)
        plot.setLabel("bottom", x_label)
        plot.setLabel("left", "Mean intensity")
        plot.plot(
            x,
            y,
            pen=pg.mkPen(color=color, width=2),
            symbol="o",
            symbolSize=6,
            symbolBrush=pg.mkBrush(color),
            symbolPen=pg.mkPen(color=color, width=1),
        )

        if x_tick_labels:
            tick_values = [
                (float(i), label) for i, label in enumerate(x_tick_labels)
            ]
            plot.getAxis("bottom").setTicks([tick_values])

        if layout is not None:
            layout.addWidget(plot)
        n_plotted += 1

    self.optari_controller.spectrum.status_label.setText(
        f"Plotted {n_plotted} spectra for frame {frame_idx}."
    )

on_time_analysis_scope_changed

on_time_analysis_scope_changed(checked)

When the user switches between "Selected ROI" and "Track ID" scopes, refresh the track-id combo, show only tracks that exist in the current ROI records.

Source code in src/optari/controllers/analysis_controller.py

def on_time_analysis_scope_changed(self, checked: bool) -> None:
    """When the user switches between "Selected ROI" and "Track ID" scopes,
    refresh the track-id combo, show only tracks that exist in the current
    ROI records.
    """
    if checked:
        self._refresh_time_analysis_track_combo()

on_time_frame_line_dragged

on_time_frame_line_dragged(line)

Show the plotted frame nearest to where the user dragged the line.

Source code in src/optari/controllers/analysis_controller.py

def on_time_frame_line_dragged(self, line) -> None:
    """Show the plotted frame nearest to where the user dragged the line."""
    time_analysis = self.optari_controller.time_analysis
    nearest = np.argmin(np.abs(time_analysis.plotted_x - line.value()))
    self.viewer.dims.set_point(
        0, int(time_analysis.plotted_frames[nearest])
    )
    self.update_frame_lines()

refresh_ui

refresh_ui()

Refresh gating for the histogram/spectrum/time-analysis buttons.

Source code in src/optari/controllers/analysis_controller.py

def refresh_ui(self) -> None:
    """Refresh gating for the histogram/spectrum/time-analysis buttons."""
    has_data = (
        self.optari_controller.shapes_layer is not None
        and self.optari_controller.active_recon_layer is not None
    )
    self.optari_controller.time_analysis.generate_button.setEnabled(
        has_data
    )
    self.optari_controller.histograms.refresh_button.setEnabled(has_data)
    self.optari_controller.spectrum.refresh_button.setEnabled(has_data)
    self.optari_controller.motion.refresh_button.setEnabled(
        self.optari_controller.active_us_layer is not None
    )

show_motion_scores

show_motion_scores(*, compute)

Plot both motion scores per frame of the loaded scan, or say why there are none.

Without compute only already cached scores are plotted, so a scan load can refresh the dock for free when a motion-based default frame scored it anyway.

Source code in src/optari/controllers/analysis_controller.py

def show_motion_scores(self, *, compute: bool) -> None:
    """Plot both motion scores per frame of the loaded scan, or say why there are none.

    Without *compute* only already cached scores are plotted, so a scan load can
    refresh the dock for free when a motion-based default frame scored it anyway.
    """
    dock = self.optari_controller.motion
    scores = self.optari_controller.scan_ctrl.motion_scores(
        compute=compute
    )
    if scores is None:
        dock.spectral_curve.setData([], [])
        dock.frame_curve.setData([], [])
        dock.stillest.setData([], [])
        dock.frame_line.hide()
        if self.optari_controller.pa_data is None:
            dock.status_label.setText("Load a scan to score its motion.")
        elif self.optari_controller.active_us_layer is None:
            dock.status_label.setText(
                "This scan has no ultrasound to score motion on."
            )
        else:
            dock.status_label.setText(
                "Click 'Refresh' to score motion per frame."
            )
        return

    spectral, frame = scores["spectral_motion"], scores["frame_motion"]
    stillest_frame = int(np.argmin(frame))
    dock.frame_curve.setData(np.arange(len(frame)), frame)
    if spectral is None:
        dock.spectral_curve.setData([], [])
        dock.stillest.setData([stillest_frame], [frame[stillest_frame]])
        dock.status_label.setText(
            "One image per wavelength set, so no spectral motion. Lowest "
            f"frame-to-frame motion: frame {stillest_frame} (green)."
        )
    else:
        stillest_spectral = int(np.argmin(spectral))
        dock.spectral_curve.setData(np.arange(len(spectral)), spectral)
        dock.stillest.setData(
            [stillest_spectral, stillest_frame],
            [spectral[stillest_spectral], frame[stillest_frame]],
        )
        dock.status_label.setText(
            f"Lowest motion: frame {stillest_spectral} spectral, frame "
            f"{stillest_frame} frame-to-frame (green)."
        )
    dock.frame_line.show()
    self.update_frame_lines()

update_frame_lines

update_frame_lines()

Move the Motion and Temporal frame lines to the viewer's current frame.

Source code in src/optari/controllers/analysis_controller.py

def update_frame_lines(self) -> None:
    """Move the Motion and Temporal frame lines to the viewer's current frame."""
    frame_channel = selected_frame_and_channel(self.viewer)
    if frame_channel is None:
        return
    frame = frame_channel[0]
    self.optari_controller.motion.frame_line.setValue(frame)

    time_analysis = self.optari_controller.time_analysis
    plotted = np.flatnonzero(time_analysis.plotted_frames == frame)
    # A frame the plot has no point for (not in the plotted layer) gets no line.
    time_analysis.frame_line.setVisible(plotted.size > 0)
    if plotted.size:
        time_analysis.frame_line.setValue(
            time_analysis.plotted_x[plotted[0]]
        )

optari.controllers.viewer_export_controller

Viewer export controller.

ViewerExportController

Export the currently visible viewer content as image or video.

on_export_clicked staticmethod

on_export_clicked(controller)

Open the export dialog and write the chosen image or video to disk.

Source code in src/optari/controllers/viewer_export_controller.py

@staticmethod
def on_export_clicked(controller) -> None:
    """Open the export dialog and write the chosen image or video to disk."""
    viewer = controller.viewer
    parent_widget = viewer.window._qt_window
    active_layer = viewer.layers.selection.active
    n_frames = viewer.dims.nsteps[0] if viewer.dims.ndim > 0 else 1
    dialog = ViewerExportDialog(
        parent_widget, video_available=n_frames > 1
    )
    if dialog.exec() != ViewerExportDialog.Accepted:
        return

    dialog_settings = dialog.get_dialog_export_settings()
    default_name = (
        f"OPTARIVIEW_{active_layer.name if active_layer else 'viewer'}"
    )
    if not dialog_settings["video"]:
        frame_id = (
            viewer.dims.current_step[0] if viewer.dims.ndim > 0 else 0
        )
        default_name += f"_F{frame_id}"
    channel_id = viewer.dims.current_step[1] if viewer.dims.ndim > 1 else 0
    default_name += f"_C{channel_id}"

    if dialog_settings["video"]:
        ViewerExportController._export_video(
            viewer, parent_widget, default_name, n_frames, dialog_settings
        )
        return

    ViewerExportController._export_image(
        viewer,
        parent_widget,
        default_name,
        dialog_settings["include_colorbars"],
    )