Optics Measurements

Measure Optics

This module contains high-level functions to manage most functionality of optics_measurements. It provides functions to compute various lattice optics parameters from frequency spectra.

omc3.optics_measurements.measure_optics.chromatic_beating(input_files: InputFiles, measure_input: DotDict, tune_dict: tune.TuneDict) → None

Main function to compute chromatic optics beating.

Parameters:

• input_files -- InputFiles object containing frequency spectra files (linx/y).

• measure_input -- OpticsInput object containing analysis settings.

• tune_dict -- TuneDict object containing measured tunes.

omc3.optics_measurements.measure_optics.copy_calibration_files(outputdir: str | Path, calibrationdir: str | Path | None) → dict[str, TfsDataFrame] | None

Copies calibration files from calibrationdir to outputdir, then reads them and returns the data.

Parameters:

• outputdir -- path to the output directory.

• calibrationdir -- path to the calibration directory. If None, returns None.

Returns:

A dictionary mapping planes to calibration DataFrames, or None if no calibration directory is given.

omc3.optics_measurements.measure_optics.measure_optics(input_files: InputFiles, measure_input: DotDict) → None

Main function to compute various lattice optics parameters from frequency spectra.

Parameters:

• input_files -- InputFiles object containing frequency spectra files (linx/y).

• measure_input -- OpticsInput object containing analysis settings.

Beta from Amplitude

This module contains some of the beta calculation related functionality of optics_measurements. It provides functions to calculate beta functions from amplitude data.

omc3.optics_measurements.beta_from_amplitude.add_rescaled_beta_columns(df: DataFrame, ratio: float, plane: str) → DataFrame

Adds rescaled beta and error columns to the DataFrame using the given ratio.

Parameters:

• df -- DataFrame of beta from amplitude to extend.

• ratio -- phase-to-amplitude rescaling factor.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

The input DataFrame with added rescaled columns.

omc3.optics_measurements.beta_from_amplitude.beta_from_amplitude(meas_input: DotDict, input_files: InputFiles, plane: str, tunes: TuneDict) → pd.DataFrame

Calculates beta function from measured amplitudes across input files.

Parameters:

• meas_input -- OpticsInput object.

• input_files -- InputFiles object containing measurement data.

• plane -- marking the horizontal or vertical plane, X or Y.

• tunes -- TuneDict contains measured tunes.

Returns:

A DataFrame with measured beta, model beta, errors, and beating columns.

omc3.optics_measurements.beta_from_amplitude.calculate(meas_input: DotDict, input_files: InputFiles, tune_dict: TuneDict, beta_phase: pd.DataFrame, header_dict: dict, plane: str) → float

Calculates beta and returns the phase to amplitude ratio. Also writes the following TfsFiles: f"{AMP_BETA_NAME}{plane.lower()}{EXT}"

Parameters:

• meas_input -- OpticsInput object.

• input_files -- InputFiles object contains measurement files.

• tune_dict -- TuneDict contains measured tunes.

• beta_phase -- Dataframe containing beta functions measured from phase.

• header_dict -- dictionary of header items common for all output files.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

The phase-to-amplitude beta ratio (rescaling factor).

omc3.optics_measurements.beta_from_amplitude.phase_to_amp_ratio(measure_input: DotDict, beta_phase: pd.DataFrame, beta_amp: pd.DataFrame, plane: str) → float

Computes the mean ratio of phase-beta to amplitude-beta for arc BPMs.

Parameters:

• measure_input -- OpticsInput object.

• beta_phase -- Dataframe with beta from phase measurement.

• beta_amp -- Dataframe with beta from amplitude measurement.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

The mean ratio of phase-beta over amplitude-beta in arc BPMs.

Beta from Phase

Implements beta and alpha function calculation from phase advance measurements

Two implementations are provided:

• The Analytical N-BPM method, described in:

  Langner, A. et al., “Analytical N beam position monitor method”, Phys. Rev. Accel. Beams 20, 111002 (2017). https://cds.cern.ch/record/2307554

  It uses all valid BPM triplet combinations within a configurable range, propagates both phase measurement uncertainties and systematic lattice errors (quadrupole field gradient errors, quadrupole longitudinal misalignments, BPM longitudinal misalignments and sextupole horizontal misalignments) through an analytical covariance matrix.

• The 3-BPM method, originally from:

  Castro, P. “Luminosity and beta-function measurements at the electron-positron collider ring LEP” Ph.D Thesis, University of Valencia, 1996. https://repository.cern/records/eny2v-4y338

  The initial, simple version using only adjacent BPM triplets, with phase measurement uncertainties only.

class omc3.optics_measurements.beta_from_phase.Methods(*values)

Bases: StrEnum

A_NBPM: str = 'Analytical N-BPM method'

NO_ERR: str = 'No Errors'

THREE_BPM: str = '3BPM method'

omc3.optics_measurements.beta_from_phase.calculate(meas_input: DotDict, tunes: TuneDict, phase_dict: PhaseDict, header_dict: dict[str, Any], plane: str) → tuple[tfs.TfsDataFrame, dict[str, Any]]

