PeakFinder¶

class PeakFinder(settings: Optional[dict] = None)

Event fitter that resolves each event into peaks and classifies them.

A MetaEventFitter subclass driven entirely by the base class: MetaEventFitter.fit_events() calls _pre_process_events once per channel, then _locate_sublevel_transitions / _populate_sublevel_metadata / _populate_event_metadata once per event, then _post_process_events when the channel’s last event is fitted. Per-event work finds peaks and measures them; the dataset-wide classifiers (folding, peak prominence, translocation direction, bound star) all run from _post_process_events once every channel has finished, because each of them fits a distribution that only exists across the whole run.

fit_fallbacks.md at the repository root describes the behaviour that is too large to carry in docstrings: every fallback path in the double-Gaussian threshold fit.

Public Methods¶

PeakFinder.classify_1d_distribution(data: ndarray, n_components: int = 2, return_centers: bool = False) → ndarray | Tuple[ndarray, ndarray, GaussianMixture]¶

Classify 1D data into clusters using Gaussian Mixture Model.

Not currently called. Kept as the sklearn alternative to the curve_fit path in fit_threshold: it works on raw samples rather than on a histogram, so it avoids the binning sensitivity MIN_FIT_BINS exists to work around, and BIC would choose the component count instead of it being fixed at two.

Parameters:

data (np.ndarray) – 1D array of data to classify
n_components (int) – Number of Gaussian components to fit
return_centers (bool) – Whether to return cluster centers and GMM model along with labels

Returns:

Cluster labels, optionally with centers and fitted GMM model

Return type:

Union[np.ndarray, Tuple[np.ndarray, np.ndarray, GaussianMixture]]

PeakFinder.classify_2d_distribution(data: ndarray, n_components: int = 2, return_centers: bool = False) → ndarray | Tuple[ndarray, ndarray, GaussianMixture]¶

Classify 2D data into clusters using Gaussian Mixture Model.

Not currently called. The two-dimensional counterpart of classify_1d_distribution, for separating populations that need two features - prominence against width, say - rather than one.

Parameters:

data (np.ndarray) – 2D array of data to classify
n_components (int) – Number of Gaussian components to fit
return_centers (bool) – Whether to return cluster centers and GMM model along with labels

Returns:

Cluster labels, optionally with centers and fitted GMM model

Return type:

Union[np.ndarray, Tuple[np.ndarray, np.ndarray, GaussianMixture]]

PeakFinder.close_resources(channel: int | None = None) → None¶

Release per-channel resources. Nothing here holds any, so this is a no-op.

Called by the base class on channel reset and on application exit, either for one channel or for all of them.

Parameters:: channel (Optional[int]) – the index of the channel to close resources for, or None to close resources for every channel

PeakFinder.construct_fitted_event(channel: int, index: int) → ndarray[tuple[int, ...], dtype[float64]] | None¶

Rebuild an idealised trace for one event from its fitted metadata.

Called by MetaEventFitter.get_fitted_event(), which validates the returned array against the raw event and hands it to the plotting layer to be drawn over the raw data. Nothing else consumes it.

The trace is assembled at baseline, then overwritten in three passes: the padding either side of the event, the mean level of every type-3 sublevel, and the max_blockage level of each peak between its two inflection points. Times held in the metadata in microseconds are converted to sample indices here, and the result is forced to the raw event’s own length so the two can be plotted against one axis.

Parameters:

channel (int) – analyze only events from this channel
index (int) – the index of the target event

Returns:

numpy array of peaked data for the event, or None

Return type:

Optional[npt.NDArray[np.float64]]

Raises:

RuntimeError – if peakfinding is not complete yet

PeakFinder.enumerate_peaks(sublevel_starts: List[Dict[str, Any]], num_states: int, sublevel_types: List[str] | None = None) → List[int | None]¶

Number the peak sublevels of one event 1..N, leaving None elsewhere.

Called by _populate_sublevel_metadata to fill peak_id, which is what every later stage uses to tell a peak sublevel from a baseline or padding one.

