Machine Data Extraction

Constants

Specific constants relating to the retrieval of machine settings information, to help with consistency.

class omc3.machine_data_extraction.constants.MSISummaryColumn(*values)[source]

Bases: StrEnum

TFS columns for machine settings information.

KNOB = 'KNOB'
TIME = 'TIME'
TIMESTAMP = 'TIMESTAMP'
VALUE = 'VALUE'
class omc3.machine_data_extraction.constants.MSISummaryHeader(*values)[source]

Bases: StrEnum

TFS headers for machine settings information.

ACCEL = 'ACCELERATOR'
BEAMPROCESS = 'BEAMPROCESS'
BEAMPROCESS_DESCRIPTION = 'BEAMPROCESS_DESCRIPTION'
BEAMPROCESS_START = 'BEAMPROCESS_START'
CONTEXT_CATEGORY = 'CONTEXT_CATEGORY'
END_TIME = 'END_TIME'
FILL = 'FILL'
OPTICS = 'OPTICS'
OPTICS_START = 'OPTICS_START'
START_TIME = 'START_TIME'
TIME = 'TIME'
class omc3.machine_data_extraction.constants.NXCalsTfsColumn(*values)[source]

Bases: StrEnum

TFS columns for MQT information.

MADX = 'MADX'
PC_NAME = 'PC_NAME'
TIME = 'TIME'
TIMESTAMP = 'TIMESTAMP'
VALUE = 'VALUE'
class omc3.machine_data_extraction.constants.NXCalsTfsHeader(*values)[source]

Bases: StrEnum

TFS headers for MQT information.

BEAM = 'BEAM'
EXTRACTION_TIME = 'EXTRACTION_TIME'

DataClasses for Machine Data

This module contains the dataclasses that are used to hold and pass on the extracted machine data.

Most of them are purely to hold the data in a structured way, some have minor functionality to convert the data into different formats such as TFS and MADX.

class omc3.machine_data_extraction.data_classes.BeamProcessInfo(name: str, accelerator: str, context_category: str, start_time: datetime, category: str, description: str)[source]

Bases: object

Dataclass to hold BeamProcess information.

This contains only the relevant fields for OMC, extracted from the Java BeamProcess object. Add more fields if needed.

accelerator: str
category: str
context_category: str
description: str
classmethod from_java_beamprocess(bp: StandAloneBeamProcessImpl) → BeamProcessInfo[source]

Create a BeamProcessInfo from a StandAloneBeamProcessImpl object.

Parameters:

bp (StandAloneBeamProcessImpl) -- The BeamProcess object (Java).

Returns:

The corresponding BeamProcessInfo dataclass instance.

Return type:

BeamProcessInfo

name: str
start_time: datetime
class omc3.machine_data_extraction.data_classes.DefinitionColumn(*values)[source]

Bases: StrEnum

Columns for Knob Definition TFS files.

CIRCUIT = 'CIRCUIT'
FACTOR = 'FACTOR'
MADX_NAME = 'NAME'
TYPE = 'TYPE'
class omc3.machine_data_extraction.data_classes.DefinitionHeader(*values)[source]

Bases: StrEnum

Headers for Knob Definition TFS files.

NAME = 'KNOB_NAME'
OPTICS = 'OPTICS'
OUTPUT_NAME = 'OUTPUT_NAME'
class omc3.machine_data_extraction.data_classes.FillInfo(no: int, accelerator: str, start_time: datetime, beamprocesses: list[tuple[datetime, str]] | None = None)[source]

Bases: object

Dataclass to hold Fill information.

This contains only the relevant fields for OMC, extracted from the Java Fill object. Add more fields if needed.

accelerator: str
beamprocesses: list[tuple[datetime, str]] | None = None
no: int
start_time: datetime
class omc3.machine_data_extraction.data_classes.KnobDefinition(name: str, optics: str, parts: list[KnobPart] = <factory>)[source]