Calculates betas and alphas from phase advances. This will dispatch to use either the n_bpm_method or three_bpm_method, based on the passed option.

The measured and model tunes passed downstream are selected based on the compensation model: free (natural) tunes Q/QM are used when no compensation was applied, while driven tunes QF/QFM are used otherwise.

Parameters:

• meas_input -- OpticsInput object with optics CLI options.

• tunes -- TuneDict contains model and measured tunes (incl. free motion).

• phase_dict -- PhaseDict containing phase advances between BPMs from model, measurements (incl. errors).

• header_dict -- dictionary of header items common for all output files.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

Tuple of the results TfsDataFrame (betas, alphas, errors, delta columns) and the output file header dictionary.

omc3.optics_measurements.beta_from_phase.calculate_beta_alpha_from_single_combination(c: tuple[int, int], sin_squared_elements: ndarray, outer_elmts: DataFrame, cot_model: ndarray, cot_meas: ndarray, outer_meas_phase_adv: Series, probed_bpm_name: str, betmdl1: float, alfmdl1: float, range_of_bpms: int) → tuple[float, float, ndarray, ndarray]

Computes β₁ and α₁ for one BPM triplet (probed BPM 1, reference BPMs x and y) and builds the corresponding rows of the Jacobian matrices T_β and T_α used for covariance propagation in n_bpm_method.

Note about the Jacobian structure (Eq. 22): T = (T^φ T^K T^s) The full Jacobian is split into three blocks: phase uncertainty T^φ, quadrupole field errors T^K and BPM longitudinal misalignments T^s. Sextupole horizontal misalignments enter into T^K via Eq (13-14) as effective quadrupole errors due to feed-down. Here K2L comes into play. Longitudinal quadrupole misalignments are decomposed into two thin-lens elements at the quadrupole edges (section II.B, Fig. 3): forward (KdS) and backward (mKdS).

The returned betaline and alfaline are 1D vectors of length (2m+1, 4*n_elements), each one being a row of T_β / T_α. Their layout matches the diag vector in n_bpm_method (see below for the blocks).

Parameters:

• c -- tuple, pair of relative indices (ix, iy) for the two reference BPMs of this triplet. Indices run 0 … 2m with the probed BPM at position m.

• sin_squared_elements -- (n_elements * n_bpms_window) array of squared sines of the model phase advance from every element l to every window BPM j. In there, entry [l, j] = sin²(φ_l - φ_mdl_j). Pre-computed in n_bpm_method.

• outer_elmts -- DataFrame of lattice elements in the window, which contains the BETA, K1L, K2L, dK1, dX, KdS and mKdS column (after loading systematic errors).

• cot_model -- 1-D array of model cotangents cot(φ_mdl_1j - φ_mdl_11), length 2m+1; where entry m is - by construction (probed BPM to itself).

• cot_meas -- 1D array of measured cotangents cot(φ_meas_1j), of length 2m+1.

• outer_meas_phase_adv -- measured phase advances (radians) from the probed BPM to each BPM in the window (index carries BPM names used for element lookup).

• probed_bpm_name -- The name of BPM 1 whose β and α are being estimated.

• betmdl1 (float) -- model β at the probed BPM.

• alfmdl1 (float) -- model α at the probed BPM.

• range_of_bpms -- total window width.

Returns:

A tuple (β₁, α₁, betaline, alfaline) where betaline and alfaline are the Jacobian rows for this combination, to be stacked into the complete Jacobian matrices T_β and T_α inside of n_bpm_method.

omc3.optics_measurements.beta_from_phase.get_elements_with_errors(meas_input: DotDict, plane: str) → tuple[pd.DataFrame, str]

Loads the accelerator lattice elements with their systematic error variances, ready to be windowed per probed BPM in n_bpm_method.

Reads the machine-specific error definition file (error_defs_file) and populates four variance columns on the element table via _assign_uncertainties: dK1, dX, KdS, mKdS.

Only elements with at least one non-zero error source are retained (see _assign_uncertainties for the filtering logic). The returned error_method string is used downstream to label the output file header and the RMS beta-beat log message.

Parameters:

• meas_input -- OpticsInput object with optics CLI options. Must have

• populated. (accelerator.error_defs_file set and accelerator.elements)

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

• elements DataFrame indexed by element name with columns:

  S, K1L, K2L, MU{plane}, BET{plane}, dK1, dX, KdS, mKdS, BPMdS.

• error_method string: Methods.A_NBPM if at least one systematic error

  was assigned, Methods.NO_ERR otherwise.

Return type:

Tuple of

Raises:

OSError -- if error_defs_file is None (no error definition file configured).

omc3.optics_measurements.beta_from_phase.n_bpm_method(meas_input: DotDict, phase: PhaseDict, plane: str, meas_and_mdl_tunes: tuple[float, float]) → tuple[tfs.TfsDataFrame, str]

Calculates betas and alphas using all BPM triplet combinations within a sliding window of range_of_bpms BPMs centred on each probed BPM. Please refer to theory at https://cds.cern.ch/record/2307554.

