NaNIsaac docs

Appearance

Python API ​

Signatures and docstrings of the public functions and classes in QoraFlow at main (0b204a2), read from the source without importing it. The guides on this site say how the pieces fit together; this page is for looking up an argument.

QoraFlow.analyzer.converter ​

load_npz ​

python
def load_npz(file: str) -> dict[str, npt.NDArray]

load_lcls_cube ​

python
def load_lcls_cube(file: str) -> dict[str, npt.NDArray]

save_npz ​

python
def save_npz(file: str, data: dict[str, npt.NDArray]) -> None

save_mat ​

python
def save_mat(file: str, data: dict[str, npt.NDArray]) -> None

convert_data ​

python
def convert_data(
    input_format: str,
    output_format: str,
    input_file: str,
    save_file: str,
) -> None

QoraFlow.analyzer.core ​

edge_background ​

python
def edge_background(
    images: npt.NDArray,
    method: Literal['edge', 'plane'],
    width: int = 3,
) -> npt.NDArray
Background of each image in a stack (n, h, w), estimated from its outer ``width`` pixels.

"edge" is the border median, a flat level. "plane" is the least-squares plane a + b·x + c·y
through the border, which also removes a gradient. Both read the peak's own tail as
background when the tail reaches the border: the median then oversubtracts a little, while
the plane takes the tail for a gradient and pulls the centre of mass back, which is worse
for a peak that sits off the ROI centre. A gradient biases the median the most. Keep the
ROI half-width several peak widths and the peak near its centre.

DataAnalyzer ​

python
class DataAnalyzer:
    def __init__(self, file: str | Path, angle: int=0) -> None
Analyzes data from a given file and performs various operations on the images.

Args:
    file (str): The path to the file containing the data.
    angle (int, optional): The angle to rotate the images. Defaults to 0.

Raises:
    FileNotFoundError: If the file does not exist.
    ValueError: If the file does not contain the required keys.

DataAnalyzer.get_summed_image ​

python
def get_summed_image(self) -> tuple[npt.NDArray | None, npt.NDArray | None]
return:
    pump off image, pump on image (None for a state the file does not have)

DataAnalyzer.pon_subtract_by_poff ​

python
def pon_subtract_by_poff(self)
Subtrack pump on images by pump off images

DataAnalyzer.analyze_by_roi ​

python
def analyze_by_roi(
    self,
    roi_rect: RoiRectangle,
    background: Background = None,
    edge_width: int = 3,
) -> pd.DataFrame
ROI centre of mass and mean intensity per delay, relative to the first delay.

Args:
    roi_rect: The region of interest.
    background: What to subtract from each ROI image before the centre of mass, estimated
        from its outer ``edge_width`` pixels: None (nothing, the default), "edge" (their
        median, a flat background) or "plane" (a least-squares plane a + b·x + c·y).
        The ROI intensities are not changed. See ``edge_background`` for when each fails.
    edge_width: Width in pixels of the ROI border used to estimate the background.

QoraFlow.analyzer.draw_figure ​

patch_rectangle ​

python
def patch_rectangle(image: npt.NDArray, y1: int, y2: int, x1: int, x2: int) -> 'Figure'

draw_intensity_figure ​

python
def draw_intensity_figure(data_df: pd.DataFrame) -> 'Figure'

draw_com_figure ​

python
def draw_com_figure(data_df: pd.DataFrame) -> 'Figure'

draw_intensity_diff_figure ​

python
def draw_intensity_diff_figure(data_df: pd.DataFrame) -> 'Figure'

draw_com_diff_figure ​

python
def draw_com_diff_figure(data_df: pd.DataFrame) -> 'Figure'

QoraFlow.analyzer.power_compare ​

power_compare ​

python
def power_compare(power_run: dict[int, int]) -> pd.DataFrame

plot_power_compare ​

python
def plot_power_compare(power_df: pd.DataFrame)

save_power_df ​