Parameters:

sublevel_starts (List[Dict[str, Any]]) – List of dictionaries describing sublevels, each with a ‘type’ key.
num_states (int) – Total number of sublevels to process.
sublevel_types (Optional[List[str]]) – List of sublevel types (‘peak’ or ‘event_baseline’). If None, falls back to edge type checking.

Returns:

List of peak IDs or None for non-peak sublevels.

Return type:

List[Optional[int]]

PeakFinder.filter_peaks(peaks: ndarray[tuple[int, ...], dtype[int64]], properties: Dict[str, Any], unfolded_level: float, folded_level: float | None, baseline_std: float | None, baseline: float | None, samplerate: float, event_length: int) → Dict[str, Any]¶

Assign a type to every peak in one event, from where its bases sit relative to the carrier levels and from how close it is to its neighbours. Writes the result into properties["filtered"].

Called by update_event_metadata_post_processing, once per event, after _classify_folded_unfolded has measured the run-wide carrier levels. Those labels are what every later stage selects on: the sequence builder and _classify_translocation_direction read type 3, _classify_bound_star reads type -1, and _classify_peak_prominences reads types 1, 2 and 3.

The branch taken depends on the Event Type setting.

Barcode. Each peak is typed from its two bases, which must both land in the same band. Writing U for the unfolded level, s for the baseline standard deviation, t1 for Lower Filter Threshold and t2 for Higher Filter Threshold, the bands are as follows. Every band and both bases are blockage depths from the baseline - deeper is always larger - so the typing does not depend on the sign of the current:

-1 - both bases above 2*U + t2*s, or the two bases in different bands. This is the catch-all, and it is where a bound star lands as well as any straddling or spurious peak.
2 - both bases within [U + t2*s, 2*U + t2*s]: a peak sitting on the folded carrier.
1 - both bases within [U + t1*s, U + t2*s]: a peak sitting on the unfolded carrier. t1 defaults negative, so this band opens below the unfolded level and closes above it.
0 - both bases at or below BASELINE_BAND_SIGMA * s: on the baseline, not on a carrier.

They are tested in that order, which is what decides the two cases where the bands overlap rather than meet. The baseline band is sized by the BASELINE_BAND_SIGMA class constant rather than by t2, because t2 sizes the tolerance around the carrier levels and at larger t2 it opened a baseline band that reached the bottom of the type-1 band; testing type 1 first means a peak seated on the unfolded carrier is typed 1 either way, and only a genuinely baseline-seated peak falls through to type 0.

A peak that qualifies as type 1 but whose base_at_edge flag is set - one of its bases is where the trim stopped find_peaks’ base walk, at either end of the trimmed event, rather than a minimum in the trace - is rejected to -1 instead. Such a base holds whatever the event was doing at its own edge, usually a carrier level, so neither it nor the prominence measured from it describes the peak. Both ends are treated alike because _classify_translocation_direction may reverse a sequence, and a rule that rejected only the leading edge would take opposite ends off a forward and a backward copy of the same molecule. Note that these peaks join the type -1 pool that _classify_bound_star draws its candidates from.

A second pass then promotes exactly Number of peaks type-1 peaks to type 3, the barcode, choosing the set that looks most like one barcode rather than the set that is largest. Candidates are the type-1 peaks with another type-1 peak within Peak to Peak Distance Ratio percent of the event length; _select_barcode_peaks then scores every legal set of that size on how alike its consecutive spacings, peak widths and peak ECDs are, lowest total winning, ties going to the largest total ECD and then to the earliest set. Adjacent members of the chosen set must be within that same spacing limit, measured after the set is chosen, since skipping a peak widens the gap it leaves behind.

The set need not be consecutive among the candidates. Every peak not chosen - skipped from the middle of the barcode included - keeps the type the first pass gave it, so a skipped interior peak stays type 1 and does not appear in the event’s sequence.