For each probed BPM i, a total of m = range_of_bpms // 2 neighbours are selected on each side. Every valid pair (j, k) within those neighbours contributes to form a BPM triplet (i, j, k) from which βi and ⍺i are estimated individually via the function calculate_beta_alpha_from_single_combination. The estimates of all combinations are combined with covariant weighting using the analytical covariance matrix:

V_β = T · Σ · Tᵀ

where T is the Jacobian of (βi, ⍺i) with respect to all error sources, and Σ is the diagonal matrix of errors sources variances (phase noise + systematic lattice errors). See _covariant_weighting for the combination step.

The best-knowledge model is used for the beta/alpha/phase-advance values fed into the error propagation (Jacobian), while the normal model provides the output reference columns (“MDL” suffix). If no best-knowledge model is available, both fall back to the same model.

Parameters:

• meas_input -- OpticsInput object with optics CLI options.

• phase -- PhaseDict of phase matrices of measurement with errors and model tfs (bpm x bpm).

• plane -- marking the horizontal or vertical plane, X or Y.

• meas_and_mdl_tunes -- measured and model tunes.

Returns:

TfsDataFrame containing betas and alfas from phase, as well as their errors.

omc3.optics_measurements.beta_from_phase.three_bpm_method(meas_input: DotDict, phase: PhaseDict, plane: str, meas_and_mdl_tunes: tuple[float, float]) → pd.DataFrame

Calculates betas and alphas using three adjacent-BPM triplet combinations (Castro, Ph.D. Thesis, University of Valencia, 1996). Unlike in the n_bpm_method this uses only the immediately neighbouring BPMs and propagates phase measurement uncertainties only (no systematic uncertainties for lattice errors).

For each probed BPM i, three triplets are formed and their β/β_mdl estimates are averaged arithmetically (see bet_frac below): - xxxABBx → probed BPM followed by two forward neighbours - xBBAxxx → two backward neighbours followed by probed BPM - xxBABxx → probed BPM flanked by skip-one neighbours

The input phase advance data from phase["MEAS"], phase["MODEL"], phase["ERRMEAS"] (from get_phases) are of the form:

	BPM1	BPM2	BPM3	BPM4
BPM1	0	phi_21	phi_31	phi_41
BPM2	phi_12	0	phi_32	phi_42
BPM3	phi_13	phi_23	0	phi_43

and tilt_slice_matrix(matrix, shift, slice, tune) brings it into the form:

	BPM1	BPM2	BPM3	BPM4
BPM_(i-1)	phi_1n	phi_21	phi_32	phi_43
BPM_i	0	0	0	0
BPM_(i+1)	phi_12	phi_23	phi_34	phi_45

The computed cot_phase_*_shift1 gives:

cot(phi_1n) - cot(phi_1n-1)	cot(phi_21) - cot(phi_2n)	cot(phi_32) - cot(phi_31)
NaN	NaN	NaN
NaN	NaN	NaN
cot(phi_13) - cot(phi_12)	cot(phi_24) - cot(phi_23)	cot(phi_35) - cot(phi_34)

-> for the combination xxxABBx: first row, -> for the combination xBBAxxx: fourth row, -> for the combination xxBABxx: second row of cot_phase_*_shift2.

Parameters:

• meas_input -- OpticsInput object with optics CLI options.

• phase -- PhaseDict of phase matrices of measurement with errors and model tfs (bpm x bpm).

• plane -- marking the horizontal or vertical plane, X or Y.

• meas_and_mdl_tunes -- measured and model tunes.

Returns:

TfsDataFrame containing betas and alfas from phase.

omc3.optics_measurements.beta_from_phase.write(beta_df: DataFrame, header: dict[str, Any], outputdir: str | Path, plane: str) → None

Chromatic

This module contains chromatic calculations functionality of optics_measurements. It provides functions to compute various chromatic beam properties.

omc3.optics_measurements.chromatic.calculate_chromatic_coupling(couplings: Sequence[pd.DataFrame], dpps: Sequence[float], input_files: InputFiles, measure_input: DotDict) → pd.DataFrame

Calculates chromatic coupling from coupling data at different dp/p values.

Parameters:

• couplings -- Sequence of coupling DataFrames, one per dp/p value.

• dpps -- Sequence of corresponding dp/p values.

• input_files -- InputFiles object containing measurement data.

• measure_input -- OpticsInput object containing analysis settings.

Returns:

A DataFrame with chromatic coupling amplitudes and components.

omc3.optics_measurements.chromatic.calculate_w_and_phi(betas: Sequence[pd.DataFrame], dpps: Sequence[float], input_files: InputFiles, measure_input: DotDict, plane: str) → pd.DataFrame

Calculates chromatic amplitude function W and its phase Phi.

Parameters:

• betas -- Sequence of beta DataFrames, one per dp/p value.

• dpps -- Sequence of corresponding dp/p values.

• input_files -- InputFiles object containing measurement data.

• measure_input -- OpticsInput object containing analysis settings.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A DataFrame with measured and model W and Phi columns.

Constants

Specific constants to be used in optics_measurements, to help with consistency.

Combined RDTs

This module contains combined resonance driving terms calculations functionality of optics_measurements. It provides functions to compute combined resonance driving terms following the derivations in https://arxiv.org/pdf/1402.1461.pdf.