python
def save_power_df(power_df: pd.DataFrame, root: str)

QoraFlow.cli ​

CliError ​

python
class CliError(click.ClickException):
    def __init__(self, message)
Generic CLI error for graceful termination.

load_image ​

python
def load_image(run_n: int, scan_n: int, suffix: str='LND') -> np.ndarray
Loads image data for a specific run and scan.

process_image ​

python
def process_image(
    image: np.ndarray,
    log: bool,
    vmin: float | None,
    vmax: float | None,
) -> np.ndarray
Applies processing (log, clip) to the image.

interactive_roi_selection ​

python
def interactive_roi_selection(image: np.ndarray) -> tuple[int, int, int, int] | None
Handles the GUI interaction for ROI selection ONLY.

data_source_options ​

python
def data_source_options(func)

processing_options ​

python
def processing_options(func)

cli ​

python
@click.group()
def cli()
Unified CLI for data processing and visualization.

get_roi ​

python
@cli.command(name='get-roi')
@data_source_options
@processing_options
def get_roi(run_n: int, scan_n: int, log: bool, vmin: float | None, vmax: float | None)
Load, process, and interactively select an ROI.
Outputs selected coordinates (x0, y0, x1, y1).

QoraFlow.config.config ​

Global configuration manager with placeholder substitution.

ConfigManager ​

python
class ConfigManager:
    ...

ConfigManager.initialize ​

python
@classmethod
def initialize(cls, config_file: Path | str) -> None
Initialize the ConfigManager with a configuration file path.

Args:
    config_file (str | Path): Path to the configuration file.

ConfigManager.load_config ​

python
@classmethod
def load_config(cls, reload: bool=False) -> ExpConfig
Load the configuration, optionally reloading.

Args:
    reload (bool): Whether to force reload the config file.

Returns:
    ExpConfig: Configuration data object with placeholders resolved.

ConfigManager.save_config ​

python
@classmethod
def save_config(cls, config_dict: dict) -> None
Save the configuration dictionary back to the file.

Args:
    config_dict (dict): Configuration dictionary.

Raises:
    RuntimeError: If config file is not initialized.

ConfigManager.reload ​

python
@classmethod
def reload(cls) -> ExpConfig
Force reload the configuration.

ConfigManager.replace_placeholders ​

python
@staticmethod
def replace_placeholders(data: Any, context: dict, max_iterations=10)
Recursively replace placeholders in the given data structure.

Placeholders are in the format `${key.subkey}`.

Args:
    data (dict, list, str, or any): Data structure containing placeholders.
    context (dict): Context dictionary for placeholder resolution.
    max_iterations (int): Maximum number of iterations to prevent infinite loops.

Returns:
    Data structure with placeholders replaced.

ConfigManager.resolve_placeholder ​

python
@staticmethod
def resolve_placeholder(placeholder: str, context: dict)
Resolve a placeholder to its corresponding value from the context.

Args:
    placeholder (str): Placeholder string in the format `${key.subkey}`.
    context (dict): Context dictionary containing the values.

Returns:
    Value corresponding to the placeholder.

Raises:
    ValueError: If the placeholder cannot be resolved.

QoraFlow.config.definitions ​

This module provides specific classes for managing configuration parameters of an experiment.
It builds upon the generic configuration management provided by the base_config module.

Classes:
    - `ConfigurationParameters`: A class representing configuration parameters for an experiment.
    - `ConfigurationPaths`: A class to manage configuration paths for an experiment.
    - `ExperimentConfiguration`: A dataclass representing the complete configuration for an experiment,
    combining configuration parameters and paths.

ExpParams ​

python
class ExpParams(BaseModel):
    merge_num: int = 1
    chunk_size: int = 100
    hutch: Hutch = Hutch.EH1
    detector: Detector = Detector.JUNGFRAU2
    xray: Xray = Xray.HARD
    pump_setting: Hertz = Hertz.FIFTEEN
    x1: int = 0
    x2: int = 1
    y1: int = 2
    y2: int = 3
    sdd: float = 1.3
    dps: float = 7.5e-05
    beam_energy: float = 9.7
