playNano.processing package¶

Submodules¶

playNano.processing.core module¶

Core functions for loading and processing AFMImageStacks.

playNano.processing.core.process_stack(input_path: Path, channel: str, steps: list[tuple[str, dict]]) → AFMImageStack[source]¶

Load an AFMImageStack, apply the given steps, and return it.

Raises LoadError on load failure.

playNano.processing.filters module¶

Module for applying flattening and filtering to AFM images in Numpy arrays.

playNano.processing.filters.gaussian_filter(data: ndarray, sigma: float = 1.0) → ndarray[source]¶

Apply a Gaussian low-pass filter to smooth high-frequency noise.

Parameters:

data (np.ndarray) – 2D AFM image data.
sigma (float) – Standard deviation for Gaussian kernel, in pixels.

Returns:

Smoothed image.

Return type:

np.ndarray

playNano.processing.filters.polynomial_flatten(data: ndarray, order: int = 2) → ndarray[source]¶

Subtract a 2D polynomial surface of given order to flatten AFM image data.

Parameters:

data (np.ndarray) – 2D AFM image data.
order (int) – Polynomial order for surface fitting (e.g., 1 for linear, 2 for quadratic).

Returns:

Flattened image with polynomial background removed.

Return type:

np.ndarray

Raises:

ValueError – If data is not a 2D array or if order is not a positive integer.

playNano.processing.filters.register_filters()[source]¶: Return list of filter options.

playNano.processing.filters.remove_plane(data: ndarray) → ndarray[source]¶

Fit a 2D plane to the image using linear regression and subtract it.

Uses a 2D plane (z = ax + by + c) to remove to remove overall tilt.

Parameters:: data (np.ndarray) – 2D AFM image data.
Returns:: Plane-removed image.
Return type:: np.ndarray

playNano.processing.filters.row_median_align(data: ndarray) → ndarray[source]¶

Subtract the median of each row from that row to remove horizontal banding.

Parameters:: data (np.ndarray) – 2D AFM image data.
Returns:: Row-aligned image.
Return type:: np.ndarray

playNano.processing.filters.zero_mean(data: ndarray, mask: ndarray = None) → ndarray[source]¶

Subtract the overall mean height to center the background around zero.

If a mask is provided, mean is computed only over background (mask == False).

Parameters:

data (np.ndarray) – 2D AFM image data.
mask (np.ndarray, optional) – Boolean mask of same shape as data; True indicates region to exclude from mean.

Returns:

Zero-mean image.

Return type:

np.ndarray

playNano.processing.mask_generators module¶

Module for masking features of AFM images in Numpy arrays.

playNano.processing.mask_generators.mask_adaptive(data: ndarray, block_size: int = 15, offset: float = 0.0) → ndarray[source]¶: Adaptive local mean threshold per block.

playNano.processing.mask_generators.mask_below_threshold(data: ndarray, threshold: float = 0.0) → ndarray[source]¶: Mask where data < threshold.

playNano.processing.mask_generators.mask_mean_offset(data: ndarray, factor: float = 1.0) → ndarray[source]¶: Mask values > mean +/- factor*std.

playNano.processing.mask_generators.mask_morphological(data: ndarray, threshold: float, structure_size: int = 3) → ndarray[source]¶: Threshold+closing to mask foreground.

playNano.processing.mask_generators.mask_threshold(data: ndarray, threshold: float = 0.0) → ndarray[source]¶: Mask where data > threshold.

playNano.processing.mask_generators.register_masking()[source]¶: Return list of masking options.

playNano.processing.masked_filters module¶

Module for filtering AFM data in NumPy arrays with a boolean mask.

playNano.processing.masked_filters.polynomial_flatten_masked(data: ndarray, mask: ndarray, order: int = 2) → ndarray[source]¶

Fit a 2D polynomial using background (mask==False) and subtract it.

Parameters:

data (np.ndarray) – 2D AFM image.
order (int) – Polynomial order. Default order=2.
mask (np.ndarray) – Boolean mask of same shape; True=foreground, False=background.

Returns:

Polynomial-flattened image.

Return type:

np.ndarray

Raises:

ValueError – If mask.shape != data.shape or order is not a positive integer.

playNano.processing.masked_filters.register_mask_filters()[source]¶: Return list of masking options.

playNano.processing.masked_filters.remove_plane_masked(data: ndarray, mask: ndarray) → ndarray[source]¶

Fit a 2D plane on background only and subtract it from the full image.

Parameters:

data (np.ndarray) – 2D AFM image.
mask (np.ndarray) – Boolean mask of same shape; True=foreground (excluded), False=background (used to fit).

Returns:

Plane-removed image.

Return type:

np.ndarray

Raises:

ValueError – If mask.shape != data.shape.

playNano.processing.masked_filters.row_median_align_masked(data: ndarray, mask: ndarray) → ndarray[source]¶

Compute each row’s median using background pixels and subtract from each full row.

Parameters:

data (np.ndarray) – 2D AFM image.
mask (np.ndarray) – Boolean mask of same shape; True=foreground, False=background.

Returns:

Row-masked-alignment image.

Return type:

np.ndarray

Raises:

ValueError – If mask.shape != data.shape.

playNano.processing.masked_filters.zero_mean_masked(data: ndarray, mask: ndarray) → ndarray[source]¶

Subtract the mean of background pixels (mask==False) from the entire image.

Parameters:

data (np.ndarray) – 2D AFM image.
mask (np.ndarray) – Boolean mask of same shape; True=foreground, False=background.

Returns:

Image with zero-mean background.

Return type:

np.ndarray