omc3.optics_measurements.crdt.add_line_and_freq_to_header(header: dict[str, Any], crdt: dict) → dict[str, Any]

Adds spectral line and frequency information to the output header.

omc3.optics_measurements.crdt.calculate(measure_input: DotDict, input_files: InputFiles, invariants: dict[str, tfs.TfsDataFrame], header: dict[str, Any]) → None

Calculates Combined Resonance Driving Terms and writes results to file.

Iterates over all defined CRDTs (see module-level CRDTS list), computing amplitudes via ODR fits to the invariants and phases via circular averaging of the spectral lines. Results are written into per-order subfolders of the CRDT output directory.

The derivation follows https://arxiv.org/pdf/1402.1461.pdf.

Parameters:

• measure_input -- OpticsInput object containing analysis settings.

• input_files -- InputFiles object containing frequency spectra files (linx/y).

• invariants -- dictionary mapping planes to DataFrames of actions and errors per kick, e.g. from omc3.optics_measurements.kick.calculate().

• header -- headers to include in the written result files.

omc3.optics_measurements.crdt.fit_amplitude(lineamplitudes: ndarray, err_lineamplitudes: ndarray, crdt_invariant: ndarray, err_crdt_invariant: ndarray) → tuple[float, float]

Was translated from using scipy.odr to using odrpack as they recommended. See https://docs.scipy.org/doc/scipy/reference/odr.html for explanations.

omc3.optics_measurements.crdt.generic_dataframe(input_files: InputFiles, measure_input: DotDict, bpm_names: Sequence[str], dpp_value: float = 0) → pd.DataFrame

Generate a dataframe based on the MU-MDL columns from each measurement.

omc3.optics_measurements.crdt.get_column_names(line: str) → dict[str, str]

Returns the mapping of CRDT column identifiers to their suffixed column names.

omc3.optics_measurements.crdt.get_crdt_amplitude(crdt: dict, invariants: dict[str, ndarray], line_amps: ndarray, line_amp_errors: ndarray) → tuple[ndarray, ndarray]

Fits CRDT amplitudes per BPM from spectral line amplitudes and invariants.

omc3.optics_measurements.crdt.get_crdt_invariant(crdt: dict, invariants: dict[str, ndarray]) → tuple[ndarray, ndarray]

Computes the combined invariant and its error for the given CRDT from per-plane invariants.

omc3.optics_measurements.crdt.write_to_crdt_folder(df: pd.DataFrame, header: dict[str, Any], meas_input: DotDict, order: str, crdt: str) → None

Data Models

Models used in optics measurements to store and pass around data.

class omc3.optics_measurements.data_models.InputFiles(files_to_analyse: Sequence[str | Path | tfs.TfsDataFrame], optics_opt: DotDict)

Bases: dict

Stores the input files, provides methods to gather quantity specific data

Public methods:

• get_dpps (plane)

• get_joined_frame (plane, columns, zero_dpp=False, how=’inner’)

• get_columns (frame, column)

• get_data (frame, column)

bpms(plane: str | None = None, dpp_value: float | None = None) → Index

Returns the intersection of BPM indices across all input files for the given plane and dpp.

Parameters:

• plane -- X or Y. If None, returns intersection of both planes.

• dpp_value -- filter for given dp/p value. If None, uses all frames.

Returns:

A pd.Index of common BPM names.

calibrate(calibs: dict[str, DataFrame])

Use calibration data to rescale amplitude and amplitude error (if present).

Parameters:

calibs (dict) -- Plane-Dictionary with DataFrames of calibration data.

dpp_frames(plane: str, dpp_value: float | None) → list[TfsDataFrame]

Returns the list of DataFrames matching the given dpp_value for the given plane.

Parameters:

• plane -- marking the horizontal or vertical plane, X or Y.

• dpp_value -- the dp/p value to filter for. If None, all frames are returned.

Returns:

A list of TfsDataFrame objects.

dpp_frames_indices(plane: str, dpp_value: float | None) → ndarray | list[int]

Return the indices of the frames that match the dpp.

dpps(plane: str) → ndarray

Gathers measured DPPs from input files corresponding to given plane

Parameters:

plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A np.ndarray of DPPs.

static get_columns(frame: DataFrame, column: str) → list[str]

Returns list of columns of frame corresponding to column in original files.

Parameters:

• frame -- joined frame.

• column -- name of column in original files.

Returns:

list of columns.

static get_data(frame: DataFrame, column: str) → ndarray

Returns data in columns of frame corresponding to column in original files.

Parameters:

• frame -- joined frame.

• column -- name of column in original files.

Returns:

A np.ndarray corresponding to column in original files.

joined_frame(plane: str, columns: Sequence[str], dpp_value: float | None = None, dpp_amp: bool = False, how: str = 'inner', dtype: np.dtype = <class 'numpy.float64'>) → pd.DataFrame

Constructs merged DataFrame from collected DataFrames in InputFiles, i.e. from the harpy output of the given measurements.

The input parameters to this function determine which data will be present in the joined frame: which plane to use, the columns to be included, a dpp_value filter (with tolerance from omc3.optics_measurements.dpp.DPP_TOLERANCE) and how to perform the merge.