Bases: object

Dataclass to hold Knob Definition information.

name: str
optics: str
property output_name: str

Generates a safe knob name for output files.

parts: list[KnobPart]
to_madx(value: float = 0.0, template: str = 'add2expr') → str[source]

Converts the knob definition to madx code.

to_tfs() → TfsDataFrame[source]

Convert KnobDefinition to TfsDataFrame.

class omc3.machine_data_extraction.data_classes.KnobPart(circuit: str, type: str, factor: float, madx_name: str | None)[source]

Bases: object

Dataclass to hold Knob Part information.

circuit: str
factor: float
madx_name: str | None
type: str
class omc3.machine_data_extraction.data_classes.MachineSettingsInfo(time: datetime, accelerator: str, fill: FillInfo | None = None, beamprocess: BeamProcessInfo | None = None, optics: OpticsInfo | None = None, trim_histories: TrimHistories | None = None, trims: dict[str, float] | None = None, knob_definitions: dict[str, KnobDefinition] | None = None)[source]

Bases: object

Dataclass to hold the extracted Machine Settings Info.

accelerator: str
beamprocess: BeamProcessInfo | None = None
fill: FillInfo | None = None
knob_definitions: dict[str, KnobDefinition] | None = None
optics: OpticsInfo | None = None
time: datetime
trim_histories: TrimHistories | None = None
trims: dict[str, float] | None = None
class omc3.machine_data_extraction.data_classes.OpticsInfo(name: str, id: str, start_time: datetime, accelerator: str | None = None, beamprocess: BeamProcessInfo | None = None)[source]

Bases: object

Dataclass to hold Optics information.

accelerator: str | None = None
beamprocess: BeamProcessInfo | None = None
id: str
name: str
start_time: datetime
class omc3.machine_data_extraction.data_classes.TrimHistories(beamprocess: str, start_time: datetime | None, end_time: datetime | None, accelerator: str, trims: dict[str, TrimTuple])[source]

Bases: object

Dataclass to hold Knob Trim History information.

accelerator: str
beamprocess: str
end_time: datetime | None
headers: dict[str, str | float | int]
start_time: datetime | None
to_tfs_dict() → dict[str, TfsDataFrame][source]

Convert all trims to a dictionary of TfsDataFrames.

trim_tuple_to_tfs(knob_name: str, trim_tuple: TrimTuple) → tfs.TfsDataFrame[source]

Convert KnobTrimHistory to TfsDataFrame.

trims: dict[str, TrimTuple]
class omc3.machine_data_extraction.data_classes.TrimHistoryColumn(*values)[source]

Bases: StrEnum

Columns for Trim History TFS files.

DATA = 'DATA'
TIME = 'TIME'
TIMESTAMP = 'TIMESTAMP'
class omc3.machine_data_extraction.data_classes.TrimHistoryHeader(*values)[source]

Bases: StrEnum

Headers for Trim History TFS files.

ACCEL = 'ACCELERATOR'
BEAMPROCESS = 'BEAMPROCESS'
END_TIME = 'END_TIME'
FILL = 'FILL'
KNOB_NAME = 'KNOB_NAME'
OPTICS = 'OPTICS'
START_TIME = 'START_TIME'

LSA Beamprocesses

This module contains functions to extract beamprocess - and related - information from LSA.

omc3.machine_data_extraction.lsa_beamprocesses.beamprocess_to_dict(bp: StandAloneBeamProcessImpl) → dict[source]

Convert all get- fields of the beamprocess (java) to a dictionary.

Will contain the original Beamprocess object under the key ‘object’ and its string representation under ‘name’.

Parameters:

bp (StandAloneBeamProcessImpl) -- Beamprocess object (Java).

Returns:

Dictionary representation of the beamprocess.

Return type:

dict

