Utils

Contexts

Provides contexts managers to use.

omc3.utils.contexts.log_out(stdout=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, stderr=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>)

Temporarily changes sys.stdout and sys.stderr.

omc3.utils.contexts.silence()

Suppress all console output, rerouting sys.stdout and sys.stderr to devnull.

omc3.utils.contexts.suppress_warnings(warning_classes)

Suppress all warnings of given classes.

omc3.utils.contexts.timeit(function)

Prints the time at the end of the context via function.

IO Tools

Helper functions for input/output issues.

class omc3.utils.iotools.OptionalStr(value)

A class that allows str or None. Can be used in string-lists when individual entries can be None.

class omc3.utils.iotools.PathOrStr(value)

A class that behaves like a Path when possible, otherwise like a string.

class omc3.utils.iotools.PathOrStrOrDataFrame(value)

A class that behaves like a Path when possible, otherwise like a string.

class omc3.utils.iotools.UnionPathStr(value)

A class that can be used as Path and string parser input, but does not convert to path.

class omc3.utils.iotools.UnionPathStrInt(value)

A class that can be used as Path, string, int parser input, but does not convert. Very special and used e.g. in the BBQ Input.

omc3.utils.iotools.always_true(*args, **kwargs) → bool

A function that is always True.

omc3.utils.iotools.convert_paths_in_dict_to_strings(dict_: dict) → dict

Converts all Paths in the dict to strings, including those in iterables.

omc3.utils.iotools.copy_content_of_dir(src_dir: Path, dst_dir: Path)

Copies all files and directories from src_dir to dst_dir.

omc3.utils.iotools.copy_item(src_item: Path, dst_item: Path)

Copies a file or a directory to dest, which may be a directory. If src_item is a directory then all containing files and dirs will be copied into dest.

omc3.utils.iotools.create_dirs(path_to_dir: str | Path)

Creates all dirs to path_to_dir if not exists. TODO: Change all calls to use only Path.

omc3.utils.iotools.get_check_suffix_func(suffix: str) → Callable[[Path], bool]

Returns a function that checks the suffix of a given path agains the suffix.

omc3.utils.iotools.glob_regex(path: Path, pattern: str) → Iterator[str]

Do a glob on the given path based on the regular expression pattern. Returns only the matching filenames (as strings).

Parameters:

• path (Path) -- Folder path to look in.

• pattern (str) -- Pattern to match.

Returns:

Matching filenames

Return type:

Iterator[str]

omc3.utils.iotools.maybe_add_command(opt: dict, script: str) → dict

Add a comment ‘;command’ to the opt-dict, which is the command used to run the script gotten from sys.argv, but only if the executed file (the file that run with the python command) equals script, i.e. the script for which you are saving the parameters is the main script being run. Otherwise the command is probably unrelated.

Parameters:

• opt (dict) -- Options datastructure

• script (str) -- Name of the script that called save_config.

Returns:

Updated dict with ;command entry, or if the script names were different, just the original opt.

omc3.utils.iotools.remove_none_dict_entries(dict_: dict) → dict

Removes None entries from dict. This can be used as a workaround to https://github.com/pylhc/generic_parser/issues/26.

omc3.utils.iotools.replace_in_path(path: Path, old: Path | str, new: Path | str) → Path

Replace a part of a path with a new path. Useful for example to replace the original path with a path to a symlink or vice versa.

Parameters:

• path (Path) -- Path object to replace the subpath in

• old (Union[Path, str]) -- Subpath to be replaced

• new (Union[Path, str]) -- Subpath to replace with

Returns:

New Path object with the replacement in.

Return type:

Path

omc3.utils.iotools.save_config(output_dir: Path, opt: dict, script: str, unknown_opt: dict | list = None)

Quick wrapper for save_options_to_config.

Parameters:

• output_dir (Path) -- Path to the output directory (does not need to exist).

• opt (dict) -- opt-structure to be saved.

• script (str) -- path/name of the invoking script (becomes name of the .ini) usually __file__.

• unknown_opt (dict|list) -- un-parsed opt-structure to be saved.