You can also specify what dtype the resulting data will have, this should normally be float64.

The columns in the resulting Dataframe will be suffixed by __#, starting from 0 with increasing integers for each of the input files.

Parameters:

• plane -- marking the horizontal or vertical plane, X or Y.

• columns -- list of columns from input files.

• dpp_value -- merges only files with given dpp_value.

• dpp_amp -- merges only files with non-zero dpp amplitude (i.e. 3Dkicks).

• how -- whi way to use for merging: inner (intersection) or outer (union), default is inner.

• dtype -- dtype of the merged DataFrame. Usually np.float64 should be used, but in case you have string- or complex- data you will need to set this accordingly or to None to avoid conversion.

Returns:

A merged TfsDataFrame from InputFiles.

omc3.optics_measurements.data_models.check_and_warn_about_offmomentum_data(input_files: InputFiles, plane: str, id_: str | None = None)

A helper function to check if off-momentum data is present in the input files, but no dpp-value is given by the user.

See https://github.com/pylhc/omc3/issues/456 .

omc3.optics_measurements.data_models.filter_for_dpp(to_filter: dict[str, Sequence], input_files: InputFiles, dpp_value: float) → dict

Filter the given data for the given dpp-value.

Dispersion

This module contains dispersion calculations related functionality of optics_measurements. It provides functions to compute orbit, dispersion and normalised dispersion.

omc3.optics_measurements.dispersion.calculate_dispersion(meas_input: DotDict, input_files: InputFiles, header_dict: dict, plane: str) → pd.DataFrame | None

Calculates dispersion data from measurements.

Parameters:

• meas_input -- OpticsInput object.

• input_files -- Stores the input files tfs.

• header_dict -- dict containing information about the analysis.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

TfsDataFrame corresponding to output file, or None if only a single dp/p bin is present.

omc3.optics_measurements.dispersion.calculate_normalised_dispersion(meas_input: DotDict, input_files: InputFiles, beta: pd.DataFrame, header_dict: dict) → pd.DataFrame | None

Calculates normalised dispersion data from measurements.

Parameters:

• meas_input -- OpticsInput object.

• input_files -- Stores the input files tfs.

• beta -- measured betas to get dispersion from normalised dispersion.

• header_dict -- dict containing information about the analysis.

Returns:

TfsDataFrame corresponding to output file, or None if only a single dp/p bin is present.

omc3.optics_measurements.dispersion.calculate_orbit(meas_input: DotDict, input_files: InputFiles, header: dict, plane: str) → tfs.TfsDataFrame

Calculates orbit data from measurements.

Parameters:

• meas_input -- OpticsInput object

• input_files -- Stores the input files tfs.

• header -- dict containing information about the analysis.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

TfsDataFrame corresponding to output file.

Dpp

This module contains deltap over p calculations related functionality of optics_measurements. It provides functions to computes and arrange dp over p.

omc3.optics_measurements.dpp.append_amp_dpp(list_of_tfs: Sequence[tfs.TfsDataFrame], dpp_values: Sequence[float]) → Sequence[tfs.TfsDataFrame]

Add the dpp values to the DPP-header of the tfs files, if larger than the DPP-tolerance, otherwise set to zero. This is intended to set the DPP value for on-momentum files to zero.

omc3.optics_measurements.dpp.append_dpp(list_of_tfs: Sequence[tfs.TfsDataFrame], dpp_values: Sequence[float]) → Sequence[tfs.TfsDataFrame]

Add the dpp values to the DPP-header of the tfs dataframes.

omc3.optics_measurements.dpp.arrange_dpps(dpps: Sequence[float], tolerance: float = 0.0001) → np.ndarray

Grouping of dpp-values and averaging them in the bins, also zeroes the bin closest to zero.

Parameters:

• dpps -- Sequence of dp/p values, one per analysed file.

• tolerance -- Maximum difference between dp/p values to be grouped into the same bin.

Returns:

Array of arranged dp/p values, same length as dpps, where each entry is the bin-averaged value with the zero-bin offset subtracted.

omc3.optics_measurements.dpp.calculate_amp_dpoverp(input_files: InputFiles, meas_input: DotDict) → float | np.ndarray

Calculates dp/p amplitude from synchrotron sideband amplitudes and model dispersion.

Parameters:

• input_files -- InputFiles object containing measurement data.

• meas_input -- OpticsInput object containing analysis settings.

Returns:

Scalar dp/p amplitude for a single file, or an array of values per file.

omc3.optics_measurements.dpp.calculate_dpoverp(input_files: InputFiles, meas_input: DotDict) → float | np.ndarray

Calculates dp/p from closed orbit and model dispersion at arc BPMs.

Parameters:

• input_files -- InputFiles object containing measurement data.

• meas_input -- OpticsInput object containing analysis settings.

Returns:

Scalar dp/p for a single file, or an array of dp/p values per file.

Isolation Forest

This module contains the isolation forest functionality of optics_measurements. It provides functions to detect and exclude BPMs with anomalies.

TODO: After discussion with lmalina: he thinks that we should not run Isolation Forest in its current form, as the binning of the frequencies in Harpy automatically leads to some clustering, which might then trigger wrong tune-filtering in here. This should be tested and possibly mitigated. (jdilly, 2024)