A dataclass to represent configuration parameters for an experiment.

sdd = 1.3m
dps = 75e-06m (75 um)
beam_energy = 9.7 keV
wavelength [A]

Attributes:
    hutch (Hutch): The hutch setting.
    detector (Detector): The detector setting.
    xray (Xray): The x-ray setting.
    pump_setting (Hertz): The pump setting.
    x1 (int): The x1 setting.
    x2 (int): The x2 setting.
    y1 (int): The y1 setting.
    y2 (int): The y2 setting.

ExpPaths ​

python
class ExpPaths(BaseModel):
    log_dir: Path = Path()
    load_dir: Path = Path()
    analysis_dir: Path = Path()
    raw_dark_dir: Path = Path()
    save_dark_dir: Path = Path()
    mat_dir: Path = Path('mat_files')
    processed_dir: Path = Path('processed_data')
    output_dir: Path = Path('output_data')
A class to manage configuration paths for an experiment.

Attributes:
    load_dir (Path): The load directory path.
    anaylsis_dir (Path): The save directory path.

ExpConfig ​

python
class ExpConfig(BaseModel):
    runs: list[int] = Field(default_factory=list)
    param: ExpParams = ExpParams()
    path: ExpPaths = ExpPaths()
A dataclass to represent the complete configuration for an experiment.

Attributes:
    runs (list[int]): The run numbers.
    param (ConfigurationParameters): The configuration parameters.
    path (ConfigurationPaths): The configuration paths.

QoraFlow.config.enums ​

This module defines enumerations for various experiment configurations.
It includes enums for detectors, hutch settings, X-ray types, and Hertz settings,
providing a structured approach to handle these configuration options.

Classes:
    - `Detector`: Enum representing different types of detectors.
    - `Hutch`: Enum representing different hutch settings.
    - `Xray`: Enum representing different X-ray types.
    - `Hertz`: Enum representing different Hertz settings.

Each enum class provides a string representation of the enum value.

Detector ​

python
class Detector(Enum):
    ...
Enum representing different types of detectors.

To add a new detector, follow these steps:

1. Add the detector name in all uppercase:
    <DETECTOR_NAME> = '<detector_name>'

Hutch ​

python
class Hutch(Enum):
    ...
Enum representing different hutch settings.

To add a new hutch, follow these steps:

1. Add the hutch name in all uppercase:
    <HUTCH_NAME> = '<hutch_name>'

Xray ​

python
class Xray(Enum):
    ...
Enum representing different X-ray types.

Hertz ​

python
class Hertz(Enum):
    ...
Enum representing different Hertz settings.

QoraFlow.datacheck.data_validate ​

find_outliers_mad ​

python
def find_outliers_mad(data, threshold=3)
Indices (as from np.where) of points whose modified z-score exceeds `threshold`.

validata_sizes ​

python
def validata_sizes(path: str | Path)

QoraFlow.filesystem ​

get_run_scan_dir ​

python
def get_run_scan_dir(
    mother: str | Path,
    run: int,
    scan: int | None = None,
    *,
    sub_path: str | None = None,
) -> Path
Generate the directory for a given run and scan number.
Robustly handles all combinations of scan and sub_path.

make_run_scan_dir ​

python
def make_run_scan_dir(
    mother: str | Path,
    run: int,
    scan: int,
    *,
    sub_path: str | Path | None = None,
) -> Path
Create a nested directory structure for the given run and scan numbers.
Sub_path will not be created, but it will be returned as a path.

get_run_nums ​

python
def get_run_nums(path: str | Path) -> list[int]
Get Run numbers from real directory

get_scan_nums ​

python
def get_scan_nums(path: str | Path, run_n: int) -> list[int]
Get Scan numbers from real directory

QoraFlow.functional ​

