dran.fitting package

Contents

dran.fitting package#

Submodules#

dran.fitting.baseline_correction module#

dran.fitting.baseline_correction.fit_polynomial(x, y, deg)[source]#

Fit a polynomial model to data. Returns the polynomial coefficients and the evaluated model values at the input x coordinates.

Parameters:
Return type:

Tuple[ndarray, ndarray]

dran.fitting.baseline_correction.calc_residual_and_rms(model, data)[source]#

Compute residuals and root-mean-square error between model and data. Returns the residual array and the RMS error value.

Parameters:
Return type:

Tuple[ndarray, float]

dran.fitting.baseline_correction.correct_baseline_linear(x, y, baseline_indices, log)[source]#

Fits linear drift over baseline indices, subtracts it from y. :returns: y_corrected, y_corrected_spline, baseline_coeffs, baseline_residual, baseline_rms

Parameters:
Return type:

Tuple[ndarray, ndarray, ndarray, ndarray, float]

dran.fitting.baseline_windows module#

dran.fitting.baseline_windows.filter_invalid_minima(local_max_positions, local_min_positions, max_points, log)[source]#

Removes minima that fall inside max windows.

Parameters:
Return type:

ndarray

dran.fitting.baseline_windows.build_baseline_windows(x, left_mins, right_mins, left_hpbw_idx, right_hpbw_idx, left_fnbw_idx, right_fnbw_idx, max_points, log)[source]#

Builds baseline sample indices by selecting blocks around minima. :returns: (indices_sorted_unique, flag)

Parameters:
Return type:

Tuple[ndarray, int]

dran.fitting.config module#

class dran.fitting.config.DualBeamQCConfig(step_sigma_mult: float = 12.0, step_count_max: int = 6, max_gap_frac: float = 0.1, min_points_per_beam: int = 20, max_edge_frac: float = 0.2, max_resid_to_baseline_rms: float = 2.5, min_snr: float = 2.0, max_amp_ratio: float = 4.0)[source]#

Bases: object

Parameters:

  • step_sigma_mult (float)

  • step_count_max (int)

  • max_gap_frac (float)

  • min_points_per_beam (int)

  • max_edge_frac (float)

  • max_resid_to_baseline_rms (float)

  • min_snr (float)

  • max_amp_ratio (float)

step_sigma_mult: float = 12.0#
step_count_max: int = 6#
max_gap_frac: float = 0.1#
min_points_per_beam: int = 20#
max_edge_frac: float = 0.2#
max_resid_to_baseline_rms: float = 2.5#
min_snr: float = 2.0#
max_amp_ratio: float = 4.0#

dran.fitting.gaussian_fit module#

dran.fitting.gaussian_fit.gauss_plus_linear_baseline(x, amp, mu, hpbw, a, c)[source]#

Gaussian main lobe with a first-order baseline: amp*exp(…) + a*x + c

Notes: - HPBW is converted to sigma via sigma = HPBW / (2*sqrt(2*ln(2))).

Parameters:
Return type:

ndarray

dran.fitting.gaussian_fit.fit_gaussian_test(x, y, p0, log)[source]#

Quick Gaussian fit used as a quality gate.

Returns:

(coeffs, model, flag) flag is None on success, else an int.

Parameters:
Return type:

Tuple[ndarray, ndarray, int | None]

dran.fitting.locate_fitting_blocks module#

dran.fitting.models module#

class dran.fitting.models.ResidualFit(rms, residuals)[source]#

Bases: object

Container for Residual fit results.

Parameters:
rms: float#
residuals: ndarray[Any, dtype[floating]]#
class dran.fitting.models.CleanedScan(x: numpy.ndarray, y: numpy.ndarray, rms: float, residual: numpy.ndarray, spline_max: float, spline: numpy.ndarray, points_deleted: numpy.ndarray, raw_rms: numpy.ndarray, raw_residuals: numpy.ndarray, flag: int = 0, message: str = '')[source]#

Bases: object