omc3.optics_measurements.iforest.clean_with_isolation_forest(input_files: Sequence[tfs.TfsDataFrame], meas_input: DotDict, plane: str) → Sequence[tfs.TfsDataFrame]

Runs isolation forest anomaly detection and removes flagged BPMs from input files. For every file where a cleaning is performed, a TfsDataFrame with information on the identified and cleaned BPMs is written to disk in an adjacent location in the corresponding output dir.

Parameters:

• input_files -- list of measurement DataFrames.

• meas_input -- OpticsInput object containing analysis settings.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

The input files with anomalous BPMs removed.

omc3.optics_measurements.iforest.detect_anomalies(contamination: float, data: DataFrame, plane: str) → tuple[DataFrame, DataFrame, ndarray]

Fits an isolation forest and separates data into anomalous and normal BPMs.

Parameters:

• contamination -- expected proportion of outliers in the data.

• data -- normalised BPM feature data.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A tuple of (bad_bpms, good_bpms, bad_bpms_scores).

omc3.optics_measurements.iforest.get_data_for_clustering(bpm_tfs_data: pd.DataFrame, plane: str, accelerator: Accelerator) → tuple[pd.DataFrame, pd.DataFrame]

Splits BPM data into arc and IR subsets and normalises features for clustering.

Parameters:

• bpm_tfs_data -- concatenated BPM data from all input files.

• plane -- marking the horizontal or vertical plane, X or Y.

• accelerator -- accelerator instance providing the element types mask.

Returns:

A tuple of (arc_bpm_data, ir_bpm_data) with normalised features.

omc3.optics_measurements.iforest.get_significant_features(bpm_tfs_data: DataFrame, data_for_clustering: DataFrame, bad_bpms: DataFrame, good_bpms: DataFrame, plane: str) → DataFrame

Determines the most significant feature for each anomalous BPM.

For each bad BPM, finds the feature (tune, noise, or amplitude) with the largest deviation from the mean of good BPMs.

Parameters:

• bpm_tfs_data -- concatenated BPM data from all input files.

• data_for_clustering -- normalised feature data used for clustering.

• bad_bpms -- DataFrame of BPMs flagged as anomalous.

• good_bpms -- DataFrame of BPMs considered normal.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A DataFrame indexed like bad_bpms with columns NAME, FEATURE, VALUE, and AVG.

omc3.optics_measurements.iforest.identify_bad_bpms(meas_input: DotDict, input_files: Sequence[tfs.TfsDataFrame], plane: str) → pd.DataFrame

Identifies anomalous BPMs across arc and IR regions using isolation forest.

Parameters:

• meas_input -- OpticsInput object containing analysis settings.

• input_files -- list of measurement DataFrames.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A DataFrame listing the detected bad BPMs with their significant features and scores.

omc3.optics_measurements.iforest.identify_single_cluster_bad_bpms(bpm_tfs_data: DataFrame, cont: float, data_for_clustering: DataFrame, plane: str) → DataFrame

Runs isolation forest on a single cluster and returns the significant features of detected anomalies.

Parameters:

• bpm_tfs_data -- concatenated BPM data from all input files.

• cont -- contamination parameter for the isolation forest.

• data_for_clustering -- normalised feature data for clustering.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A DataFrame with anomalous BPM names, their most significant feature, and anomaly scores.

omc3.optics_measurements.iforest.reassign_index(data: DataFrame) → DataFrame

Resets the index of the DataFrame to a sequential integer range.

omc3.optics_measurements.iforest.remove_bad_bpms(tfs_dfs: Sequence[tfs.TfsDataFrame], bad_bpm_names: list[str], plane: str) → list[tfs.TfsDataFrame]

Removes flagged BPMs from all input file DataFrames and recalculates tune statistics.

Parameters:

• tfs_dfs -- list of measurement DataFrames.

• bad_bpm_names -- names of BPMs to remove.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A list of the input TfsDataFrames with bad BPMs filtered out.

Interaction Point

This module contains IP properties calculations related functionality of optics_measurements. It provides functions to compute beta* from phase.

Todo

• Put columns into the constants

omc3.optics_measurements.interaction_point.betastar_from_phase(meas_input: DotDict, phase_d: phase.PhaseDict) → pd.DataFrame | None

Calculate beta* and l* from the phase advance of the IP-BPMs.

Parameters:

• meas_input -- Measurement_input object.

• phase_d -- PhaseDict output of the phasecalculation.

Returns:

A DataFrame with the beta* and l* as well as the phases used for the calculation as columns and the IP names as index. Can return None if the provided accelerator class has no .get_ips method.

omc3.optics_measurements.interaction_point.phase_to_betastar(lstar: float, phase: float, errphase: float) → tuple[float, float]

Return the betastar and its error given the phase advance across the IP.

This function computes the betastar using the phase advance between the BPMs around the IP and their distance (lstar). The phase and error in the phase must be given in radians.

Parameters:

• lstar (float) -- The distance between the BPMs and the IR.

• phase (float) -- The phase advance between the BPMs at each side of the IP. Must be given in radians.