Functional Programming in Python
This module is designed to support the functional programming paradigm.

identity ​

python
def identity(*args)
A function that returns its input unchanged.

compose ​

python
def compose(*funcs: Callable)
Combines multiple functions from right to left.
This means the rightmost function is executed first,
and its result is passed as input to the next function.
In this way, creates a single function from multiple functions

compose(h, g, f)(x) is equivalent to h(g(f(x)))

pipe ​

python
def pipe(*funcs: Callable)
Combines multiple functions from left to right.
This means the leftmost function is executed first,
and its result is passed as input to the next function.
In this way, creates a single function from multiple functions

pipe(f, g, h)(x) is equivalent to h(g(f(x)))

batched ​

python
def batched(iterable, n)
Batch data into tuples of length n. The last batch may be shorter.

QoraFlow.gui.preprocess_gui ​

find_outliers_gui ​

python
def find_outliers_gui(y: npt.NDArray, x: npt.NDArray) -> float

find_outliers_run_scan_gui ​

python
def find_outliers_run_scan_gui(run: int, scan: int) -> float

RANSAC_regression_gui ​

python
def RANSAC_regression_gui(run: int, scan: int) -> None

QoraFlow.gui.roi_core ​

RoiSelector ​

python
class RoiSelector:
    def __init__(self)

RoiSelector.on_mouse_press ​

python
def on_mouse_press(self, event)

RoiSelector.on_mouse_release ​

python
def on_mouse_release(self, event)

RoiSelector.on_mouse_move ​

python
def on_mouse_move(self, event)

RoiSelector.select_roi ​

python
def select_roi(self, image: npt.NDArray) -> tuple[int, int, int, int] | None

select_roi_by_run_scan ​

python
def select_roi_by_run_scan(
    run: int,
    scan: int,
    index_mode: int | None = None,
) -> RoiRectangle | None

get_roi_auto ​

python
def get_roi_auto(image, half_width: int=5) -> RoiRectangle
get roi_rect by max pixel

QoraFlow.gui.select_roi ​

get_metadata_roi ​

python
def get_metadata_roi(scan_dir: str | Path, config: ExpConfig) -> RoiRectangle
Get Roi from metadata

select_roi ​

python
def select_roi(
    scan_dir: str | Path,
    config: ExpConfig,
    index_mode: int | None = None,
) -> RoiRectangle
Select ROI from GUI

auto_roi ​

python
def auto_roi(scan_dir: str | Path, config: ExpConfig, index_mode: int | None=None)
Get Roi based on the maximum value of the image

QoraFlow.integrator.core ​

CoreIntegrator ​

python
class CoreIntegrator:
    def __init__(
        self,
        LoaderStrategy: type[RawDataLoader],
        merge_num: int = 1,
        chunk_size: int = 100,
        preprocessor: dict[str, ImagesQbpmProcessor] | None = None,
        logger: Logger | None = None,
    ) -> None
Core pipeline class for integrating XFEL scan data.

Logic:
1. Group files by 'merge_num' (Experimental Repeats).
2. Load ALL images from these files (e.g., 3 files * 300 frames = 900 frames).
3. Group internal data by Delay (Time).
4. Preprocess (RANSAC, etc.) & Average -> Single Representative Image.

CoreIntegrator.run ​

python
def run(self, scan_dir: str | Path) -> dict[str, dict[str, npt.NDArray]]

CoreIntegrator.result ​

python
@property
def result(self) -> dict[str, dict[str, npt.NDArray]]

CoreIntegrator.save ​

python
def save(self, saver: SaverStrategy, run_n: int, scan_n: int)

QoraFlow.integrator.loader ​

RawDataLoader ​

python
class RawDataLoader(ABC):
    @abstractmethod
    def __init__(self, file: Path | str) -> None

RawDataLoader.get_data ​

python
@abstractmethod
def get_data(self) -> dict[str, npt.NDArray]
Retrieve data

RawDataLoader.get_chunked_data ​