Single Peak. Peaks are typed 11, 12, 13, 21 or 22 by which transition their bases describe, and everything but the single most prominent survivor is then reset to -1.

Unspecified. No typing at all.

Parameters:

peaks (npt.NDArray[np.intp]) – indices of the located peaks, as returned by scipy.signal.find_peaks
properties (Dict[str, Any]) – the peak properties dict returned alongside the peak indices
unfolded_level (float) – the blockage level of the unfolded carrier
folded_level (Optional[float]) – the blockage level of the folded carrier, if one is known
baseline_std (Optional[float]) – the local standard deviation of the baseline current
baseline (Optional[float]) – the local mean value of the baseline current
samplerate (float) – the sampling rate
event_length (int) – the length of the event

Returns:

the properties dict, with its “filtered” entry updated with peak classifications

Return type:

Dict[str, Any]

Raises:

RuntimeError – if baseline_std is None, since it scales every classification threshold

PeakFinder.find_mode_blockage_level(data: ndarray[tuple[int, ...], dtype[float64]], baseline_mean: float | None, baseline_std: float | None) → Tuple[float | None, float | None]¶

Extract the most populated blockage level from the data.

Called by _locate_sublevel_transitions, both for the event’s carrier blockage and for each peak’s max_blockage, and by _populate_event_metadata for the event’s primary_level. Those primary levels are what _classify_folded_unfolded later fits.

Uses numpy histogram to find the most common current level. Data should already be trimmed to the longest continuous segment above threshold by _locate_sublevel_transitions. Folded/unfolded classification is deferred to post-processing across all events.

Parameters:

data (npt.NDArray[np.float64]) – Array of current values (already trimmed to longest segment).
baseline_mean (Optional[float]) – Mean value of the baseline level.
baseline_std (Optional[float]) – Standard deviation of the baseline level.

Returns:

Tuple of (primary_blockage_level, secondary_blockage_level) - the 2 most populated distinct blockage levels

Return type:

Tuple[Optional[float], Optional[float]]

Raises:

RuntimeError – if baseline_mean is None, since blockage levels are measured relative to the baseline

PeakFinder.fit_threshold(data: ndarray[tuple[int, ...], dtype[float64]]) → Dict[str, Any]¶

Estimate a binary threshold from 1-D data by fitting a double Gaussian.

Shared by all three _classify_* methods and their plotting code, which consume the returned dict generically. "params" is ordered per component, (amp1, mean1, std1, amp2, mean2, std2, offset), where the trailing offset is the flat constant fitted alongside the two Gaussians (see FIT_CONSTANT_OFFSET) and is 0.0 wherever the constant is switched off or lost the residual comparison in _curve_fit_bounded; it is also given its own "offset" key so a consumer need not index for it. The key is called "threshold" rather than anything implying a midpoint, because it is placed from the histogram’s shape (see _threshold_between_populations) and is not the midpoint of anything.

The constant does not enter the threshold. Being common to both components it cancels out of the crossing that _gaussian_intersection solves for, and it is excluded by construction from the Gaussian-mixture posterior _classification_confidence computes, so adding it moves no class boundary - it only stops the two Gaussians from having to widen to cover a background that belongs to neither population.

This raises when the fit fails rather than falling back to raw histogram peak locations. A failed fit is a result worth seeing, not one worth papering over: a fallback would make “no two populations here” indistinguishable from “two populations, found and measured”.

A fitted double Gaussian always has two centres, whether or not the data actually contains two populations - on genuinely unimodal data it either collapses one component or converges with both centres on the same mode, both diagnosed by _fit_and_check_double_gaussian. "n_components" surfaces that diagnosis as 1 or 2 so a caller can act on it directly instead of only finding it in the log.