• errphase (float) -- The error in phase. Must be given in radians.

Returns:

betastar and its error.

Return type:

tuple[float, float]

omc3.optics_measurements.interaction_point.write(df_ips: DataFrame, headers: dict[str, Any], output_dir: str | Path, plane: str) → None

Write the interaction point data to disk. Empty DataFrames are skipped on write.

Parameters:

• df_ips (pd.DataFrame) -- The interaction point data.

• headers (dict[str, Any]) -- The headers for the tfs file.

• output_dir (str|Path) -- The path to the output directory.

• plane (str) -- The plane of the interaction point data.

Kick

This module contains kick functionality of optics_measurements. It provides functions to compute kick actions.

omc3.optics_measurements.kick.calculate(measure_input: DotDict, input_files: InputFiles, scale: float, header_dict: dict, plane: str) → np.ndarray

Parameters:

• measure_input -- OpticsInput object.

• input_files -- Stores the input files tfs.

• scale -- measured beta functions.

• header_dict -- dict containing information about the analysis.

• plane -- marking the horizontal or vertical plane, X or Y.

Returns:

A 2D numpy array containing the sqrt(2J) and associated errors for the given plane.

Phase Advance

This module contains phase calculation functionality of optics_measurements. It provides functions to compute betatron phase advances and structures to store them.

class omc3.optics_measurements.phase.CompensationMode

Bases: object

Available modes for AC-dipole driven motion compensation in phase advance calculations.

EQUATION: str = 'equation'

MODEL: str = 'model'

NONE: str = 'none'

classmethod all() → list[str]

class omc3.optics_measurements.phase.PhaseDict

Bases: TypedDict

ERRMEAS: DataFrame

MEAS: DataFrame

MODEL: DataFrame

omc3.optics_measurements.phase.calculate(meas_input: DotDict, input_files: InputFiles, tunes: TuneDict, plane: str, no_errors: bool = False) → tuple[dict[str, PhaseDict], list[pd.DataFrame]]

Calculate phases for ‘compensated’ (aka ‘free’) and ‘uncompensated’ (aka ‘driven’) cases from the measurement files, and return a dictionary combining the results for each transverse plane.

Parameters:

• meas_input (DotDict) -- OpticsInput object containing analysis settings from the command-line.

• input_files (InputFiles) -- InputFiles object containing frequency spectra files (linx/y).

• tunes (TuneDict) -- TuneDict contains measured tunes.

• plane (str) -- marking the horizontal or vertical plane, X or Y.

• no_errors (bool) -- if True, measured errors shall not be propagated (only their spread).

Returns:

A tuple of a dictionary and a list of corresponding phase DataFrames. The dictionary contains the compensated and uncompensated results in PhaseDict form, i.e. a dictionary of DataFrames with the phase models, measured and errors. The list contains the DataFrames with the phase advances between BPMs and the total phase advances, for both compensated and uncompensated cases.

omc3.optics_measurements.phase.write(dfs: Sequence[pd.DataFrame], headers: Sequence[dict[str, Any]] | dict[str, Any], output: Path | str, plane: str)

Write out the phase advance data into TFS-files.

omc3.optics_measurements.phase.write_special(meas_input: DotDict, phase_advances: pd.DataFrame, plane_tune: float, plane: str)

Writes out the special phase advances, if any given by the accelerator class.

Resonance Driving Terms

This module contains RDT calculations related functionality of optics_measurements. It provides functions to compute global resonance driving terms f_jklm.

omc3.optics_measurements.rdt.add_freq_to_header(header: dict[str, Any], plane: str, rdt: tuple[int, int, int, int]) → dict[str, Any]

omc3.optics_measurements.rdt.calculate(measure_input: DotDict, input_files: InputFiles, tunes: TuneDict, phases: dict[str, dict[str, PhaseDict]], invariants: dict[str, pd.DataFrame], header: dict) → None

Computes the RDTs for the given input files and settings up to the magnet order given in the inputs, and writes the results to file.

Parameters:

• measure_input -- OpticsInput object containing analysis settings.

• input_files -- InputFiles object containing frequency spectra files (linx/y).

• tunes -- TuneDict object mapping planes to tunes, as given by omc3.optics_measurements.tune.calculate().

• phases -- dictionary mapping planes to PhaseDict for compensated and uncompensated cases.

• invariants -- dictionary mapping planes to dataframes of actions/errors per kick, e.g. from omc3.optics_measurements.kick.calculate().

• header -- headers to include to the written result files.

omc3.optics_measurements.rdt.complex_secondary_lines(phase_adv: ArrayLike, err_padv: ArrayLike, sig1: ArrayLike, sig2: ArrayLike) → tuple[ndarray, ndarray, ndarray, ndarray]

Extracts amplitude and phase of secondary spectral lines from BPM pairs.

Parameters:

• phase_adv -- phase advances between two BPMs.

• err_padv -- error on the phase advance between two BPMs.

• sig1 -- Complex coefficients of a secondary lines at the first BPM of the pairs.

• sig2 -- Complex coefficients of a secondary lines at the second BPM of the pairs.

Returns:

Tuple with amplitudes, phases, err_amplitudes and err_phases of the complex signal.

omc3.optics_measurements.rdt.get_line_sign_and_suffix(line: tuple[int, int, int], input_files: InputFiles, plane: str) → tuple[int, str]

omc3.optics_measurements.rdt.get_linearized_problem(invs: dict[str, ndarray], plane: str, rdt: tuple[int, int, int, int]) → ndarray

2 * j * f_jklm * (powers of 2Jx and 2Jy) : f_jklm is later a parameter of a fit we use sqrt(2J): unit is sqrt(m).

omc3.optics_measurements.rdt.to_complex(amplitudes: ArrayLike, phases: ArrayLike, period: float = 1) → ndarray

omc3.optics_measurements.rdt.write(df: pd.DataFrame, header: dict[str, Any], meas_input: DotDict, plane: str, rdt: RDTTuple) → None

Toolbox

This module contains helper functionality for optics_measurements. It provides functions to perform regularly used simple calculations.

omc3.optics_measurements.toolbox.ang_diff(a: ndarray, b: ndarray) → ndarray

Returns the angular difference between angles a and b in [-0.5, 0.5].

omc3.optics_measurements.toolbox.ang_interval_check(ang: ndarray) → ndarray

Returns ang wrapped into [-0.5, 0.5]

omc3.optics_measurements.toolbox.ang_sum(a: ndarray, b: ndarray) → ndarray

Returns a column containing the angular sum between angles a and b in [-0.5 , 0.5]

omc3.optics_measurements.toolbox.df_ang_diff(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the angular difference between angles a and b in [-0.5 , 0.5]

omc3.optics_measurements.toolbox.df_diff(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the difference between a_col and b_col

omc3.optics_measurements.toolbox.df_diff_with_err(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → tuple[np.ndarray, np.ndarray]

Returns two columns containing the difference and the total errors of the given columns.

omc3.optics_measurements.toolbox.df_err_sum(df: pd.DataFrame, a_err_col: str, b_err_col: str) → np.ndarray

Returns a column containing the root of the sum-of-squares of a_err_col and b_err_col

omc3.optics_measurements.toolbox.df_prod(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the product of a_col and b_col

omc3.optics_measurements.toolbox.df_prod_with_err(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → tuple[np.ndarray, np.ndarray]

Returns two columns containing the product and the total errors of the given columns.

omc3.optics_measurements.toolbox.df_ratio(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the ratio between a_col and b_col

omc3.optics_measurements.toolbox.df_ratio_with_err(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → tuple[np.ndarray, np.ndarray]

Returns two columns containing the ratio and the total errors of the given columns.

omc3.optics_measurements.toolbox.df_rel_diff(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the difference between a_col and b_col relative to b_col

omc3.optics_measurements.toolbox.df_rel_diff_with_err(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → tuple[np.ndarray, np.ndarray]

Returns two columns containing the relative difference and the total errors of the given columns.

omc3.optics_measurements.toolbox.df_rel_err_sum(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → np.ndarray

Returns a column containing the root of the relative sum-of-square of a_col/a_err_col and b_col/b_err_col

omc3.optics_measurements.toolbox.df_sum(df: pd.DataFrame, a_col: str, b_col: str) → np.ndarray

Returns a column containing the sum of a_col and b_col

omc3.optics_measurements.toolbox.df_sum_with_err(df: pd.DataFrame, a_col: str, b_col: str, a_err_col: str, b_err_col: str) → tuple[np.ndarray, np.ndarray]

Returns two columns containing the sum and the total errors of the given columns.

Tune

This module contains tune calculations functionality of optics_measurements. It provides functions to compute betatron tunes and structures to store them.

class omc3.optics_measurements.tune.TuneDict

Bases: dict

Data structure to hold tunes.

get_lambda(plane: str) → float

Computes lambda compensation factor.

Parameters:

plane -- marking the horizontal or vertical plane, X or Y.

Returns:

lambda compensation factor (driven vs free motion).

phase_ac2bpm(df_idx_by_bpms: pd.DataFrame, plane: str, accelerator) → tuple[str, float, int, str]

Returns the necessary values for the exciter compensation. See DOI: 10.1103/PhysRevSTAB.11.084002

Parameters:

• df_idx_by_bpms (pandas.DataFrame) -- commonbpms (see GetLLM._get_commonbpms)

• plane (str) -- marking the horizontal or vertical plane, X or Y.

• accelerator -- an Accelerator object.

Returns:

A Tuple consisting of four elements a, b, c, d.

• a (string): name of the nearest BPM.

• b (float): compensated phase advance between the exciter and the nearest BPM.

• c (int): k of the nearest BPM.

• d (string): name of the exciter element.

omc3.optics_measurements.tune.calculate(measure_input: DotDict, input_files: InputFiles) → TuneDict

Computes measured, model and free tunes from input files.

Averages per-file tune values weighted by their RMS, and determines the free tune when AC-dipole excitation compensation is active.

Parameters:

• measure_input -- OpticsInput object containing analysis settings.

• input_files -- InputFiles object containing frequency spectra files.

Returns:

A TuneDict populated with model, measured and free tunes per plane.