python
@abstractmethod
def get_chunked_data(self, chunk_size: int) -> Iterator[dict[str, npt.NDArray]]
Yield Data

RawDataLoader.close ​

python
def close(self)
Optional close method

PalXFELLoader ​

python
class PalXFELLoader(RawDataLoader):
    def __init__(self, file: Path | str)
Load hdf5 file and remove unmatching data, supporting lazy image loading.

PalXFELLoader.close ​

python
def close(self)
Explicitly close the HDF5 file handle.

PalXFELLoader.get_chunked_data ​

python
def get_chunked_data(self, chunk_size: int=100) -> Iterator[dict[str, Any]]
Yields data in small chunks to prevent OOM errors.
This is the preferred method for large-scale processing.

PalXFELLoader.get_data ​

python
def get_data(self) -> dict[str, npt.NDArray]
Retrieves data by loading images from HDF5 lazily.
Returns: Images and qbpm data for both pump-on and pump-off states.

get_hdf5_images ​

python
def get_hdf5_images(file: str, config: ExpConfig) -> npt.NDArray
get images form hdf5

QoraFlow.integrator.saver ​

get_file_base_name ​

python
def get_file_base_name(run_n: int, scan_n: int) -> str
Return formatted file name: run=XXXX_scan=XXXX

SaverStrategy ​

python
class SaverStrategy(ABC):
    def __init__(self) -> None
Abstract Base Class for saving data dictionaries to files.

SaverStrategy.save ​

python
@abstractmethod
def save(
    self,
    run_n: int,
    scan_n: int,
    data_dict: dict[str, npt.NDArray],
    comment: str = '',
) -> None

SaverStrategy.file ​

python
@property
def file(self) -> str
Return the path of the last saved file as a string.

SaverStrategy.file_type ​

python
@property
@abstractmethod
def file_type(self) -> str
Return File Type extension (e.g., 'npz')

MatSaverStrategy ​

python
class MatSaverStrategy(SaverStrategy):
    ...
Saves 3D arrays from the data dictionary as individual .mat files.

MatSaverStrategy.file_type ​

python
@property
def file_type(self) -> str

MatSaverStrategy.save ​

python
def save(
    self,
    run_n: int,
    scan_n: int,
    data_dict: dict[str, npt.NDArray],
    comment: str = '',
) -> None

NpzSaverStrategy ​

python
class NpzSaverStrategy(SaverStrategy):
    ...
Saves the entire dictionary into a single compressed .npz file.

NpzSaverStrategy.file_type ​

python
@property
def file_type(self) -> str

NpzSaverStrategy.save ​

python
def save(
    self,
    run_n: int,
    scan_n: int,
    data_dict: dict[str, npt.NDArray],
    comment: str = '',
) -> None

TifSaverStrategy ​

python
class TifSaverStrategy(SaverStrategy):
    ...
Saves arrays from the data dictionary as individual .tif files.

TifSaverStrategy.file_type ​

python
@property
def file_type(self) -> str

TifSaverStrategy.save ​

python
def save(
    self,
    run_n: int,
    scan_n: int,
    data_dict: dict[str, npt.NDArray],
    comment: str = '',
) -> None

get_saver_strategy ​

python
def get_saver_strategy(file_type: str) -> SaverStrategy
Factory function to get SaverStrategy by file type.

QoraFlow.logger ​

setup_logger ​

python
def setup_logger(level='INFO') -> Logger
Configures the logger. This should be called ONLY ONCE at the application startup.

QoraFlow.mathematics ​

min_max_scale ​

python
def min_max_scale(arr) -> np.ndarray

axis_np2mat ​

python
def axis_np2mat(arr: npt.NDArray) -> npt.NDArray
Reorders the axes of an array to move the last two axes to the front.

This function is useful when treating the last two axes of an array
as matrix-like dimensions (e.g., for image patches, or per-pixel matrices),
and you want to operate on them first.