omc3.machine_data_extraction.lsa_beamprocesses.get_active_beamprocess_at_time(lsa_client: LSAClient, time: datetime, accelerator: str = 'lhc', bp_group: str = 'POWERCONVERTERS') → BeamProcessInfo[source]

Find the active beam process at the time given.

Note: This function has been derived from the original KnobExtractor implementation in the Java Online Model.

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance.

  • time (datetime) -- The time at which to find the active beam process.

  • accelerator (str) -- Name of the accelerator.

  • bp_group (str) -- BeamProcess Group, choices : ‘POWERCONVERTERS’, ‘ADT’, ‘KICKERS’, ‘SPOOLS’, ‘COLLIMATORS’

Returns:

The corresponding BeamProcessInfo dataclass instance.

Return type:

BeamProcessInfo

omc3.machine_data_extraction.lsa_beamprocesses.get_beamprocess_from_name(lsa_client: LSAClient, name: str) → BeamProcessInfo[source]

Get the BeamProcess by its name.

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance.

  • name (str) -- The name of the beam process.

Returns:

The corresponding BeamProcessInfo dataclass instance.

Return type:

BeamProcessInfo

omc3.machine_data_extraction.lsa_beamprocesses.get_beamprocess_with_fill_at_time(lsa_client: LSAClient, spark: SparkSession, time: datetime, data_retrieval_days: float = 1, accelerator: str = 'lhc') → tuple[FillInfo, BeamProcessInfo][source]

Get the info about the active beamprocess at time.

omc3.machine_data_extraction.lsa_beamprocesses.get_fills_with_beamprocesses(lsa_client: LSAClient, spark: SparkSession, time: datetime, data_retrieval_days: float, accelerator: str = 'lhc') → list[FillInfo][source]

Retrieves fills with their associated beamprocesses between the specified time window. Adapted from pjlsa’s FindBeamProcessHistory.

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance used to query beamprocess history.

  • spark (SparkSession) -- Spark session used to query NXCALS fill-number data.

  • time (datetime) -- End time of the retrieval window.

  • data_retrieval_days (float) -- Number of days to look back from time.

  • accelerator (str) -- Name of the accelerator.

Returns:

List of FillInfo objects, each containing the fill number, accelerator, start time, and beam processes. They are sorted by fill number.

LSA Knobs

This module contains functions to extract knob information from LSA.

omc3.machine_data_extraction.lsa_knobs.filter_non_existing_knobs(lsa_client: LSAClient, knobs: list[str]) → list[str][source]

Return only the knobs that exist from the given list. This function was created out of the need to filter these first, as knobs that exist but do not belong to a beamprocess return noting in _getTrimsByBeamprocess, while knobs that do not exist at all crashed pjlsa. This filter should probably have been in pjlsa’s _buildParameterList.

Parameters:

knobs (list) -- List of strings of the knobs to check.

Returns:

A list of the knob names that actually exist.

omc3.machine_data_extraction.lsa_knobs.find_knob_names(lsa_client: LSAClient, accelerator: str = 'lhc', regexp: str | None = None) → list[source]

Return knobs for accelerator.

Parameters:

  • lsa_client -- The LSA client instance.

  • accelerator (str) -- Accelerator name.

  • regexp (str) -- Regular Expression to filter knob names.

Returns:

Sorted list of knob names.

omc3.machine_data_extraction.lsa_knobs.get_knob_definition(lsa_client: LSAClient, knob: str, optics: str) → KnobDefinition[source]

Get a dataframe of the structure of the knob. Similar to online model extractor (KnobExtractor.getKnobHiercarchy)

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance.

  • knob (str) -- The name of the knob.

  • optics (str) -- The optics name.

Returns:

The definition of the knob.

Return type:

KnobDefinition

omc3.machine_data_extraction.lsa_knobs.get_last_trim(trims: dict[str, TrimTuple]) → dict[str, float][source]

Returns the last trim in the trim history.

Parameters:

trims (dict) -- trim history as extracted via LSA.get_trim_history()