Where the threshold came from an actual feature of the curve - a valley between the two centres, or the above-floor fallback on single-population data - the two components are then re-fit jointly, with each mean hard- bounded to its own side of that point, and the reported "threshold" is replaced by the x position where the two refit curves cross. See _fit_double_gaussian_bounded_at_valley and _gaussian_intersection. This also happens on single-population data, where the higher component ends up describing only the sparse region above the threshold and so has a small amplitude. "params_method" records whether this ran and succeeded ("constrained") or not ("joint"): when it is "joint", "threshold" is still the raw valley/floor value _threshold_between_populations found; when it is "constrained", "threshold" is the Gaussian crossing instead, which is also exactly where _classification_confidence reads 0.5, so the shape-based label and the Gaussian-mixture vote agree exactly at the boundary. The two fallback methods are excluded from this entirely: those thresholds are not read off any feature, and fallback_degenerate is the midpoint of two co-located means, so constraining a refit there would carve one mode arbitrarily in half.

Parameters:: data (npt.NDArray[np.float64]) – 1-D array of values from which to estimate a binary threshold
Returns:: dict with keys "threshold" (float), "centers" (np.ndarray of the two fitted means), "hist" (tuple of counts and bin edges), "params" (tuple of the six fitted component parameters followed by the fitted flat constant), "offset" (that same constant on its own, 0.0 where none was fitted), "n_components" (1 if the fit describes one population, 2 if it describes two), "threshold_method" (how the valley/floor point that anchored the fit was picked - see _threshold_between_populations; unaffected by whether the constrained refit below ran), "spline" (the smoothing spline that search was run on, or None) paired with "spline_domain" (the populated core it was fit over; the returned BSpline extrapolates without bound outside those knots, so any consumer must clip to it), and "params_method" ("constrained" if the components were re-fit jointly with their means bounded at that point and "threshold" replaced by their crossing, "joint" if "params" and "threshold" are straight from the unconstrained double-Gaussian fit and _threshold_between_populations respectively)
Return type:: Dict[str, Any]
Raises:: ValueError – if the data cannot be histogrammed, or if the double-Gaussian fit fails to converge

PeakFinder.get_empty_settings(globally_available_plugins: Dict[str, List[str]] | None = None, standalone: bool = False) → Dict[str, Dict[str, Any]]¶

Declare the settings this fitter exposes, on top of the base contract.

Called by poriscope when the plugin is instantiated or reconfigured, to build the settings dialog and to sanity-check whatever the user enters; the accepted values are then readable through self.settings. See get_empty_settings() for the structure of the dict and what Type, Value, Min, Max, Options and Units mean in it.

The super() call supplies the mandatory "MetaEventLoader" key, which is how this plugin is wired to its data source.

The keys added here, and where each one is consumed:

Event Type - selects the branch filter_peaks takes: "Barcode", "Single Peak", or "Unspecified" (no peak typing at all).
Number of peaks - in the barcode branch, both the smallest cluster that may be called a barcode and the number of most-prominent peaks considered for clustering.
Lower Filter Threshold / Higher Filter Threshold - in units of the baseline standard deviation, the tolerance bands filter_peaks places around the carrier levels when typing peaks, and the margin _classify_bound_star requires above the folded level.
Peak to Peak Distance Ratio - as a percentage of event length, how close two peaks must be to join the same barcode cluster.
Window Length Percentage - as a percentage of event length, the wlen prominence window _locate_sublevel_transitions hands to scipy.signal.find_peaks.
Min Carrier Blockage - the smallest carrier blockage an event may have and still be admitted to the folding fit.

Parameters:

globally_available_plugins (Optional[Dict[str, List[str]]]) – a dict containing all data plugins that exist to date, keyed by metaclass. Must include “MetaEventLoader” as a key, with explicitly set Type MetaEventLoader.
standalone (bool) – False if this is called as part of a GUI, True otherwise. Default False

Returns:

the dict that must be filled in to initialize the filter

Return type:

Dict[str, Dict[str, Any]]

PeakFinder.get_metadata_columns(channel: int) → List[str]¶

List the event metadata columns, less this plugin’s working values.

