Event analysis

Bases: BaseAnalysis

Event interval analysis dependent on Radon velocity plot data.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

@register_analysis_class
class EventAnalysis(BaseAnalysis):
    """Event interval analysis dependent on Radon velocity plot data."""

    analysis_name = "event"
    analysis_version = EVENT_SUMMARY_VERSION
    summary_columns = (
        "analysis_date",
        "analysis_time",
        "analysis_version",
        "version",
        "parent_analysis_name",
        "num_events",
        "user_events",
        "rise_events",
        "fall_events",
        "transient_events",
        "mean_duration",
    )
    exclusive_group = None
    depends_on = (RADON_VELOCITY_ANALYSIS_NAME,)
    detection_schema = (
        DetectionParamSchema(
            name="pre_post_win_sec",
            display_name="Pre/post window",
            value_type=DetectionValueType.FLOAT,
            default=DEFAULT_PRE_POST_WIN_SEC,
            description="Window width used before and after each event for derived statistics.",
            unit="s",
        ),
    )

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

        Args:
            channel: Channel index.
            roi_id: ROI identifier.
            detection_params: Optional detection parameters.
        """
        super().__init__(channel=channel, roi_id=roi_id, detection_params=detection_params)
        self.events = AcqImageEventStore()
        self._sync_summary()

    @property
    def pre_post_win_sec(self) -> float:
        """Return the configured pre/post stats window in seconds."""
        value = float(self.detection_params["pre_post_win_sec"])
        if value < 0:
            raise ValueError("pre_post_win_sec must be >= 0")
        return value

    def set_detection_params(self, detection_params: dict[str, object]) -> None:
        """Replace detection parameters and mark this analysis dirty.

        Args:
            detection_params: New detection parameter mapping.
        """
        params = self.get_default_detection_params()
        self.validate_detection_params(detection_params)
        params.update(detection_params)
        self.detection_params = params
        _ = self.pre_post_win_sec
        self.set_dirty()

    def run(
        self,
        data_provider: AnalysisDataProvider,
        *,
        context: AnalysisRunContext | None = None,
        dependencies: dict[str, BaseAnalysis] | None = None,
    ) -> AnalysisResult:
        """Recalculate event stats from the required parent analysis.

        Args:
            data_provider: Unused analysis data provider.
            context: Optional run context.
            dependencies: Dependency mapping containing ``radon_velocity``.

        Returns:
            Current analysis result with recalculated event statistics.

        Raises:
            ValueError: If dependency data is missing or has no plot data.
        """
        _ = data_provider
        if context is not None:
            context.report_progress(0.0, "Preparing event analysis")
            context.raise_if_cancelled()
        plot_data = self._required_parent_plot_data(dependencies)
        self.events.reanalyze_events(plot_data, self.pre_post_win_sec)
        self._sync_summary()
        self.result.summary = self.finalize_summary(self.result.summary)
        self.set_dirty()
        if context is not None:
            context.report_progress(1.0, "Event analysis complete")
        return self.result

    def add_event(
        self,
        x0: float,
        x1: float,
        *,
        plot_data: AnalysisPlotData,
        event_type: EventType = EventType.USER,
        event_id: int | None = None,
    ) -> AcqImageEvent:
        """Add one event and mark analysis dirty.

        Args:
            x0: First x-coordinate.
            x1: Second x-coordinate.
            plot_data: Parent plot data used to calculate stats.
            event_type: Event category.
            event_id: Optional caller-supplied id.

        Returns:
            Created event.
        """
        event = self.events.add_event(
            x0,
            x1,
            plot_data=plot_data,
            pre_post_win_sec=self.pre_post_win_sec,
            event_type=event_type,
            event_id=event_id,
        )
        self._sync_summary()
        self.set_dirty()
        return event

    def add_events(self, events: list[AcqImageEvent]) -> list[AcqImageEvent]:
        """Add multiple already-calculated events and mark analysis dirty.

        Args:
            events: Events to add.

        Returns:
            Added events.
        """
        added = self.events.add_events(events)
        self._sync_summary()
        self.set_dirty()
        return added

    def delete_event(self, event_id: int) -> None:
        """Delete one event and mark analysis dirty.

        Args:
            event_id: Event id to delete.
        """
        self.events.delete_event(event_id)
        self._sync_summary()
        self.set_dirty()

    def update_event(
        self,
        event_id: int,
        *,
        plot_data: AnalysisPlotData,
        x0: float | None = None,
        x1: float | None = None,
        event_type: EventType | None = None,
    ) -> AcqImageEvent:
        """Update one event and mark analysis dirty.

        Args:
            event_id: Event id.
            plot_data: Parent plot data used to calculate stats.
            x0: Optional replacement x0.
            x1: Optional replacement x1.
            event_type: Optional replacement category.

        Returns:
            Updated event.
        """
        event = self.events.update_event(
            event_id,
            plot_data=plot_data,
            pre_post_win_sec=self.pre_post_win_sec,
            x0=x0,
            x1=x1,
            event_type=event_type,
        )
        self._sync_summary()
        self.set_dirty()
        return event

    def get_events(self) -> list[AcqImageEvent]:
        """Return current events.

        Returns:
            List of events sorted by id.
        """
        return self.events.get_events()

    def get_summary_values(self) -> dict[str, object]:
        """Return flat event summary values for analysis pools.

        Returns:
            Mapping with exactly the keys declared in
            :attr:`summary_columns`. The nested event records are summarized as
            scalar counts and mean event duration.
        """
        summary = self.result.summary
        records = summary.get("events")
        events = records if isinstance(records, list) else []
        type_counts = {event_type.value: 0 for event_type in EventType}
        durations: list[float] = []
        for record in events:
            if not isinstance(record, dict):
                continue
            event_type = str(record.get("event_type", ""))
            if event_type in type_counts:
                type_counts[event_type] += 1
            duration = record.get("duration")
            if duration is not None:
                try:
                    durations.append(float(duration))
                except (TypeError, ValueError):
                    continue

        values: dict[str, object] = {
            "analysis_date": summary.get("analysis_date", pd.NA),
            "analysis_time": summary.get("analysis_time", pd.NA),
            "analysis_version": summary.get("analysis_version", pd.NA),
            "version": summary.get("version", pd.NA),
            "parent_analysis_name": summary.get("parent_analysis_name", pd.NA),
            "num_events": len(events),
            "user_events": type_counts[EventType.USER.value],
            "rise_events": type_counts[EventType.RISE.value],
            "fall_events": type_counts[EventType.FALL.value],
            "transient_events": type_counts[EventType.TRANSIENT.value],
            "mean_duration": (sum(durations) / len(durations)) if durations else pd.NA,
        }
        return {key: values.get(key, pd.NA) for key in self.get_summary_columns()}

    def load_json_dict(self, record: dict[str, Any]) -> None:
        """Load event analysis state from a JSON analysis record.

        Args:
            record: Analysis sidecar record.
        """
        detection_params = dict(record["detection_params"])
        self.set_detection_params(detection_params)
        self.result.summary = dict(record["summary"])
        self.events = AcqImageEventStore.from_summary_dict(self.result.summary)
        self._sync_summary()
        self.set_clean()

    def _sync_summary(self) -> None:
        """Synchronize ``AnalysisResult.summary`` with the event store."""
        self.result.summary = self.events.to_summary_dict()

    @staticmethod
    def _required_parent_plot_data(dependencies: dict[str, BaseAnalysis] | None) -> AnalysisPlotData:
        """Return required parent plot data or raise.

        Args:
            dependencies: Analysis dependencies keyed by analysis name.

        Returns:
            Parent analysis plot data.

        Raises:
            ValueError: If the required dependency or its plot data is missing.
        """
        if dependencies is None or RADON_VELOCITY_ANALYSIS_NAME not in dependencies:
            raise ValueError("Event analysis requires radon_velocity dependency")
        parent = dependencies[RADON_VELOCITY_ANALYSIS_NAME]
        plot_data = parent.get_plot_data()
        if plot_data is None:
            raise ValueError("Event analysis requires radon_velocity plot data")
        return plot_data

pre_post_win_sec property

pre_post_win_sec: float

Return the configured pre/post stats window in seconds.

set_detection_params

set_detection_params(
    detection_params: dict[str, object],
) -> None

Replace detection parameters and mark this analysis dirty.

Parameters:

Name	Type	Description	Default
detection_params	dict[str, object]	New detection parameter mapping.	required

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def set_detection_params(self, detection_params: dict[str, object]) -> None:
    """Replace detection parameters and mark this analysis dirty.

    Args:
        detection_params: New detection parameter mapping.
    """
    params = self.get_default_detection_params()
    self.validate_detection_params(detection_params)
    params.update(detection_params)
    self.detection_params = params
    _ = self.pre_post_win_sec
    self.set_dirty()

run

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

Recalculate event stats from the required parent analysis.

Parameters:

Name	Type	Description	Default
data_provider	AnalysisDataProvider	Unused analysis data provider.	required
context	AnalysisRunContext | None	Optional run context.	None
dependencies	dict[str, BaseAnalysis] | None	Dependency mapping containing radon_velocity.	None

Returns:

Type	Description
AnalysisResult	Current analysis result with recalculated event statistics.

Raises:

Type	Description
ValueError	If dependency data is missing or has no plot data.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def run(
    self,
    data_provider: AnalysisDataProvider,
    *,
    context: AnalysisRunContext | None = None,
    dependencies: dict[str, BaseAnalysis] | None = None,
) -> AnalysisResult:
    """Recalculate event stats from the required parent analysis.

    Args:
        data_provider: Unused analysis data provider.
        context: Optional run context.
        dependencies: Dependency mapping containing ``radon_velocity``.

    Returns:
        Current analysis result with recalculated event statistics.

    Raises:
        ValueError: If dependency data is missing or has no plot data.
    """
    _ = data_provider
    if context is not None:
        context.report_progress(0.0, "Preparing event analysis")
        context.raise_if_cancelled()
    plot_data = self._required_parent_plot_data(dependencies)
    self.events.reanalyze_events(plot_data, self.pre_post_win_sec)
    self._sync_summary()
    self.result.summary = self.finalize_summary(self.result.summary)
    self.set_dirty()
    if context is not None:
        context.report_progress(1.0, "Event analysis complete")
    return self.result

add_event

add_event(
    x0: float,
    x1: float,
    *,
    plot_data: AnalysisPlotData,
    event_type: EventType = EventType.USER,
    event_id: int | None = None,
) -> AcqImageEvent

Add one event and mark analysis dirty.

Parameters:

Name	Type	Description	Default
x0	float	First x-coordinate.	required
x1	float	Second x-coordinate.	required
plot_data	AnalysisPlotData	Parent plot data used to calculate stats.	required
event_type	EventType	Event category.	USER
event_id	int | None	Optional caller-supplied id.	None

Returns:

Type	Description
AcqImageEvent	Created event.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def add_event(
    self,
    x0: float,
    x1: float,
    *,
    plot_data: AnalysisPlotData,
    event_type: EventType = EventType.USER,
    event_id: int | None = None,
) -> AcqImageEvent:
    """Add one event and mark analysis dirty.

    Args:
        x0: First x-coordinate.
        x1: Second x-coordinate.
        plot_data: Parent plot data used to calculate stats.
        event_type: Event category.
        event_id: Optional caller-supplied id.

    Returns:
        Created event.
    """
    event = self.events.add_event(
        x0,
        x1,
        plot_data=plot_data,
        pre_post_win_sec=self.pre_post_win_sec,
        event_type=event_type,
        event_id=event_id,
    )
    self._sync_summary()
    self.set_dirty()
    return event

add_events

add_events(
    events: list[AcqImageEvent],
) -> list[AcqImageEvent]

Add multiple already-calculated events and mark analysis dirty.

Parameters:

Name	Type	Description	Default
events	list[AcqImageEvent]	Events to add.	required

Returns:

Type	Description
list[AcqImageEvent]	Added events.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def add_events(self, events: list[AcqImageEvent]) -> list[AcqImageEvent]:
    """Add multiple already-calculated events and mark analysis dirty.

    Args:
        events: Events to add.

    Returns:
        Added events.
    """
    added = self.events.add_events(events)
    self._sync_summary()
    self.set_dirty()
    return added

delete_event

delete_event(event_id: int) -> None

Delete one event and mark analysis dirty.

Parameters:

Name	Type	Description	Default
event_id	int	Event id to delete.	required

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def delete_event(self, event_id: int) -> None:
    """Delete one event and mark analysis dirty.

    Args:
        event_id: Event id to delete.
    """
    self.events.delete_event(event_id)
    self._sync_summary()
    self.set_dirty()

update_event

update_event(
    event_id: int,
    *,
    plot_data: AnalysisPlotData,
    x0: float | None = None,
    x1: float | None = None,
    event_type: EventType | None = None,
) -> AcqImageEvent

Update one event and mark analysis dirty.

Parameters:

Name	Type	Description	Default
event_id	int	Event id.	required
plot_data	AnalysisPlotData	Parent plot data used to calculate stats.	required
x0	float | None	Optional replacement x0.	None
x1	float | None	Optional replacement x1.	None
event_type	EventType | None	Optional replacement category.	None

Returns:

Type	Description
AcqImageEvent	Updated event.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def update_event(
    self,
    event_id: int,
    *,
    plot_data: AnalysisPlotData,
    x0: float | None = None,
    x1: float | None = None,
    event_type: EventType | None = None,
) -> AcqImageEvent:
    """Update one event and mark analysis dirty.

    Args:
        event_id: Event id.
        plot_data: Parent plot data used to calculate stats.
        x0: Optional replacement x0.
        x1: Optional replacement x1.
        event_type: Optional replacement category.

    Returns:
        Updated event.
    """
    event = self.events.update_event(
        event_id,
        plot_data=plot_data,
        pre_post_win_sec=self.pre_post_win_sec,
        x0=x0,
        x1=x1,
        event_type=event_type,
    )
    self._sync_summary()
    self.set_dirty()
    return event

get_events

get_events() -> list[AcqImageEvent]

Return current events.

Returns:

Type	Description
list[AcqImageEvent]	List of events sorted by id.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def get_events(self) -> list[AcqImageEvent]:
    """Return current events.

    Returns:
        List of events sorted by id.
    """
    return self.events.get_events()

get_summary_values

get_summary_values() -> dict[str, object]

Return flat event summary values for analysis pools.

Returns:

Type	Description
dict[str, object]	Mapping with exactly the keys declared in
dict[str, object]	attr:summary_columns. The nested event records are summarized as
dict[str, object]	scalar counts and mean event duration.

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def get_summary_values(self) -> dict[str, object]:
    """Return flat event summary values for analysis pools.

    Returns:
        Mapping with exactly the keys declared in
        :attr:`summary_columns`. The nested event records are summarized as
        scalar counts and mean event duration.
    """
    summary = self.result.summary
    records = summary.get("events")
    events = records if isinstance(records, list) else []
    type_counts = {event_type.value: 0 for event_type in EventType}
    durations: list[float] = []
    for record in events:
        if not isinstance(record, dict):
            continue
        event_type = str(record.get("event_type", ""))
        if event_type in type_counts:
            type_counts[event_type] += 1
        duration = record.get("duration")
        if duration is not None:
            try:
                durations.append(float(duration))
            except (TypeError, ValueError):
                continue

    values: dict[str, object] = {
        "analysis_date": summary.get("analysis_date", pd.NA),
        "analysis_time": summary.get("analysis_time", pd.NA),
        "analysis_version": summary.get("analysis_version", pd.NA),
        "version": summary.get("version", pd.NA),
        "parent_analysis_name": summary.get("parent_analysis_name", pd.NA),
        "num_events": len(events),
        "user_events": type_counts[EventType.USER.value],
        "rise_events": type_counts[EventType.RISE.value],
        "fall_events": type_counts[EventType.FALL.value],
        "transient_events": type_counts[EventType.TRANSIENT.value],
        "mean_duration": (sum(durations) / len(durations)) if durations else pd.NA,
    }
    return {key: values.get(key, pd.NA) for key in self.get_summary_columns()}

load_json_dict

load_json_dict(record: dict[str, Any]) -> None

Load event analysis state from a JSON analysis record.

Parameters:

Name	Type	Description	Default
record	dict[str, Any]	Analysis sidecar record.	required

Source code in src/acqstore/acq_image/analysis/event_analysis/event_analysis.py

def load_json_dict(self, record: dict[str, Any]) -> None:
    """Load event analysis state from a JSON analysis record.

    Args:
        record: Analysis sidecar record.
    """
    detection_params = dict(record["detection_params"])
    self.set_detection_params(detection_params)
    self.result.summary = dict(record["summary"])
    self.events = AcqImageEventStore.from_summary_dict(self.result.summary)
    self._sync_summary()
    self.set_clean()