omc3.utils.iotools.strip_quotes(value: Any) → Any

Strip quotes around string-objects. If not a string, nothing is changed. This is because the input from commandline or json files could be surrounded by quotes (if they are strings). The dict-parser removes them automatically as well. This behaviour is important for basically every string-faker!

Parameters:

value (Any) -- The input value that goes into the function. Can be of any type.

Returns:

If the input was a string, then it will be the string with stripped quotes (if there were any). Otherwise just the value.

Logging Tools

Functions for easier use of logging, like automatic logger setup (see: get_logger()).

class omc3.utils.logging_tools.DebugMode(active=True, log_file=None)

Context Manager for the debug mode. Hint: Does not work with @contextmanager from contextlib (even though nicer code), as the _get_caller would find contextlib.py.

Parameters:

• active (bool) -- Defines if this manager is doing anything. Defaults to True.

• log_file (str) -- File to log into.

class omc3.utils.logging_tools.MaxFilter(level)

To get messages only up to a certain level.

class omc3.utils.logging_tools.TempFile(file_path, log_func)

Context Manager. Lets another function write into a temporary file and logs its contents. It won’t open the file, so only the files path is returned.

Parameters:

• file_path (str) -- Place to write the tempfile to.

• log_func (func) -- The function with which the content should be logged (e.g. LOG.info).

class omc3.utils.logging_tools.TempStringLogger(module=None, level=20)

Temporarily log into a string that can be retrieved by get_log.

Parameters:

• module -- module to log, defaults to the caller file.

• level -- logging level, defaults to INFO.

get_log()

Get the log as string.

omc3.utils.logging_tools.add_module_handler(handler)

Add handler at current module level.

omc3.utils.logging_tools.add_root_handler(handler)

Add handler at root level.

omc3.utils.logging_tools.file_handler(logfile, level=10, fmt='%(levelname)7s | %(message)s | %(name)s')

Convenience function so the caller does not have to import logging.

omc3.utils.logging_tools.getLogger(name)

Convenience function so the caller does not have to import logging.

omc3.utils.logging_tools.get_logger(name, level_root=10, level_console=None, fmt='%(levelname)7s | %(message)s | %(name)s', color=None)

Sets up logger if name is __main__. Returns logger based on module name.

Parameters:

• name -- only used to check if __name__ is __main__.

• level_root -- main logging level, defaults to DEBUG.

• level_console -- console logging level, defaults to INFO.

• fmt -- Format of the logging. For default see BASIC_FORMAT.

• color -- If None colors are used if tty is detected. False will never use colors and True will always enforce them.

Returns:

Logger instance.

omc3.utils.logging_tools.get_my_logger_name()

Return the logger name for the caller.

omc3.utils.logging_tools.list2str(list_: list) → str

Returns string representation of list_, but without brackets.

omc3.utils.logging_tools.log_pandas_settings_with_copy(log_func)

Logs pandas SettingsWithCopy warning to loc_func instead of printing the warning.

omc3.utils.logging_tools.logging_silence()

Remove temporarily all loggers from root logger.

omc3.utils.logging_tools.odr_pprint(printer, odr_out)

Logs the odr output results. Adapted from odr_output pretty print.

omc3.utils.logging_tools.stream_handler(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, level=10, fmt='%(levelname)7s | %(message)s | %(name)s', max_level=None)

Convenience function so the caller does not have to import logging.

omc3.utils.logging_tools.unformatted_console_logging()

Log only to console and only unformatted.

Mock

Provides mock functionality for packages necessitating the CERN GPN and only installable from the acc-py package index, such as pytimber, pjlsa etc.

from omc3.utils.mock import cern_network_import
pytimber = cern_network_import("pytimber")
db = pytimber.LoggingDB(source="nxcals")  # will raise if pytimber not installed

class omc3.utils.mock.CERNNetworkMockPackage(name: str)

Mock class to raise an error if the desired package functionality is called when the package is not actually installed. Designed for packages installable only from inside the CERN network, that are declared as cern extra. See module documentation.

omc3.utils.mock.cern_network_import(package: str)