NOTE (merge fix): this override is new. PRIVATE_EVENT_METADATA is stripped from get_single_event_metadata and from the declared types and units, but the base get_metadata_columns reads the keys of event_metadata directly, so it went on naming the four private fields - leaving the three accessors disagreeing about what a column is. That is what the behavioural conformance suite caught: every produced column must carry a declared type and unit, and the keys of get_single_event_metadata must be exactly the produced columns. Both hold again once the same tuple is stripped here.

Parameters:: channel (int) – analyze only events from this channel
Returns:: a list of column names
Return type:: List[str]

Build the reference lines and labels drawn over one fitted event.

Called by the plotting layer through the MetaEventFitter.get_plot_features() contract, alongside construct_fitted_event() for the same event. Returns horizontal lines (baseline, unfolded level, and that level offset by each of the two filter thresholds), one marker per peak, and the label strings that go with them. The three unfolded-level lines are omitted when the event has no unfolded level - before the run-wide folded/unfolded classification has run, or when it could not separate two populations.

Only metadata that is actually populated reaches a label: an event the classifiers never reached reads as having no value rather than a value of nan, which would look like a classification that ran and failed. SQLitePeakDBLoader.get_plot_features mirrors this method for events read back out of a database and has to be kept in step with it.

Parameters:

channel (int) – analyze only events from this channel
index (int) – the index of the target event

Returns:

a list of x locations to plot vertical lines and a list of y locations to plot horizontal lines, labels for the vertical lines, labels for the horizontal lines. Must be lists of equal length, or None

Return type:

Tuple[Optional[List[float]], Optional[List[float]], Optional[List[Tuple[float, float]]], Optional[List[str]], Optional[List[str]], Optional[List[str]]]

PeakFinder.get_single_event_metadata(channel: int, index: int) → Tuple[dict, dict, ndarray[tuple[int, ...], dtype[float64]], ndarray[tuple[int, ...], dtype[float64]], ndarray[tuple[int, ...], dtype[float64]] | None]¶

Return one event’s metadata with this plugin’s working values removed.

Called through get_event_metadata_generator, which is what MetaDatabaseWriter iterates when writing a database - the only consumer of this method - so this is the single seam where “what PeakFinder computes” and “what gets persisted” part company. See PRIVATE_EVENT_METADATA for what is dropped and why it has to be dropped here as well as from the declared types.

The dicts returned are fresh shallow copies, so the internal metadata this strips from is left intact for get_plot_features and the classifiers, which keep reading the full set.

Parameters:

channel (int) – Channel from which to retrieve the event
index (int) – Index of the event to retrieve

Returns:

Tuple of event metadata, sublevel metadata, filtered data, raw data, and fitted data

Return type:

Tuple[dict, dict, npt.NDArray[np.float64], npt.NDArray[np.float64], Optional[npt.NDArray[np.float64]]]

PeakFinder.get_sublevel_columns(channel: int) → List[str]¶

List the sublevel metadata columns, less this plugin’s working values.

The sublevel half of get_metadata_columns above, stripping PRIVATE_SUBLEVEL_METADATA for the same reason.

Parameters:: channel (int) – analyze only events from this channel
Returns:: a list of column names
Return type:: List[str]

PeakFinder.redefine_padding(data: ndarray[tuple[int, ...], dtype[float64]], samplerate: float, baseline_std: float | None) → ndarray[tuple[int, ...], dtype[float64]]¶

Locate sublevel edges within an event using a CUSUM change-point detector.

Re-derives an event’s padding boundaries instead of trusting the values the event loader supplied.

Not currently called. Its only call site, in _locate_sublevel_transitions, is commented out: the padding is taken from the event loader and then trimmed to the longest above-threshold segment instead. Kept because it is the only change-point detector in this plugin and the trimming approach may not suit every recording.

Parameters:

data (npt.NDArray[np.float64]) – an array of data from which to locate sublevel edges
samplerate (float) – the sampling rate
baseline_std (Optional[float]) – the local standard deviation of the baseline current

Returns:

the located edge positions, beginning at 0 and ending at len(data)