Parameters:
x: ndarray#
y: ndarray#
rms: float#
residual: ndarray#
spline_max: float#
spline: ndarray#
points_deleted: ndarray#
raw_rms: ndarray#
raw_residuals: ndarray#
flag: int#
message: str#
class dran.fitting.models.BeamFitResult(qc: dran.fitting.plot_qc.ScanQualityResult = <factory>, peak_coeffs: numpy.ndarray | None = None, peak_model: numpy.ndarray | None = None, ta_peak: float = nan, ta_peak_err: float = nan, peak_loc_index: int | None = None, baseline_coeffs: numpy.ndarray | None = None, baseline_residual: numpy.ndarray | None = None, baseline_rms: float = nan, baseline_window_indices: numpy.ndarray | None = None, y_corrected: numpy.ndarray | None = None, y_corrected_spline: numpy.ndarray | None = None, clean_rms: float = nan, flag: int = 0, message: str = '')[source]#

Bases: object

Parameters:
qc: ScanQualityResult#
peak_coeffs: ndarray | None#
peak_model: ndarray | None#
ta_peak: float#
ta_peak_err: float#
peak_loc_index: int | None#
baseline_coeffs: ndarray | None#
baseline_residual: ndarray | None#
baseline_rms: float#
baseline_window_indices: ndarray | None#
y_corrected: ndarray | None#
y_corrected_spline: ndarray | None#
clean_rms: float#
flag: int#
message: str#
class dran.fitting.models.DualBeamQCResult(is_bad: bool = True, flag: int = 0, reasons: list[str] = <factory>, step_max: float = 0.0, step_count: int = 0, max_gap_frac: float = 0.0, snr_a: float = 0.0, snr_b: float = 0.0, resid_rms_a: float = 0.0, resid_rms_b: float = 0.0)[source]#

Bases: object

Parameters:
is_bad: bool = True#
flag: int = 0#
reasons: list[str]#
step_max: float = 0.0#
step_count: int = 0#
max_gap_frac: float = 0.0#
snr_a: float = 0.0#
snr_b: float = 0.0#
resid_rms_a: float = 0.0#
resid_rms_b: float = 0.0#
to_dict()[source]#
Return type:

dict

class dran.fitting.models.DualBeamFitResult(qc: dran.fitting.models.DualBeamQCResult = <factory>, apeak_residuals: float = nan, ata_peak: float = nan, ata_peak_err: float = nan, apeak_loc_index: int | None = None, bpeak_residuals: float = nan, bta_peak: float = nan, bta_peak_err: float = nan, bpeak_loc_index: int | None = None, baseline_coeffs: numpy.ndarray | None = None, baseline_residuals: numpy.ndarray | None = None, baseline_rms: float = nan, baseline_window_indices: numpy.ndarray | None = None, y_corrected: numpy.ndarray | None = None, y_corrected_spline: numpy.ndarray | None = None, clean_rms: float = nan, flag: int = 0, message: str = '')[source]#

Bases: object

Parameters:
qc: DualBeamQCResult#
apeak_residuals: float#
ata_peak: float#
ata_peak_err: float#
apeak_loc_index: int | None#
bpeak_residuals: float#
bta_peak: float#
bta_peak_err: float#
bpeak_loc_index: int | None#
baseline_coeffs: ndarray | None#
baseline_residuals: ndarray | None#
baseline_rms: float#
baseline_window_indices: ndarray | None#
y_corrected: ndarray | None#
y_corrected_spline: ndarray | None#
clean_rms: float#
flag: int#
message: str#
class dran.fitting.models.IterativeCleaningResult(x: numpy.ndarray, y: numpy.ndarray, rms: float, residual: numpy.ndarray, spline_max: float, spline: numpy.ndarray, points_deleted: int, flag: int = 0, names: list[Any] | None = None, message: str = '')[source]#

Bases: object

Parameters:
x: ndarray#
y: ndarray#
rms: float#
residual: ndarray#
spline_max: float#
spline: ndarray#
points_deleted: int#
flag: int#
names: list[Any] | None#
message: str#
dran.fitting.models.sig_to_noise(ta_peak, residual, log)[source]#

Calculate the signal to noise ratio. i.e. Amplitude / (stdDeviation of noise) Taken from paper on ‘Signal to Noise Ratio (SNR) Enhancement Comparison of Impulse-, Coding- and Novel Linear-Frequency-Chirp-Based Optical Time Domain Reflectometry (OTDR) for Passive Optical Network (PON) Monitoring Based on Unique Combinations of Wavelength Selective Mirrors’

Photonics 2014, 1, 33-46; doi:10.3390/photonics1010033 https://www.mdpi.com/2304-6732/1/1/33 https://www.mdpi.com/68484