playNano.processing.pipeline module¶

Module containing the ProcessingPipeline class for AFMImageStack processing.

This module provides ProcessingPipeline, which runs a sequence of mask/filter/method/plugin steps on an AFMImageStack. Each step’s output is stored in stack.processed (for filters) or stack.masks (for masks), and detailed provenance (timestamps, parameters, step type, version info, keys) is recorded in stack.provenance[“processing”]. Environment metadata at pipeline start is recorded in stack.provenance[“environment”].

class playNano.processing.pipeline.ProcessingPipeline(stack: AFMImageStack)[source]¶

Bases: object

Orchestrates a sequence of masks and filters (or other processing steps).

Acts on an AFMImageStack, recording outputs and detailed provenance. Each step is specified by name and kwargs:

“clear” resets any active mask

mask steps compute boolean masks stored under stack.masks[…]

filter/method/plugin steps apply to the current data (and mask if present), storing results in stack.processed[…]

Provenance for each step (index, name, params, timestamp, step_type, version, keys, summaries) is appended to stack.provenance[“processing”][“steps”]. A mapping from step name to list of snapshot keys is built as stack.provenance[“processing”][“keys_by_name”]. The final processed array overwrites stack.data. Environment info is captured once under stack.provenance[“environment”].

add_filter(filter_name: str, **kwargs) → ProcessingPipeline[source]¶

Add a filter step to the pipeline.

Parameters:

filter_name (str) – The name of the registered filter function to apply.
**kwargs – Additional keyword arguments for the filter function.

Returns:

The pipeline instance (for method chaining).

Return type:

ProcessingPipeline

Notes

If a mask is currently active, the pipeline will attempt to use a masked version of the filter (from MASK_FILTERS_MAP) if available. Otherwise, the unmasked filter is applied to the whole dataset.

add_mask(mask_name: str, **kwargs) → ProcessingPipeline[source]¶

Add a masking step to the pipeline.

Parameters:

mask_name (str) – The name of the registered mask function to apply.
**kwargs – Additional parameters passed to the mask function.

Returns:

The pipeline instance (for method chaining).

Return type:

ProcessingPipeline

Notes

If a mask is currently active (i.e. not cleared), this new mask will be logically combined (ORed) with the existing one.

clear_mask() → ProcessingPipeline[source]¶

Add a step to clear the current mask.

Returns:: The pipeline instance (for method chaining).
Return type:: ProcessingPipeline

Notes

Calling this resets the masking state, so subsequent filters will be applied to the entire dataset unless a new mask is added.

run() → ndarray[source]¶

Execute configured steps on the AFMImageStack, stores outputs and provenance.

Behavior: 1. Record or update environment metadata via gather_environment_info()

into stack.provenance[“environment”].

Reset previous processing provenance under stack.provenance[“processing”], ensuring keys “steps” (list) and “keys_by_name” (dict) exist and are cleared.

3. If not already present, snapshot the original data as “raw” in stack.processed. 4. Iterate over self.steps in order (1-based index):

Resolve step type via stack._resolve_step(step_name), yielding

(step_type, fn). - Record a timestamp (utc_now_iso()), index, name, params, step_type,

function version (via fn.__version__ or plugin lookup), and module name.

If step_type == “clear”:

reset current mask to None

record “mask_cleared”: True in provenance entry

If step_type == “mask”:

call stack._execute_mask_step(fn, arr, **kwargs) to get a boolean

mask array * if no existing mask: store under a new key step_<idx>_<mask_name> in stack.masks

else overlay with previous mask (logical OR) under derived key

update current mask, record “mask_key” and “mask_summary” in

provenance

Else (filter/method/plugin):

call stack._execute_filter_step(fn, arr, mask, step_name, **kwargs)

to get new array * store result under stack.processed[“step_<idx>_<safe_name>”], update arr * record “processed_key” and “output_summary” in provenance

After all steps, overwrite stack.data = arr.

6. Build stack.provenance[“processing”][“keys_by_name”] mapping each step name to the list of stored keys (processed_key or mask_key) in order. 7. Return the final processed array.

Returns:

The final processed data array (shape (n_frames, height, width)), now stored as stack.data.

Return type:

np.ndarray

Raises:

RuntimeError – If a step cannot be resolved or executed due to misconfiguration.
ValueError – If overlaying a mask fails due to missing previous mask key (propagated).
Exception – Any exception from individual steps is logged then re-raised.

Examples

>>> stack = AFMImageStack(data, pixel_size_nm=1.0, channel="h", file_path=".")
>>> pipeline = ProcessingPipeline(stack)
>>> pipeline.add_filter("gaussian_filter", sigma=1.0)
>>> pipeline.add_mask("threshold_mask", threshold=0.5)
>>> result = pipeline.run()
>>> # Final array in stack.data:
>>> result.shape
(n_frames, height, width)
>>> # Provenance entries:
>>> for rec in stack.provenance["processing"]["steps"]:
...     print(
...         rec["index"],
...         rec["name"],
...         rec.get("processed_key") or rec.get("mask_key")
...     )
>>> # Keys by name:
>>> print(stack.provenance["processing"]["keys_by_name"])
>>> # Environment info:
>>> print(stack.provenance["environment"])

playNano.processing.versioning module¶

Module for applying verision numbes to filters and masks.

playNano.processing.versioning.versioned_filter(version: str)[source]¶

Add decorator tag to a filter or mask function with a __version__ attribute.

Usage:: @versioned_filter(“1.0.0”) def gaussian_filter(frame, sigma=1.0):

…

Module contents¶

Public package initialization.