Return type:

npt.NDArray[np.float64]

Raises:

RuntimeError – if baseline_std is None, since it sets the CUSUM step size

PeakFinder.report_channel_status(channel: int | None = None, init: bool = False) → str¶

Return the fitting report plus the run’s full classification section.

Called by the UI to show a channel’s status, and by _save_classification_report to build the report file written at the end of post-processing. The classification section is appended only once _post_process_events has produced _classification_results; before that, or when init is set, only the base report is returned.

The classification section covers, in order: the folding fit and its folded/unfolded counts, the peak type breakdown from _collect_peak_statistics, the prominence fit and its class counts, the translocation direction fit and its counts, the sequence tallies, the bound-star tallies with a per-sequence breakdown, and finally - when self._classification_warnings is non-empty - a “Warnings during classification” section listing every distinct WARNING-level message logged while the classifiers ran, each with a (xN) suffix if it repeated. See _post_process_events for how those are collected.

Parameters:

channel (Optional[int]) – the channel to report on, or None for all channels
init (bool) – whether this is an initialization report

Returns:

the status report as a string

Return type:

str

Raises:

RuntimeError – if the channel’s peak statistics cannot be assembled

PeakFinder.update_event_metadata_post_processing(channel: int, event_index: int, unfolded_level: float | None = None, folded_level: float | None = None) → None¶

Write the run-wide carrier levels onto one event and re-type its peaks.

Called by _classify_folded_unfolded, once per event, as soon as the folding fit has decided that event’s class. Nothing else calls it.

Three things happen here, and only the first is bookkeeping:

unfolded_level and folded_level are recorded on the event.
peak_height, prominence and max_blockage are divided by the unfolded level into their normalized_* counterparts, which _populate_sublevel_metadata could only leave as NaN.
filter_peaks is re-run for the event against those levels. The per-event pass could only guess the carrier levels from that one event; this pass re-types every peak against levels measured over the whole run, and the filtered labels it writes are what every later stage - sequences, translocation direction, bound star, the report - actually reads.

Parameters:

channel (int) – Channel number
event_index (int) – Event index within the channel
unfolded_level (Optional[float]) – Determined unfolded level for normalization
folded_level (Optional[float]) – Determined folded level for classification

Private Methods¶

PeakFinder._define_event_metadata_types() → Dict[str, Type[int | float | str | bool]]¶

Declare the column name and dtype of every piece of event metadata.

Called once by MetaEventFitter during initialization, and used by the database writer downstream to build the events table. The keys must match those _populate_event_metadata returns exactly, and must be kept in step with _define_event_metadata_units.

Returns:: a dict of metadata keys and associated base dtypes
Return type:: Dict[str, Type[Union[int, float, str, bool]]]

PeakFinder._define_event_metadata_units() → Dict[str, str | None]¶

Declare the unit of every piece of event metadata, or None if unitless.

Called once by MetaEventFitter during initialization, and carried into the database so downstream plots can label their axes. The keys must match _define_event_metadata_types exactly.

Returns:: a dict of metadata keys and associated units
Return type:: Dict[str, Optional[str]]

PeakFinder._define_sublevel_metadata_types() → Dict[str, Type[int | float | str | bool]]¶

Declare the column name and dtype of every piece of sublevel metadata.

Called once by MetaEventFitter during initialization, and used by the database writer downstream to build the sublevels table. The keys must match those _populate_sublevel_metadata returns exactly, and must be kept in step with _define_sublevel_metadata_units. The type given is that of a single entry, not of the list holding them.

Returns:: a dict of metadata keys and associated base dtypes
Return type:: Dict[str, Type[Union[int, float, str, bool]]]

PeakFinder._define_sublevel_metadata_units() → Dict[str, str | None]¶

Declare the unit of every piece of sublevel metadata, or None if unitless.

Called once by MetaEventFitter during initialization, and carried into the database so downstream plots can label their axes. The keys must match _define_sublevel_metadata_types exactly.

