Velocity analysis

Bases: BaseAnalysis

Measure velocity from one channel/ROI using a Radon transform.

The analysis runs on a full-resolution rectangular ROI crop. For line-scan kymographs, rows correspond to time and columns correspond to distance. The Radon core analyzes sliding windows along the time axis and reports velocity as a function of time.

Detection parameters are serialized with the analysis and affect scientific results. Execution options set with :meth:set_execution_options control multiprocessing for speed and are not serialized.

Examples:

Create and run one analysis through an AcqImage analysis set::

key = acq.analysis_set.create(
    "radon_velocity",
    channel=0,
    roi_id=1,
    detection_params={"window_width": 64},
).key
acq.analysis_set.run_analysis(key)
plot = acq.analysis_set.get(key).get_plot_data()

Parameters:

Name	Type	Description	Default
channel	int	Zero-based channel index for analysis.	required
roi_id	int	Rectangular ROI identifier for analysis.	required
detection_params	dict[str, object] | None	Optional detection parameters. Missing values are filled from detection_schema defaults.	None

Source code in src/acqstore/acq_image/analysis/velocity_analysis/radon_velocity_analysis.py

@register_analysis_class
class RadonVelocityAnalysis(BaseAnalysis):
    """Measure velocity from one channel/ROI using a Radon transform.

    The analysis runs on a full-resolution rectangular ROI crop. For line-scan
    kymographs, rows correspond to time and columns correspond to distance. The
    Radon core analyzes sliding windows along the time axis and reports velocity
    as a function of time.

    Detection parameters are serialized with the analysis and affect scientific
    results. Execution options set with :meth:`set_execution_options` control
    multiprocessing for speed and are not serialized.

    Examples:
        Create and run one analysis through an ``AcqImage`` analysis set::

            key = acq.analysis_set.create(
                "radon_velocity",
                channel=0,
                roi_id=1,
                detection_params={"window_width": 64},
            ).key
            acq.analysis_set.run_analysis(key)
            plot = acq.analysis_set.get(key).get_plot_data()

    Args:
        channel: Zero-based channel index for analysis.
        roi_id: Rectangular ROI identifier for analysis.
        detection_params: Optional detection parameters. Missing values are
            filled from ``detection_schema`` defaults.
    """

    def __init__(
        self,
        *,
        channel: int,
        roi_id: int,
        detection_params: dict[str, object] | None = None,
    ) -> None:
        """Create a Radon velocity analysis instance.

        Args:
            channel: Channel index for analysis.
            roi_id: ROI identifier for analysis.
            detection_params: Optional detection parameters.
        """
        super().__init__(
            channel=channel,
            roi_id=roi_id,
            detection_params=detection_params,
        )
        self._use_multiprocessing = True
        self._processes: int | None = None

    def set_execution_options(
        self,
        *,
        use_multiprocessing: bool = True,
        processes: int | None = None,
    ) -> None:
        """Set runtime execution options for the next run.

        These options are not detection parameters and are not serialized. They
        control how the current Python process executes the expensive Radon
        computation.

        Args:
            use_multiprocessing: Whether to use multiprocessing for Radon
                windows.
            processes: Optional number of worker processes. ``None`` lets the
                core algorithm choose a CPU-count based default.

        Returns:
            None.
        """
        self._use_multiprocessing = bool(use_multiprocessing)
        self._processes = None if processes is None else int(processes)

    analysis_name = "radon_velocity"
    exclusive_group = "primary_kymograph"
    detection_schema = (
        DetectionParamSchema(
            name="window_width",
            display_name="Window Width",
            value_type=DetectionValueType.INT,
            default=64,
            description="Number of time samples per Radon analysis window.",
            choices=(16, 64, 128),
            visible=True,
            editable=True,
        ),
    )

    def run(
        self,
        data_provider: AnalysisDataProvider,
        *,
        context: AnalysisRunContext | None = None,
        dependencies: dict[str, BaseAnalysis] | None = None,
    ) -> AnalysisResult:
        """Run Radon velocity analysis on one ROI crop.

        Args:
            data_provider: Provider for full-resolution ROI image data and
                physical spacing. ``get_roi_image`` must return a 2D ``(Y, X)``
                array; for kymographs this is ``(time, space)``.
            context: Optional progress/cancellation context.
            dependencies: Dependency analyses. Radon velocity currently does not
                require dependencies.

        Returns:
            Analysis result populated with a summary dictionary and table. The
            table includes at least ``time_s`` and ``velocity`` when successful.

        Raises:
            AnalysisCancelled: If the run is cancelled through ``context``.
        """
        context = context or AnalysisRunContext()
        context.raise_if_cancelled()
        image = data_provider.get_roi_image(channel=self.key.channel, roi_id=self.key.roi_id)
        physical_units = data_provider.get_image_physical_units()
        window_width = int(self.detection_params["window_width"])

        try:
            result = run_radon_velocity(
                image,
                window_width=window_width,
                physical_units=physical_units,
                progress_callback=context.report_progress,
                cancel_callback=context.is_cancelled,
                use_multiprocessing=self._use_multiprocessing,
                processes=self._processes,
            )
        except RadonVelocityCancelled as exc:
            raise AnalysisCancelled(str(exc)) from exc

        self.result.summary = result.summary
        self.result.table = result.table
        self.set_dirty()
        return self.result

    def get_plot_data(self) -> AnalysisPlotData | None:
        """Return canonical velocity-versus-time plot data.

        Returns:
            Plot data using ``time_s`` for the x axis and ``velocity`` for the y
            axis, or ``None`` when the analysis has no table output yet.

        Raises:
            KeyError: If the table is present but missing required columns.
        """
        if self.result.table is None:
            return None
        table = self.result.table
        if "time_s" not in table.columns:
            raise KeyError("Radon velocity plot requires 'time_s' column")
        if "velocity" not in table.columns:
            raise KeyError("Radon velocity plot requires 'velocity' column")
        return AnalysisPlotData(
            x=tuple(float(value) for value in table["time_s"].tolist()),
            y=tuple(float(value) for value in table["velocity"].tolist()),
            x_label="Time (s)",
            y_label="Velocity",
            series_name="Radon velocity",
        )