Convenience function to try and import packages only available (and installable) on the CERN network. If installed, the module is returned, otherwise a mock class is returned, which will raise an insightful ImportError on attempted use.

Parameters:

package (str) -- name of the package to try and import.

Outliers

Helper functions for outlier detection.

omc3.utils.outliers.get_filter_mask(data: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], x_data: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes] = None, limit: float = 0.0, niter: int = 20, nsig: int = None, mask: Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes] = None) → Buffer | _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes]

Filters the array of values which are meant to be constant or a linear function of the x-data array if that is provided, by checking how many sigmas they are deviating from the average.

The outlier filtering function is utilized at multiple stages of the data analysis. It removes data points in the tails of the measured distribution, which are too populated due to the finite sample size, assuming a normal distribution specified by measured mean and standard deviation of the given data. A data point, outside of a user-specified limit, is removed if there is less than a 50% chance that it is from the specified normal distribution.

In particular: The function gets an array \(y\) of data of length \(n_y\), which can be any scalar data and is currently used for e.g. tune data measured per BPM or a time series from the BBQ. In addition, a cleaning limit can be given, inside which the data points are always kept. Further an optional array \(x\) (via the parameter x_data) can be given, in which case a linear fit \(y = ax + c\) is attempted to remove any slope on the data, before calculating momenta on the data. An upper boundary in \(n_\sigma\) is then determined from the percent point function (ppf) of a Student’s t-distribution, given \(n_x\) degrees of freedom at \(100\% - 50\% / n_y\). Iteratively, mean \(\left< y \right>\) and standard deviation \(\sigma\) of the data is calculated and only data within \(n_\sigma \cdot \sigma\), or the given cleaning limit, around \(\left< y \right>\) is kept for the next iteration. The loop stops either after 20 iterations, when there is no more data outside the boundaries or when there are less than three data points remaining.

Returns a filter mask for the original array (meaning True for elements that should be kept).

Parameters:

• data (ArrayLike) -- Data to filter.

• x_data (ArrayLike) -- Additional x-data. If given a linear function is fit on the data, and the outliers filtered on the difference to this function. Otherwise, the distribution of data is used directly for filtering.

• limit (float) -- Only data deviating more than this limit from the average is filtered.

• niter (int) -- Maximum number of filter iterations to do. Iterations are interrupted if the number of datapoints did not shrink in the last iteration or if there is only two or less data-points left.

• nsig (int) -- number of sigmas within the data points are considered okay, outside they are considered outliers. If not given, nsigmas is got from the 1-0.5/length(data) percentile of a student-t distribution.

• mask (ArrayLike) -- Boolean mask of data-points to consider. If None, all data is considered.

Returns:

Boolean array containing True entries for data-points that are good,

and False for the ones that should be filtered.

Return type:

ArrayLike

Stats

Helper module providing statistical methods to compute various weighted averages along specified axis and their errors as well as unbiased error estimator of infinite normal distribution from finite-sized sample.

• TODO: use weighted average and its error in circular calculations

• TODO: write tests

• TODO: LOGGER or Raising error and warnings?

• TODO: if zeros or nans occur in errors, fallback to uniform weights only in affected cases

omc3.utils.stats.circular_error(data, period=6.283185307179586, errors=None, axis=None, t_value_corr=True)

Computes error of weighted circular average along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• period -- scalar, optional Periodicity period of data, default is (2 * np.pi)

• errors -- array-like, optional Contains errors associated with the values in data, it is used to calculated weights

• axis -- int or tuple of ints, optional Axis or axes along which to average data

• t_value_corr -- bool, optional Species if the error is corrected for small sample size, default True

Returns:

Returns the error of weighted circular average along the specified axis.

omc3.utils.stats.circular_mean(data, period=6.283185307179586, errors=None, axis=None)

Computes weighted circular average along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• period -- scalar, optional, default (2 * np.pi) Periodicity period of data

• errors -- array-like, optional Contains errors associated with the values in data, it is used to calculated weights

• axis -- int or tuple of ints, optional Axis or axes along which to average data

Returns:

Returns the weighted circular average along the specified axis.