Returns:: a dict of metadata keys and associated units
Return type:: Dict[str, Optional[str]]

PeakFinder._init() → None¶

Hook for subclass setup at the start of base class initialization. Nothing to do here, so this is a no-op.

Called once by MetaEventFitter while the plugin is being constructed, before any settings are applied.

Find the peaks in one event and return the sublevel boundaries around them.

Called by MetaEventFitter.fit_events() once per event, first of the three per-event hooks; whatever it returns is passed verbatim to _populate_sublevel_metadata and _populate_event_metadata. Here that is a list of (edge_index, sublevel_type) tuples plus the scipy.signal.find_peaks properties dict, so the downstream hooks do not have to re-detect anything.

The stages, and the filters each one applies:

Re-measure the baseline mean and standard deviation from the padding the event loader reported.
Trim the event to the longest continuous run more than min(|lower threshold|, |higher threshold|, 3) baseline sigma away from the baseline, and fold everything outside it into the padding. An event with no such run, or whose longest run is under 100 samples, is rejected.
Take the carrier blockage as the modal level of the trimmed data and reject the event if it is below Min Carrier Blockage.
Detect peaks with find_peaks, using a minimum prominence of the larger of the noise floor and the carrier blockage, a minimum height of the carrier blockage plus that prominence, a prominence window wlen set by Window Length Percentage, and a minimum peak separation of half that window.
Turn the peak inflection points into sublevel edges and label each sublevel "peak" or "event_baseline".

Rejection at any of these stages raises ValueError, which drops the one event and lets the run continue.

Parameters:

data (npt.NDArray[np.float64]) – an array of data from which to extract the locations of sublevel transitions
samplerate (float) – the sampling rate
padding_before (Optional[int]) – the number of data points before the estimated start of the event in the chunk
padding_after (Optional[int]) – the number of data points after the estimated end of the event in the chunk
baseline_mean (Optional[float]) – the local mean value of the baseline current
baseline_std (Optional[float]) – the local standard deviation of the baseline current

Returns:

a list of integers corresponding to sublevel transitions

Return type:

Optional[List[Any]]

Raises:

ValueError – if the event is rejected. Note that ValueError will skip and reject the event but will not stop processing of the rest of the dataset

Reduce one event’s sublevel measurements to a single row of event metadata.

Called by MetaEventFitter.fit_events() once per event, last of the three per-event hooks, with the dict _populate_sublevel_metadata returned. 'start_time_s' and 'index' are handled by the base class and must not appear here.

Peak count, duration, ECD and maximum deviation are summed or reduced over the sublevels between the two padding sublevels. The baseline current is the duration-weighted mean of the two padding sublevels, and the baseline standard deviation the smallest of that weighted mean and either padding sublevel on its own, so a padding region that happens to contain part of the event cannot inflate it. primary_level is the modal blockage of the event between its first and last sublevel.

Six fields are set to None here and filled in during post-processing, once the whole run has been fitted: unfolded_level and folded_level by _classify_folded_unfolded, translocation_direction and translocation_confidence by _classify_translocation_direction, sequence by _post_process_events, and bound_star by _classify_bound_star.

Parameters:

data (npt.NDArray[np.float64]) – an array of data from which to extract the locations of sublevel transitions
samplerate (float) – the sampling rate
baseline_mean (Optional[float]) – the local mean value of the baseline current
baseline_std (Optional[float]) – the local standard deviation of the baseline current
sublevel_metadata (Dict[str, List[Numeric]]) – the dict of sublevel metadata built by self._populate_sublevel_metadata()

Returns:

a dict of event metadata values

Return type:

Dict[str, Union[int, float, str, bool]]

Raises:

RuntimeError – if the baseline current or standard deviation is non-numeric, or if no primary blockage level can be determined

PeakFinder._populate_sublevel_metadata(data: ndarray[tuple[int, ...], dtype[float64]], samplerate: float, baseline_mean: float | None, baseline_std: float | None, sublevel_starts: List[Any]) → Dict[str, ndarray[tuple[int, ...], dtype[int | float | number]]]¶