Parameters:
    arr (np.ndarray): Input array of at least 2 dimensions.

Returns:
    np.ndarray: Transposed array with the last two axes moved to the front.
                If the input has fewer than 2 dimensions, it is returned unchanged.

Example:
    >>> arr = np.zeros((2, 3, 4, 5))
    >>> axis_np2mat(arr).shape
    (4, 5, 2, 3)

axis_mat2np ​

python
def axis_mat2np(arr: npt.NDArray) -> npt.NDArray
Reorders the axes of an array to move the first two axes to the end.

This is the inverse operation of `axis_np2mat`, restoring the original
axis order before applying `axis_np2mat`.

Parameters:
    arr (np.ndarray): Input array of at least 2 dimensions.

Returns:
    np.ndarray: Transposed array with the first two axes moved to the end.
                If the input has fewer than 2 dimensions, it is returned unchanged.

Example:
    >>> arr = np.zeros((4, 5, 2, 3))
    >>> axis_mat2np(arr).shape
    (2, 3, 4, 5)

gaussian ​

python
def gaussian(x: npt.NDArray, a: float, mu: float, sig: float) -> npt.NDArray

integrate_FWHM ​

python
def integrate_FWHM(a: float, mu: float, sig: float) -> float

gaussian2d ​

python
def gaussian2d(
    xy,
    amplitude: float,
    x0: float,
    y0: float,
    sigma_x: float,
    sigma_y: float,
    theta: float,
    offset: float,
) -> npt.NDArray
Calculate the 2D Gaussian distribution at the given coordinates.

Parameters:
    xy (tuple of arrays): A tuple containing two arrays 'x' and 'y' representing the 2D coordinates.
    amplitude (float): The amplitude (peak value) of the Gaussian.
    x0 (float): The x-coordinate of the center of the Gaussian.
    y0 (float): The y-coordinate of the center of the Gaussian.
    sigma_x (float): The standard deviation in the x-direction.
    sigma_y (float): The standard deviation in the y-direction.
    theta (float): The rotation angle in radians. The Gaussian will be rotated counterclockwise by this angle.
    offset (float): The constant offset added to the Gaussian distribution.

Returns:
    numpy.ndarray: A 1D array containing the values of the 2D Gaussian distribution flattened into a 1D array.

integrate_fwhm_2d ​

python
def integrate_fwhm_2d(
    amplitude: float,
    xo: float,
    yo: float,
    sigma_x: float,
    sigma_y: float,
    theta: float,
    offset: float,
) -> float
Calculate the integral of a 2D Gaussian with an offset over its FWHM

get_wavelength ​

python
def get_wavelength(beam_energy: float) -> float

pixel_to_del_q ​

python
def pixel_to_del_q(pixels: npt.NDArray) -> npt.NDArray
pixel_to_q of the shift from pixels[0]: the chord between k_f at pixels[0] and at each pixel
when pixels[0] is where the detector normal meets it.

pixel_to_q ​

python
def pixel_to_q(pixels: npt.NDArray) -> npt.NDArray
two_theta = arctan(dps * pixels / sdd)
Q = (4 * pi / wavelength) * sin(two_theta / 2)
  ≈ (4 * pi / wavelength) * two_theta / 2
  ≈ pixels * (4 * pi / wavelength) * arctan(dps / sdd) / 2   (small angles)

mul_delta_q ​

python
def mul_delta_q(pixels: npt.NDArray) -> npt.NDArray
|Δq| [1/Å] of a shift of `pixels` from pixels[0], detector normal to the scattered beam.

One pixel turns the scattered wavevector by delta = arctan(dps / sdd), so
|Δq| = k * delta = (2 * pi / wavelength) * delta per pixel, along either detector axis
(the per-pixel factor in `pixel_to_q`). This is |Δq|, not the change of |q|: along 2θ
that is |Δq| * cos(theta), and across the scattering plane it is zero to first order.

non_outlier_indices_percentile ​