omc3.utils.stats.circular_nanerror(data, period=6.283185307179586, errors=None, axis=None, t_value_corr=True)

“Wrapper around circular_error with added nan handling

omc3.utils.stats.circular_nanmean(data, period=6.283185307179586, errors=None, axis=None)

“Wrapper around circular_mean with added nan handling

omc3.utils.stats.circular_rms(data, period=6.283185307179586, axis=None)

Computes the circular root mean square along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• period -- scalar, optional Periodicity period of data, default is (2 * np.pi)

• axis -- int or tuple of ints, optional Axis or axes along which to average data

Returns:

Returns the circular root mean square along the specified axis.

omc3.utils.stats.effective_sample_size(data, weights, axis=None)

Computes effective sample size of weighted data along specified axis, the minimum value returned is 2 to avoid non-reasonable error blow-up.

It is calculated via Kish’s approximate formula from the (not necessarily normalized) weights \(w_i\) (see wikipedia):

\[n_\mathrm{eff} = \frac{\left(\sum_i w_i\right)^2}{\sum_i w_i^2}\]

What it represents: “In most instances, weighting causes a decrease in the statistical significance of results. The effective sample size is a measure of the precision of the survey (e.g., even if you have a sample of 1,000 people, an effective sample size of 100 would indicate that the weighted sample is no more robust than a well-executed un-weighted simple random sample of 100 people).” - https://wiki.q-researchsoftware.com/wiki/Weights,_Effective_Sample_Size_and_Design_Effects

Parameters:

• data -- array-like

• weights -- array-like Contains weights associated with the values in data

• axis -- int or tuple of ints, optional Axis or axes along which the effective sample size is computed

Returns:

Returns the error of weighted circular average along the specified axis.

omc3.utils.stats.rms(data, axis=None)

Computes the root mean square along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• axis -- int or tuple of ints, optional Axis or axes along which to average data

Returns:

Returns the root mean square along the specified axis.

omc3.utils.stats.t_value_correction(sample_size)

Calculates the multiplicative correction factor to determine standard deviation of a normally distributed quantity from standard deviation of its finite-sized sample. The minimum allowed sample size is 2 to avoid non-reasonable error blow-up for smaller sample sizes 2 is used instead.

Note (jdilly): In other words, this transforms the area of 1 sigma under the given student t distribution to the 1 sigma area of a normal distribution (this transformation is where the CONFIDENCE LEVEL comes in). I hope that makes the intention more clear.

Parameters:

sample_size -- array-like

Returns:

multiplicative correction factor(s) of same shape as sample_size. Can contain nans.

omc3.utils.stats.unbias_variance(data, weights, axis=None)

Computes a correction factor to unbias variance of weighted average of data along specified axis, e.g. transform the standard deviation 1

\[\sigma^2 = \frac{1}{N} \sum_{i=1}^N (x_i - x_\mathrm{mean})^2\]

into an un-biased estimator

\[\sigma^2 = \frac{1}{N-1} \sum_{i=1}^N (x_i - x_\mathrm{mean})^2\]

Parameters:

• data -- array-like

• weights -- array-like Contains weights associated with the values in data

• axis -- int or tuple of ints, optional Axis or axes along which the effective sample size is computed

Returns:

Returns the error of weighted circular average along the specified axis.

omc3.utils.stats.weighted_error(data, errors=None, axis=None, t_value_corr=True)

Computes error of weighted average along the specified axis. This is similar to calculating the standard deviation on the data, but with both, the average to which the deviation is calculated, as well as then the averaging over the deviations weighted by weights based on the errors.

In addition, the weighted variance is unbiased by an unbias-factor n / (n-1), where n is the omc3.utils.stats.effective_sample_size() . Additionally, a (student) t-value correction can be performed (done by default) which corrects the estimate for small data sets.

Parameters:

• data -- array-like Contains the data on which the weighted error on the average is calculated.

• errors -- array-like, optional Contains errors associated with the values in data, it is used to calculated weights

• axis -- int or tuple of ints, optional Axis or axes along which to average data

• t_value_corr -- bool, optional Species if the error is corrected for small sample size, default True

Returns:

Returns the error of weighted average along the specified axis.

omc3.utils.stats.weighted_mean(data, errors=None, axis=None)

Computes weighted average along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• errors -- array-like, optional Contains errors associated with the values in data, it is used to calculated weights

• axis -- int or tuple of ints, optional Axis or axes along which to average data

Returns:

Returns the weighted average along the specified axis.

omc3.utils.stats.weighted_nanmean(data, errors=None, axis=None)

“Wrapper around weighted_mean with added nan handling

omc3.utils.stats.weighted_nanrms(data, errors=None, axis=None)

“Wrapper around weigthed_rms with added nan handling

omc3.utils.stats.weighted_rms(data, errors=None, axis=None)

Computes weighted root mean square along the specified axis.

Parameters:

• data -- array-like Contains the data to be averaged

• errors -- array-like, optional Contains errors associated with the values in data, it is used to calculated weights

• axis -- int or tuple of ints, optional Axis or axes along which to average data

Returns:

Returns weighted root mean square along the specified axis.

omc3.utils.stats.weights_from_errors(errors, period=6.283185307179586)

Computes weights from measurement errors, weights are not output if errors contain zeros or nans

Parameters:

• errors -- array-like Contains errors which are used to calculated weights

• period -- scalar, optional Periodicity period of data, default is (2 * np.pi)

Returns:

Returns the error of weighted circular average along the specified axis.

Time Tools

Provides tools to handle times more easily, in particular to switch easily between local time and UTC time.

class omc3.utils.time_tools.AccDatetime(*args, **kwargs)

Wrapper for a datetime object to easily convert between local and UTC time as well as give different presentations.

add(**kwargs)

Add timedelta and return as new object.

cern_utc_string()

Get utc time as string (CERN format).

property datetime

Return normal datetime object (in case ducktyping does not work. Looking at you, mpl!).

classmethod from_cern_utc_string(s)

Create AccDatetime object from datetime in utc string.

classmethod from_local(dt)

Create AccDatetime object from datetime in local time.

classmethod from_local_string(s)

Create AccDatetime object from datetime in local time string.

classmethod from_timestamp(ts)

Create AccDatetime object from timestamp.

classmethod from_utc(dt)

Create AccDatetime object from datetime in utc.

classmethod from_utc_string(s)

Create AccDatetime object from datetime in utc string.

property local

Get local datetime object.

property local_string

Get local time as string.

property local_timezone

Get local timezone.

classmethod now()

Create AccDatetime object at now.

sub(**kwargs)

Subtract timedelta and return as new object.

property utc

Get UTC datetime object.

property utc_string

Get utc time as string.

omc3.utils.time_tools.AcceleratorDatetime = {'lhc': <class 'omc3.utils.time_tools.CERNDatetime'>, 'ps': <class 'omc3.utils.time_tools.CERNDatetime'>, 'sps': <class 'omc3.utils.time_tools.CERNDatetime'>}

Accelerator name to AccDatetime mapping.

Type:

dict

class omc3.utils.time_tools.CERNDatetime(*args, **kwargs)

AccDatetime class for all accelerators at CERN.

omc3.utils.time_tools.cern_utc_string_to_utc(utc_string)

Convert a time string in cern-utc to a utc datetime object.

omc3.utils.time_tools.check_tz(localized_dt, timezone)

Checks if timezone is correct.

omc3.utils.time_tools.get_cern_time_format()

Default time format.

omc3.utils.time_tools.get_cern_timezone()

Get time zone for cern measurement data.

omc3.utils.time_tools.get_readable_time_format()

Human readable time format.

omc3.utils.time_tools.local_string_to_utc(local_string, timezone)

Converts a time string in local time to UTC time.

omc3.utils.time_tools.local_to_utc(dt_local, timezone)

Convert local datetime object to utc datetime object.

omc3.utils.time_tools.utc_now()

Get UTC now as time.

omc3.utils.time_tools.utc_string_to_utc(utc_string)

Convert a time string in utc to a UTC datetime object.

omc3.utils.time_tools.utc_to_local(dt_utc, timezone)

Convert UTC datetime object to local datetime object.