Measure every sublevel of one event and return one list per quantity.

Called by MetaEventFitter.fit_events() once per event, immediately after _locate_sublevel_transitions, whose return value arrives here as sublevel_starts. Every list returned has exactly one entry per sublevel; 'index' is handled by the base class and must not appear.

Each sublevel is typed "peak", "event_baseline" or "padding" from the edge that opens it, and then measured: median current, deviation from the local baseline, start and end time, duration, raw and cumulative ECD, maximum deviation, and standard deviation. Peak sublevels additionally carry everything find_peaks measured - location, width, height, prominence, bases, inflection points, modal blockage - plus three fields that only later stages can fill:

the _normalized variants, which divide by the event’s unfolded level and are written in update_event_metadata_post_processing once that level is known across the run;
filtered, the peak type, initialised to 0 and assigned by filter_peaks;
classified and classification_confidence, initialised to NaN and assigned by _classify_peak_prominences.

Non-peak sublevels carry NaN in all of the peak fields, which is what distinguishes them downstream.

Parameters:

data (npt.NDArray[np.float64]) – an array of data from which to extract the locations of sublevel transitions
samplerate (float) – the sampling rate
baseline_mean (Optional[float]) – the local mean value of the baseline current
baseline_std (Optional[float]) – the local standard deviation of the baseline current
sublevel_starts (List[Any]) – the list of sublevel start indices located in self._locate_sublevel_transitions()

Returns:

a dict of lists of sublevel metadata values, one list entry per sublevel for each piece of metadata

Return type:

Dict[str, npt.NDArray[Numeric]]

PeakFinder._post_process_events(channel: int) → None¶

Run every dataset-wide classifier, once the last channel is fitted.

Called by MetaEventFitter.fit_events() once per channel, after that channel’s last event. Each classifier below fits a distribution built from every event in the run, so all but the final call return early: the guard is _global_postprocessing_done, armed for a new fitting session by _pre_process_events.

On the final call, in order:

Pool one primary_level per event across all channels, dropping events with no primary level or with a non-positive ECD.
_classify_folded_unfolded fits that pool and, for every event, writes unfolded_level / folded_level and re-runs filter_peaks against the run-wide levels through update_event_metadata_post_processing.
_classify_peak_prominences fits the normalized prominences of all type 1, 2 and 3 peaks and writes classified / classification_confidence per peak.
_classify_translocation_direction fits the log ratio of the ECD before and after each event’s barcode and writes translocation_direction / translocation_confidence.
Build each event’s sequence from the prominence classes of its type-3 peaks, reversed for a backward event.
_classify_bound_star writes bound_star. It runs last because it needs both the direction, to put the star’s position into the molecule’s frame, and the sequences, to break its report down.
_save_classification_report writes the run’s report to disk.

Steps 1-6 run with a _ClassificationWarningCollector attached to self.logger, so every WARNING-level message any of them logs - a degraded fit, a declined classifier, a fitted mean off its peak - is captured into self._classification_warnings and surfaced in the report report_channel_status builds for step 7, rather than living only in the log file.

Parameters:: channel (int) – the index of the channel to postprocess

PeakFinder._pre_process_events(channel: int) → None¶

Arm the run-wide post-processing pass for a channel about to be fitted.

Called by MetaEventFitter.fit_events() once per channel, before any of that channel’s events are fitted. Clearing _global_postprocessing_done here is what lets the classifiers run again on a second fitting session in the same app instance; _post_process_events sets the flag when they have run.

Parameters:: channel (int) – the channel to preprocess

PeakFinder._validate_settings(settings: dict) → None¶

Check a candidate settings dict beyond the base class’s type and range checks. There is nothing to add here, so this is a no-op.

Called by MetaEventFitter whenever settings are applied, after the generic validation against get_empty_settings() has passed.

Parameters:: settings (dict) – Parameters for event detection.