Parameters:

  • signalPeak (float) – The maximum valueof a desired signal

  • noise (array) – array of fit residuals

  • log (object) – file logging object

  • ta_peak (float)

  • residual (ndarray)

Returns:

signal to noise ratio

Return type:

sig2noise (float)

Returns SNR, or None if it cannot be computed.

dran.fitting.peak_fitting module#

dran.fitting.peak_fitting.peak_window_indices(x, y_spline, hfnbw, cut_fraction)[source]#

Selects indices within |x| <= hfnbw and spline >= cut_fraction * max(spline).

Parameters:
Return type:

ndarray

dran.fitting.peak_fitting.fit_quadratic_peak(x, y, log)[source]#

Quadratic fit. Returns coeffs, model, rms.

Parameters:
Return type:

Tuple[ndarray, ndarray, float]

dran.fitting.peak_fitting.calc_residual(model, data)[source]#

Calculate the residual and rms between the model and the data.

Parameters:

  • model (array) – 1D array containing the model data

  • data (array) – 1D array containing the raw data

  • log (object) – file logging object

Returns:

  • res (1d array) – the residual

  • rms (int) – the rms value

dran.fitting.peak_fitting.peak_location_index(model)[source]#

Return the index of the maximum value in a model array. Identifies the peak location by finding the position of the largest element.

Parameters:

model (ndarray)

Return type:

int

dran.fitting.pipeline module#

dran.fitting.pipeline.find_step_locations(y, sigma, sigma_mult)[source]#

Identify step change locations in a signal. Finds indices where the absolute first difference exceeds a multiple of the noise level.

Parameters:
Return type:

ndarray

dran.fitting.pipeline.plot_dual_beam_final(*, x, y, y_corr, a_idx, b_idx, a_model_x, a_model_y, b_model_x, b_model_y, a_label, b_label, qc_flag, qc_reasons, base_sigma, save, paths, log)[source]#

Generate and save the final dual-beam diagnostic plot. Detects step edges in the corrected scan, builds plotting series for the corrected data, beam windows, fitted beam models, and optional step markers, then calls plot_diagnostics to render the final QC plot.

Parameters:
Return type:

None

dran.fitting.pipeline.robust_sigma(y)[source]#

Estimate robust noise level using the median absolute deviation. Falls back to standard deviation for small samples and returns zero for empty input.

Parameters:

y (ndarray)

Return type:

float

dran.fitting.pipeline.step_metrics(y, sigma, sigma_mult)[source]#

Compute step change metrics for a signal. Returns the maximum absolute step size and the count of steps exceeding a noise-scaled threshold.

Parameters:
Return type:

tuple[float, int]

dran.fitting.pipeline.max_gap_fraction(x)[source]#

Compute the largest sampling gap relative to the median spacing. Returns the ratio of the maximum gap to the median step size, or zero for insufficient or degenerate input.

Parameters:

x (ndarray)

Return type:

float

dran.fitting.pipeline.edge_distance_fraction(peak_x, window_left, window_right)[source]#

Measure normalized distance of a peak from the window edges. Returns the minimum distance to either edge divided by the window width, with a fallback value for invalid windows.

Parameters:
Return type:

float

dran.fitting.pipeline.compute_dual_beam_qc(x, y_corrected, baseline_sigma, a_idx, b_idx, a_peak, a_peak_x, a_resid, b_peak, b_peak_x, b_resid, a_window, b_window, cfg)[source]#

Evaluate dual-beam scan quality and return QC diagnostics. Computes robust noise, step discontinuities, sampling gaps, per-beam point counts, residual RMS, SNR, amplitude balance, edge proximity, and peak sign consistency. Produces a QC flag and reason list, plus summary metrics used for logging and plotting.

Parameters:
Return type:

DualBeamQCResult

dran.fitting.pipeline.dual_beam_qc_to_scan_quality(qc)[source]#
Parameters:

qc (DualBeamQCResult)

Return type:

ScanQualityResult

dran.fitting.pipeline.fit_scan(x, y, band, hfnbw, hhpbw, force, log, save, theofit, autofit, paths)[source]#

Refactored pipeline wrapper around the existing logic. Long, scan-specific heuristics stay in this file, while reusable parts live in modules.

This version removes sys.exit calls. It returns BeamFitResult with flag and message.

Parameters:
Return type:

BeamFitResult

dran.fitting.pipeline.fit_scan_db(x, y, band, hfnbw, hhpbw, force, log, save, theofit, autofit, factor, paths)[source]#

Refactored pipeline wrapper around the existing logic. Long, scan-specific heuristics stay in this file, while reusable parts live in modules.

This version removes sys.exit calls. It returns DualBeamFitResult with flag and message.

Parameters:
Return type:

DualBeamFitResult

dran.fitting.plot_qc module#

class dran.fitting.plot_qc.ScanQualityResult(ok: bool = False, flag: int = 0, message: str = 'QC not computed', metrics: Dict[str, Any]=<factory>)[source]#

Bases: object

Parameters:
ok: bool#
flag: int#
message: str#
metrics: Dict[str, Any]#
to_dict()[source]#
Return type:

dict

dran.fitting.plot_qc.check_scan_quality(x, y_corrected, y_spline, hfnbw, hhpbw, peak_coeffs, peak_x, peak_y, peak_model, band)[source]#
Parameters:
Return type:

ScanQualityResult

dran.fitting.plotting module#

dran.fitting.plotting.plot_diagnostics(plots, paths, log, save_path='', plot_type='diagnostic', suffix='')[source]#

plots: sequence of dicts with keys: x, y, lab, fmt Optional keys per dict: axlinet, axlineb

Parameters:
Return type:

None

dran.fitting.plotting.plot_fail(x, y, paths, message, log, save_path, flag, fmt='r')[source]#

Plot a single series to document a failure and save it.

This reproduces your old behavior, but without returning a long tuple.

Parameters:
Return type:

None

dran.fitting.rfi_cleaning module#

dran.fitting.rfi_cleaning.clean_rfi_sigma_clip(x, y, log)[source]#

Sort x, build a spline, sigma-clip residuals, return inliers plus artefacts.

Parameters:
Return type:

CleanedScan

dran.fitting.rfi_cleaning.clean_data(x, y, log=None)[source]#

Clean a drift scan using iterative RFI rejection and refitting.

Pipeline 1) Spline fit to y 2) Compute residuals and RMS 3) Iteratively reject outliers and refit until RMS no longer improves

Parameters:
Returns:

Container holding cleaned arrays and diagnostic products.

Return type:

CleanedScan

dran.fitting.rfi_cleaning.clean_data_iterative_fitting(x, y, res, rms, log, x2=None, cut=3.0, max_iters=25)[source]#

Iteratively remove RFI-like outliers and refit a spline until RMS no longer improves.

A point is kept if |residual| <= cut * rms.

Parameters:

  • x (ndarray | Sequence[float]) – 1D arrays of equal length.

  • y (ndarray | Sequence[float]) – 1D arrays of equal length.

  • res (ndarray | Sequence[float]) – 1D residual array aligned with x and y (for the current iteration).

  • rms (float) – Current RMS estimate used for thresholding.

  • log (Logger) – Logger instance.

  • x2 (Sequence[Any] | None) – Optional aligned metadata (filenames, IDs). If provided, it is filtered too.

  • cut (float) – Residual threshold multiplier.

  • max_iters (int) – Safety cap on number of iterations.

Returns:

  • Without x2 – finalX, finalY, finalRms, finalRes, finalMaxSpl, finalSplinedData, pointsDeleted

  • With x2 – finalX, finalY, finalRms, finalRes, finalMaxSpl, finalSplinedData, pointsDeleted, finalNames

Return type:

IterativeCleaningResult

dran.fitting.rfi_cleaning.calc_residual(model, data, log)[source]#

Calculate the residual and rms between the model and the data.

Parameters:

  • model (array) – 1D array containing the model data

  • data (array) – 1D array containing the raw data

  • log (object) – file logging object

Returns:

  • res (1d array) – the residual

  • rms (int) – the rms value

Return type:

ResidualFit

dran.fitting.spline_models module#

dran.fitting.spline_models.spline_fit_1d(y, anchor_points=9, order=3, log=None)[source]#

Spline smoothing over y values using index space.

Parameters:
Return type:

ndarray

dran.fitting.validation module#

dran.fitting.validation.validate_xy(x, y, log)[source]#
Returns:

flag int. 0 means pass.

Raises:

ValueError on invalid inputs. –

Parameters:
Return type:

int

Module contents#

Contents