Returns:

Dictionary of knob names and their values.

omc3.machine_data_extraction.lsa_knobs.get_trim_history(lsa_client: LSAClient, beamprocess: str, knobs: list | None = None, start_time: datetime | None = None, end_time: datetime | None = None, accelerator: str = 'lhc') → TrimHistories[source]

Get trim history for knobs between specified times. If any of the times are not given, all available data in that time-direction is extracted for that beamprocess.

Parameters:

  • beamprocess (str) -- Name of the beamprocess.

  • knobs (list) -- List of strings of the knobs to check.

  • start_time (datetime) -- Earliest time to extract trims from.

  • end_time (datetime) -- Latest time to extract trims to.

  • accelerator (str) -- Name of the accelerator.

Returns:

TrimHistory object containing trim extraction and history information. The actual trims are in the ‘trims’ attribute as a dictionary of knob names and their data (as TrimTuples, i.e. NamedTuple of lists of time and data).

LSA Optics

This module contains functions to extract the used optics from LSA.

omc3.machine_data_extraction.lsa_optics.get_optics_for_beamprocess_at_time(lsa_client: LSAClient, time: datetime, beamprocess: BeamProcessInfo) → OpticsInfo[source]

Get the optics information for the given beamprocess at the specified time.

Note

This function has been derived from old implementations and returns the right optics, but in my tests the optics table only ever had one entry with time 0. (jdilly, 2026)

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance.

  • time (datetime) -- The time at which to get the optics.

  • beamprocess (BeamProcessInfo) -- The beamprocess information.

Returns:

The corresponding OpticsInfo dataclass instance.

Return type:

OpticsInfo

LSA Utilities

This module contains utility functions for working with LSA and data extracted from it.

omc3.machine_data_extraction.lsa_utils.calc_k_from_iref(lsa_client: LSAClient, currents: dict[str, float], energy: float) → dict[str, float][source]

Calculate K values in the LHC from IREF using the LSA service.

Parameters:

  • lsa_client (LSAClient) -- The LSA client instance.

  • currents (dict[str, float]) -- Dictionary of current values keyed by variable name.

  • energy (float) -- The beam energy in GeV.

Returns:

Dictionary of K values keyed by variable name.

Return type:

dict[str, float]

NXCals Results to MAD-X Conversion

This module provides functions to convert NXCALS knob extraction results into a MAD-X script format for further processing.

omc3.machine_data_extraction.madx_conversion.map_lsa_name_to_madx(lsa_client: LSAClient, name: str)[source]

Returns the MAD-X name (Circuit/Knob) from the given circuit name in LSA.

omc3.machine_data_extraction.madx_conversion.map_pc_name_to_madx(pc_name: str) → str[source]

Convert an LHC power converter name or circuit name to its corresponding MAD-X name.

This function processes the input name by removing the ‘:I_MEAS’ suffix if present, extracting the circuit name from full power converter names (starting with ‘RPMBB’ or ‘RPL’), applying specific string replacements for MAD-X compatibility, and converting the result to lowercase.

..note::

This function currently supports only some of the LHC power converters: Corrector bends (RCB), Sextupoles (RS), Quadrupoles (RQ), and Bends (RB). Additional logic may be required to handle other magnet types.

Parameters:

pc_name (str) -- The power converter or circuit name to convert.

Returns:

The converted MAD-X name.

Return type:

str

Examples

>>> map_pc_name_to_madx("RPMBB.UJ33.RCBH10.L1B1:I_MEAS")
'acb10.l1b1'
>>> map_pc_name_to_madx("RQT12.L5B1")
'kqt12.l5b1'

MQT knobs extraction

This module provides functions to retrieve MQT (Quadrupole Trim) knob values for the LHC for a specified beam and time using NXCALS and LSA.

There are two types per arc (focusing ‘f’ and defocusing ‘d’), resulting in 16 MQT circuits per beam (8 arcs * 2 planes).