python
def non_outlier_indices_percentile(
    arr: npt.NDArray,
    lower_percentile: float,
    upper_percentile: float,
) -> npt.NDArray[np.bool_]
Get the indices of non-outliers in a NumPy array.

Parameters:
arr (np.ndarray): Input NumPy array containing data.

Returns:
np.ndarray: Boolean array with the same shape as 'arr' where True indicates non-outlier data points.

QoraFlow.preprocessor.generic_preprocessors ​

ransac_regression ​

python
def ransac_regression(
    y: np.ndarray,
    x: np.ndarray,
    min_samples: int | None = None,
) -> tuple[npt.NDArray[np.bool_], npt.NDArray, npt.NDArray]
Perform RANSAC (Random Sample Consensus) regression to identify inliers and estimate the regression model.

Parameters:
- y (np.ndarray): The target variable array.
- x (np.ndarray): The feature variable array.
- min_samples (int, optional): The minimum number of samples to fit the model. Default is 3.

Returns:
- tuple[npt.NDArray[np.bool_], npt.NDArray, npt.NDArray]:
    A tuple containing the inlier mask, coefficient, and intercept of the linear model.

get_linear_regression_confidence_bounds ​

python
def get_linear_regression_confidence_bounds(
    y: npt.NDArray,
    x: npt.NDArray,
    sigma: float,
) -> npt.NDArray
Get lower and upper bounds for data points based on their confidence interval in a linear regression model.

Statistical Explanation:
This function applies the principle of propagation of uncertainty. It fits a linear
regression model y = mx + b, calculates the standard errors of the model parameters
m and b, and uses these to generate prediction intervals for identifying outliers.

The prediction interval is calculated using the formula:
y ± sqrt((m_err * x)^2 + b_err^2) * sigma
where m_err and b_err are the standard errors of m and b respectively.

Parameters:
y (NDArray): Dependent variable data. Shape: (N,)
x (NDArray): Independent variable data. Shape: (N,)
sigma (float): Number of standard deviations for the confidence interval. Default is 3.0.

Returns:
lowerbound (NDArray): Lower bounds of y
upperbound (NDArray): Upper bounds of y
y_fit (NDArray): Fitted y

Note:
This method assumes a linear relationship in the data. For strong non-linearities,
a different approach may be necessary.

filter_images_qbpm_by_linear_model ​

python
def filter_images_qbpm_by_linear_model(
    images: npt.NDArray,
    qbpm: npt.NDArray,
    sigma: float,
) -> tuple[npt.NDArray, npt.NDArray]
Filter images based on the confidence interval of their intensities
using a linear regression model with QBPM values.

This function computes the total intensity of each image, applies a linear regression model
to the intensity and QBPM values, and generates a mask to filter out images whose intensities
fall outside the specified confidence interval.

Parameters:
images (NDArray): Array of images. Shape: (N, H, W), where N is the number of images,
    and H and W are the height and width of each image.
qbpm (NDArray): Array of QBPM (Quadrature Balanced Photodetector Measurements) values. Shape: (N,)
sigma (float): Number of standard deviations for the confidence interval.

Returns:
tuple[NDArray, NDArray]: Tuple of filtered intensities and QBPM values.
    - Filtered intensities (NDArray):
        Array of intensities within the confidence interval. Shape: (M,), where M <= N.
    - Filtered qbpm (NDArray):
        Array of QBPM values corresponding to the filtered intensities. Shape: (M,), where M <= N.

Note:
This method uses the `get_linear_regression_confidence_lower_upper_bound` function to generate the mask based
on the linear regression model and confidence interval.

div_images_by_qbpm ​

python
def div_images_by_qbpm(images: npt.NDArray, qbpm: npt.NDArray) -> npt.NDArray
Divide images by qbpm.

Parameters:
images (NDArray): Array of images.
    shape=(N, H, W), where N is the number of images,
    and H and W are the height and width of each image.