set_execution_options

set_execution_options(
    *,
    use_multiprocessing: bool = True,
    processes: int | None = None,
) -> None

Set runtime execution options for the next run.

These options are not detection parameters and are not serialized. They control how the current Python process executes the expensive Radon computation.

Parameters:

Name	Type	Description	Default
use_multiprocessing	bool	Whether to use multiprocessing for Radon windows.	True
processes	int | None	Optional number of worker processes. None lets the core algorithm choose a CPU-count based default.	None

Returns:

Type	Description
None	None.

Source code in src/acqstore/acq_image/analysis/velocity_analysis/radon_velocity_analysis.py

def set_execution_options(
    self,
    *,
    use_multiprocessing: bool = True,
    processes: int | None = None,
) -> None:
    """Set runtime execution options for the next run.

    These options are not detection parameters and are not serialized. They
    control how the current Python process executes the expensive Radon
    computation.

    Args:
        use_multiprocessing: Whether to use multiprocessing for Radon
            windows.
        processes: Optional number of worker processes. ``None`` lets the
            core algorithm choose a CPU-count based default.

    Returns:
        None.
    """
    self._use_multiprocessing = bool(use_multiprocessing)
    self._processes = None if processes is None else int(processes)

run

run(
    data_provider: AnalysisDataProvider,
    *,
    context: AnalysisRunContext | None = None,
    dependencies: dict[str, BaseAnalysis] | None = None,
) -> AnalysisResult

Run Radon velocity analysis on one ROI crop.

Parameters:

Name	Type	Description	Default
data_provider	AnalysisDataProvider	Provider for full-resolution ROI image data and physical spacing. get_roi_image must return a 2D (Y, X) array; for kymographs this is (time, space).	required
context	AnalysisRunContext | None	Optional progress/cancellation context.	None
dependencies	dict[str, BaseAnalysis] | None	Dependency analyses. Radon velocity currently does not require dependencies.	None

Returns:

Type	Description
AnalysisResult	Analysis result populated with a summary dictionary and table. The
AnalysisResult	table includes at least time_s and velocity when successful.

Raises:

Type	Description
AnalysisCancelled	If the run is cancelled through context.

Source code in src/acqstore/acq_image/analysis/velocity_analysis/radon_velocity_analysis.py

def run(
    self,
    data_provider: AnalysisDataProvider,
    *,
    context: AnalysisRunContext | None = None,
    dependencies: dict[str, BaseAnalysis] | None = None,
) -> AnalysisResult:
    """Run Radon velocity analysis on one ROI crop.

    Args:
        data_provider: Provider for full-resolution ROI image data and
            physical spacing. ``get_roi_image`` must return a 2D ``(Y, X)``
            array; for kymographs this is ``(time, space)``.
        context: Optional progress/cancellation context.
        dependencies: Dependency analyses. Radon velocity currently does not
            require dependencies.

    Returns:
        Analysis result populated with a summary dictionary and table. The
        table includes at least ``time_s`` and ``velocity`` when successful.

    Raises:
        AnalysisCancelled: If the run is cancelled through ``context``.
    """
    context = context or AnalysisRunContext()
    context.raise_if_cancelled()
    image = data_provider.get_roi_image(channel=self.key.channel, roi_id=self.key.roi_id)
    physical_units = data_provider.get_image_physical_units()
    window_width = int(self.detection_params["window_width"])

    try:
        result = run_radon_velocity(
            image,
            window_width=window_width,
            physical_units=physical_units,
            progress_callback=context.report_progress,
            cancel_callback=context.is_cancelled,
            use_multiprocessing=self._use_multiprocessing,
            processes=self._processes,
        )
    except RadonVelocityCancelled as exc:
        raise AnalysisCancelled(str(exc)) from exc

    self.result.summary = result.summary
    self.result.table = result.table
    self.set_dirty()
    return self.result

get_plot_data

get_plot_data() -> AnalysisPlotData | None

Return canonical velocity-versus-time plot data.

Returns:

Type	Description
AnalysisPlotData | None	Plot data using time_s for the x axis and velocity for the y
AnalysisPlotData | None	axis, or None when the analysis has no table output yet.

Raises:

Type	Description
KeyError	If the table is present but missing required columns.

Source code in src/acqstore/acq_image/analysis/velocity_analysis/radon_velocity_analysis.py

def get_plot_data(self) -> AnalysisPlotData | None:
    """Return canonical velocity-versus-time plot data.

    Returns:
        Plot data using ``time_s`` for the x axis and ``velocity`` for the y
        axis, or ``None`` when the analysis has no table output yet.

    Raises:
        KeyError: If the table is present but missing required columns.
    """
    if self.result.table is None:
        return None
    table = self.result.table
    if "time_s" not in table.columns:
        raise KeyError("Radon velocity plot requires 'time_s' column")
    if "velocity" not in table.columns:
        raise KeyError("Radon velocity plot requires 'velocity' column")
    return AnalysisPlotData(
        x=tuple(float(value) for value in table["time_s"].tolist()),
        y=tuple(float(value) for value in table["velocity"].tolist()),
        x_label="Time (s)",
        y_label="Velocity",
        series_name="Radon velocity",
    )