The extraction uses the underlying knob_extraction module with MQT-specific patterns.

omc3.machine_data_extraction.mqt_extraction.generate_mqt_names(beam: int) → set[str][source]

Generate the set of MAD-X MQT (Quadrupole Trim) variable names for a given beam.

Parameters:

beam (int) -- The beam number (1 or 2).

Returns:

A set of MAD-X variable names for MQT magnets, e.g., ‘kqt12.a12b1’.

Return type:

set[str]

Raises:

ValueError -- If beam is not 1 or 2.

Examples

>>> get_mqts(1)
{'kqt12.a12b1', 'kqt12.a23b1', ..., 'kqtd.a81b1'}
omc3.machine_data_extraction.mqt_extraction.get_mqt_vals(spark: SparkSession, time: datetime, beam: int, data_retrieval_days: float = 0.25, energy: float | None = None) → list[NXCALSResult][source]

Retrieve MQT (Quadrupole Trim) knob values from NXCALS for a specific time and beam.

This function queries NXCALS for current measurements of MQT power converters, calculates the corresponding K-values (integrated quadrupole strengths) using LSA, and returns them in MAD-X format with timestamps.

Parameters:

  • spark (SparkSession) -- Active Spark session for NXCALS queries.

  • time (datetime) -- The timestamp for which to retrieve the data (timezone-aware required).

  • beam (int) -- The beam number (1 or 2).

  • data_retrieval_days (float) -- Number of days to look back for data in NXCALS. Will always take the latest available data within this window. default: 0.25 (e.g. 6 hours)

  • energy (float | None) -- Beam energy in GeV. If None, the energy is retrieved from the HX:ENG variable.

Returns:

List of NXCalResult objects containing the MAD-X knob names, K-values, and timestamps.

Return type:

list[NXCalResult]

Raises:

  • ValueError -- If beam is not 1 or 2 (propagated from get_mqts), or if time is not timezone-aware.

  • RuntimeError -- If no data is found in NXCALS or LSA calculations fail.

NXCals Knob Extraction

This module provides functionality to extract knob values from NXCALS for the LHC and convert them to MAD-X compatible format using LSA services.

It handles retrieval of raw variable data from NXCALS, conversion of power converter currents to K-values, and mapping of power converter names to MAD-X naming conventions.

This module requires the installation of jpype, pyspark, and access to the CERN network to connect to NXCALS and LSA services. You can install the required packages via pip:

` python -m pip install omc3[cern] `