qbpm (NDArray): Array of QBPM (Quadrature Balanced Photodetector Measurements) values. Shape: (N,)

Returns:
NDArray: Images that divided by qbpm

subtract_dark ​

python
def subtract_dark(images: npt.NDArray) -> npt.NDArray
Subtract dark background. Uses caching for performance.

Raises:
    FileNotFoundError, ValueError: the configured dark image is missing or unusable.

add_bias ​

python
def add_bias(images: npt.NDArray)

equalize_brightness ​

python
def equalize_brightness(images: np.ndarray) -> np.ndarray
Equalize the brightness of each image in the 3D array while maintaining the overall average brightness.

Parameters:
- images: np.ndarray, 3D array of images (num_images, height, width).

Returns:
- np.ndarray: The brightness-equalized images.

QoraFlow.preprocessor.image_qbpm_preprocessor ​

make_qbpm_roi_normalizer ​

python
def make_qbpm_roi_normalizer(roi_rect: RoiRectangle | None) -> ImagesQbpmProcessor

make_thresholder ​

python
def make_thresholder(n: float) -> ImagesQbpmProcessor

no_negative ​

python
def no_negative(images_qbpm: ImagesQbpm) -> ImagesQbpm
No value below zero

shift_to_positive ​

python
def shift_to_positive(images_qbpm: ImagesQbpm) -> ImagesQbpm
Shift the images to ensure all values are non-negative by adding a bias.

This function adds a bias to the images to ensure that all pixel values are non-negative.
The bias is calculated as the absolute value of the minimum pixel value in the images.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[Images, Qbpm]: A tuple containing the shifted images and the original QBPM values.

subtract_dark_background ​

python
def subtract_dark_background(images_qbpm: ImagesQbpm) -> ImagesQbpm
Remove the dark background from the images.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[Images, Qbpm]: The images with dark background removed and the original Qbpm values.

normalize_images_by_qbpm ​

python
def normalize_images_by_qbpm(images_qbpm: ImagesQbpm) -> ImagesQbpm
Normalize the images by the Qbpm values.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[Images, Qbpm]: The normalized images and the original Qbpm values.

lsb_quantization ​

python
def lsb_quantization(images_qbpm: ImagesQbpm) -> ImagesQbpm
Apply LSB quantization to the images while keeping the Qbpm values unchanged.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[np.ndarray, np.ndarray]: A tuple containing the LSB-quantized images and the original Qbpm values.

remove_outliers_using_ransac ​

python
def remove_outliers_using_ransac(images_qbpm: ImagesQbpm) -> ImagesQbpm
Remove outliers from the images and Qbpm values using RANSAC regression.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[Images, Qbpm]: The images and Qbpm values with outliers removed.

create_ransac_roi_outlier_remover ​

python
def create_ransac_roi_outlier_remover(roi_rect: RoiRectangle) -> ImagesQbpmProcessor
Create a function to remove outliers using a linear model with a given sigma.

Parameters:
- sigma: float, the sigma value for the outlier removal.

Returns:
- ImageQbpmProcessor: A function that takes ImagesQbpm and returns the filtered ImagesQbpm.

equalize_intensities ​

python
def equalize_intensities(images_qbpm: ImagesQbpm) -> ImagesQbpm
Equalize the intensities of the images while keeping the Qbpm values unchanged.

Parameters:
- images_qbpm (tuple[Images, Qbpm]): tuple of Images and Qbpm

Returns:
- tuple[np.ndarray, np.ndarray]: A tuple containing the brightness-equalized images and the original Qbpm values.

create_linear_model_outlier_remover ​

python
def create_linear_model_outlier_remover(sigma) -> ImagesQbpmProcessor
Create a function to remove outliers using a linear model with a given sigma.

Parameters:
- sigma: float, the sigma value for the outlier removal.

Returns:
- ImageQbpmProcessor: A function that takes ImagesQbpm and returns the filtered ImagesQbpm.

Each project section is built from that project’s docs/ directory.

Isaac Yong