See the [NXCALS documentation](https://nxcals-docs.web.cern.ch/current/user-guide/data-access/quickstart/) for more information on getting access and using the Python API.

class omc3.machine_data_extraction.nxcals_knobs.NXCALSResult(name: 'str', value: 'float', datetime: 'datetime', pc_name: 'str')[source]

Bases: object

datetime: datetime
name: str
pc_name: str
to_madx() → str[source]
to_series() → Series[source]
classmethod to_tfs(results: list[NXCALSResult], time: datetime, beam: int) → DataFrame[source]
value: float
omc3.machine_data_extraction.nxcals_knobs.get_energy(spark: SparkSession, time: datetime) → tuple[float, pd.Timestamp][source]

Retrieve the beam energy of the LHC from NXCALS.

Parameters:

  • spark (SparkSession) -- Active Spark session.

  • time (datetime) -- Python datetime (timezone-aware recommended).

Returns:

Beam energy in GeV and its times.

Return type:

tuple[float, pd.Timestamp]

Raises:

RuntimeError -- If no energy data is found.

omc3.machine_data_extraction.nxcals_knobs.get_knob_vals(spark: SparkSession, time: datetime, beam: int, patterns: list[str], expected_knobs: set[str] | None = None, log_prefix: str = '', data_retrieval_days: float = 0.25, energy: float | None = None) → list[NXCALSResult][source]

Retrieve knob values for a given beam and time using specified patterns for the LHC.

This is the main entry point for extracting magnet knob values from NXCALS. The function performs a complete workflow:

  1. Queries NXCALS for power converter current measurements (I_MEAS) using variable patterns

  2. Retrieves the beam energy at the specified time

  3. Converts currents to K-values (integrated quadrupole strengths) using LSA

  4. Maps power converter names to MAD-X naming conventions

  5. Returns knob values with their datetimnes

The difference between patterns and knob names:

  • patterns: NXCALS variable patterns (e.g., “RPMBB.UA%.RQT%.A%B1:I_MEAS”) used to query raw power converter current measurements. These follow CERN naming conventions and may include wildcards (%).

  • expected_knobs: MAD-X element names (e.g., “kqt12.a12b1”) representing the final knob names as used in MAD-X scripts. These are lowercase, simplified names.

Parameters:

  • spark (SparkSession) -- Active Spark session for NXCALS queries.

  • time (datetime) -- Time to retrieve data for.

  • beam (int) -- Beam number (1 or 2).

  • patterns (list[str]) -- List of NXCALS variable patterns to query for power converter currents. Patterns can include SQL-like wildcards (%). Example: “RPMBB.UA%.RQT%.A%B1:I_MEAS” matches all MQT quadrupole trim magnets for beam 1.

  • expected_knobs (set[str] | None) -- Set of expected MAD-X knob names to validate and filter results. If None, returns all found knobs without validation.

  • log_prefix (str) -- Prefix for logging messages to distinguish different extraction runs.

  • data_retrieval_days (float) -- Number of days to look back for data. Default is 0.25.

  • energy (float | None) -- Beam energy in GeV. If None, the energy is retrieved from the HX:ENG variable.

Returns:

List of NXCalResult objects containing MAD-X knob names, K-values,

times, and power converter names.

Return type:

list[NXCalResult]

omc3.machine_data_extraction.nxcals_knobs.get_raw_vars(spark: SparkSession, time: datetime, var_name: str, data_retrieval_days: float = 0.25, latest_only: bool = True) → list[NXCALSResult][source]

Retrieve raw variable values from NXCALS.

Parameters:

  • spark (SparkSession) -- Active Spark session.

  • time (datetime) -- Python datetime (timezone-aware recommended).

  • var_name (str) -- Name or pattern of the variable(s) to retrieve.

  • data_retrieval_days (float) -- Number of days to look back for data. Default is 0.25.

  • latest_only (bool) -- If True, only the latest sample for each variable is returned. Default is True.

Returns:

List of NXCalResult containing variable name, value, times, and power converter name for the latest sample of each matching variable at the given time, or all samples if so required.

Return type:

list[NXCalResult]

Raises:

  • RuntimeError -- If no data is found for the variable in the given interval.

  • You may need to increase the data_retrieval_days if necessary. --

Utils

Utilities to help with machine data extraction.

omc3.machine_data_extraction.utils.knob_to_output_name(knob_name: str) → str[source]

Convert a knob name to an output-friendly name for file names.

Parameters:

knob_name (str) -- The original knob name.

Returns:

The modified knob name suitable for output.

Return type:

str

omc3.machine_data_extraction.utils.strip_i_meas(text: str) → str[source]

Remove the I_MEAS suffix from a variable name.

Parameters:

text (str) -- The variable name possibly ending with ‘:I_MEAS’.

Returns:

The variable name without the ‘:I_MEAS’ suffix.

Return type:

str

omc3.machine_data_extraction.utils.timestamp_to_utciso(timestamp: float) → str[source]

Convert a timestamp to an ISO format string.

omc3.machine_data_extraction.utils.try_to_acquire_data(function: Callable, *args, **kwargs)[source]

Tries to get data from function multiple times.

Args:

function (Callable): function to be called, e.g. db.get args, kwargs: arguments passed to function

Returns:

Return arguments of function