Utils

Command Line Utilities

Utility class and functions to help run commands and access the command line.

class pyhdtoolkit.utils.cmdline.CommandLine[source]

New in version 0.2.0.

A high-level object to encapsulate the different methods for interacting with the commandline.

static check_pid_exists(pid: int) bool[source]

New in version 0.2.0.

Check whether the given PID exists in the current process table.

Parameters

pid (int) -- the Process ID you want to check.

Returns

A boolean stating the result.

Example

CommandLine.check_pid_exists(os.getpid())
True
static run(command: str, shell: bool = True, env: Mapping = None, timeout: float = None) Tuple[Optional[int], bytes][source]

New in version 0.2.0.

Runs command through subprocess.Popen and returns the tuple of (returncode, stdout).

Note

Note that stderr is redirected to stdout. Here shell is identical to the same parameter of subprocess.Popen.

Parameters
  • command (str) -- string, the command you want to run.

  • shell (bool) -- same as subprocess.Popen argument. Setting the shell argument to a True value causes subprocess to spawn an intermediate shell process, and tell it to run the command. In other words, using an intermediate shell means that variables, glob patterns, and other special shell features in the command string are processed before the command is ran. Defaults to True.

  • env (Mapping) -- mapping that defines the environment variables for the new process.

  • timeout (float) -- same as the subprocess.Popen.communicate argument, number of seconds to wait for a response before raising a TimeoutExpired exception.

Returns

The tuple of (returncode, stdout). Beware, the stdout will be a byte array (id est b"some returned text"). This output, returned as stdout, needs to be decoded properly before you do anything with it, especially if you intend to log it into a file. While it will most likely be “utf-8”, the encoding can vary from system to system so the standard output is returned in bytes format and should be decoded later on.

Raises
  • If the process does not terminate after timeout seconds, a TimeoutExpired exception --

  • will be raised. --

Examples

CommandLine.run("echo hello")
# returns (0, b"hello\r\n")
import os
modified_env = os.environ.copy()
modified_env["ENV_VAR"] = "new_value"
CommandLine.run("echo $ENV_VAR", env=modified_env)
# returns (0, b"new_value")
static terminate(pid: int) bool[source]

New in version 0.2.0.

Terminates the process corresponding to the given PID. On other platforms, using os.kill with signal.SIGTERM to kill.

Parameters

pid (int) -- the process ID to kill.

Returns

A boolean stating the success of the operation.

Example

CommandLine.terminate(500_000)  # max PID is 32768 (99999) on linux (macOS).
# returns False

Context Utilities

Provides useful contexts to use functions in.

pyhdtoolkit.utils.contexts.timeit(function: Callable) Iterator[None][source]

New in version 0.4.0.

Returns the time elapsed when executing code in the context via function. Original code from is from Jaime Coello de Portugal.

Parameters

function (Callable) -- any callable taking one argument. This was conceived with a lambda in mind.

Returns

The elapsed time as an argument for the provided function.

Example

with timeit(lambda spanned: logger.debug(f"Did some stuff in {spanned} seconds")):
    some_stuff()
    some_other_stuff()

Executors Utilities

A module providing two classes to execute functions + arguments couples through either a multiprocessing approach, or a multithreading approach. Before using these, here are a few tidbits to keep in mind:

  • There can only be one thread running at any given time in a python process because of the GIL.

  • Multiprocessing is parallelism. Multithreading is concurrency.

  • Multiprocessing is for increasing speed. Multithreading is for hiding latency.

  • Multiprocessing is best for computations. Multithreading is best for IO.

  • If you have CPU heavy tasks, use multiprocessing with n_process = n_cores and never more. Never.

  • If you have IO heavy tasks, use multithreading with n_threads = m * n_cores with m a number bigger than 1 that you can tweak on your own. Try many values and choose the one with the best speedup because there isn’t a general rule.

For instance the default value of m in ThreadPoolExecutor is set to 5 which I think is quite random.

class pyhdtoolkit.utils.executors.MultiProcessor[source]

New in version 0.2.0.

A class to easily wrap a multi-processing context manager call to a function.

Important

  • Reminder: multiprocessing is good for cpu-heavy tasks.

  • Reminder: only picklable objects can be executed and returned.

Example

Processor = MultiProcessor()
results_one_tuple_per_run = Processor.execute_function(
    func=your_cpu_heavy_function,
    func_args=list_of_args_for_each_call,
    n_processes=some_int_up_to_you,
)
static execute_function(func: Callable, func_args: list, n_processes: int) List[tuple][source]

New in version 0.2.0.

Executes the function with the provided arguments func_args as multiple processes.

Warning

Do not fire up more processes than you have cores! Never!

Parameters
  • func (Callable) -- the function to call.

  • func_args (list) -- list of the different parameters for each call. If function takes more than one parameter, wrap them up in tuples, e.g.: [(params, run, one), (params, run, two), (params, run, three)].

  • n_processes (int) -- the number of processes to fire up. No more than your number of cores! If n_processes is None or not given, ProcessPoolExecutor will default it to the number of processors on the machine.

Returns

A list of tuples, each tuple being the returned value(s) of function for the given call, for instance [(results, run, one), (results, run, two), (results, run, three)].

Example

MultiProcessor.execute_function(
    func=np.square, func_args=list(range(6)), n_processes=2
)
# returns [0, 1, 4, 9, 16, 25]
class pyhdtoolkit.utils.executors.MultiThreader[source]

New in version 0.2.0.

A class to easily wrap a multi-threading context manager call to a function.

Important

Reminder: multithreading is good for IO-heavy tasks.

Example

Threader = MultiThreader()
results_one_tuple_per_run = Threader.execute_function(
    func=your_io_heavy_function,
    func_args=list_of_args_for_each_call,
    n_processes=some_int_up_to_you,
)
static execute_function(func: Callable, func_args: list, n_threads: int) List[tuple][source]

New in version 0.2.0.

Executes the function with the provided arguments func_args as multiple threads.

Note

Remember: there is no point of having more threads than the number calls to be executed, the excess threads would be idle and you wouldd lose the time spent to fire them up.

Parameters
  • func (Callable) -- the function to call.

  • func_args (list) -- list of the different parameters for each call. If function takes more than one parameter, wrap them up in tuples, e.g.: [(params, run, one), (params, run, two), (params, run, three)].

  • n_threads (int) -- the number of threads to fire up. No more than your number of tasks. If n_threads is None or not given, ThreadPoolExecutor will default it to the number of processors on the machine multiplied by 5, assuming that it is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for a ProcessPoolExecutor.

Returns

A list of tuples, each tuple being the returned value(s) of function for the given call, for instance [(results, run, one), (results, run, two), (results, run, three)].

Example

MultiThreader.execute_function(
    func=np.square, func_args=list(range(6)), n_processes=2
)
# returns [0, 1, 4, 9, 16, 25]

HTCondor Monitoring

A module with utility to query the HTCondor queue, process the returned data and display it nicely.

Note

This module is meant to be called as a script, but some of the individual functionality is made public API and one shoule be able to build a different monitor from the functions in here.

pyhdtoolkit.utils.htc_monitor.query_condor_q() str[source]

New in version 0.9.0.

Returns a decoded string with the result of the condor_q command, to get the status of the caller’ jobs.

pyhdtoolkit.utils.htc_monitor.read_condor_q(report: str) Tuple[List[pyhdtoolkit.models.htc.HTCTaskSummary], pyhdtoolkit.models.htc.ClusterSummary][source]

New in version 0.9.0.

Splits information from different parts of the condor_q command’s output into one clean, validated data structure.

Parameters

report (str) -- the utf-8 decoded string returned by the condor_q command.

Returns

A tuple. The first element is a list of each task summary given by condor_q, each as a validated HTCTaskSummary` object. The second element is a validated ClusterSummary object with scheduler identification and summaries of the user as well as all users’ statistics on this scheduler cluster.

Example

condor_q_output = get_the_string_as_you_wish(...)
tasks, cluster = read_condor_q(condor_q_output)

Logging Utilities

New in version 1.0.0.

The loguru package is used for logging throughout pyhdtoolkit, and this module provides utilities to easily set up a logger configuration. Different pre-defined formats are provided to choose from:

  • FORMAT1: will display the full time of the log message, its level, the calling line and the message itself. This is the default format.

  • FORMAT2: similar to FORMAT1, but the caller information is displayed at the end of the log line.

  • SIMPLE_FORMAT: minimal, displays the local time, the level and the message.

pyhdtoolkit.utils.logging.config_logger(level: Union[str, int] = 'INFO', format: str = '<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>', **kwargs) None[source]

New in version 0.8.2.

Resets the logger object from loguru, with sys.stdout as a sink and the provided format.

Parameters
  • level (Union[str, int]) -- the logging level to set. Case-insensitive if a string is given. Can be any of the loguru levels or their integer values equivalents. Defaults to INFO.

  • format (str) -- the format to use for the logger to display messages. Defaults to a pre-defined format in this module.

  • **kwargs -- any keyword argument is transmitted to the add call.

Example

Using the defaults and setting the logging level:

config_logger(level="DEBUG")

Specifying a custom format and setting the logging level:

from pyhdtoolkit.utils.logging import config_logger, SIMPLE_FORMAT
config_logger(level="DEBUG", format=SIMPLE_FORMAT)

Miscellanous Personnal Utilities

Private module that provides miscellaneous personnal utility functions.

Warning

The functions in here are intented for personal use, and will most likely not work on other people’s machines.

pyhdtoolkit.utils._misc.add_noise_to_arc_bpms(df: pandas.core.frame.DataFrame, min_index: int, stdev: float, columns: Sequence[str] = None) pandas.core.frame.DataFrame[source]

New in version 1.2.0.

Selects the appropriate non-IR BPMs according to the min index provided, and adds gaussian noise to each relevant column with the provided standard deviation.

Warning

This selects BPMs by ommission. It will find all IR BPMs up to min_index and will excluse these from the selection.

Important

The BPM names should be in the index of the dataframe. Selection is case-insensitive.

Parameters
  • df (pandas.DataFrame) -- the dataframe to add noise to.

  • min_index (int) -- the minimum index of the BPMs to add noise to. See warning caveat right above. This number is inclusive (i.e. the BPMs with this index will be selected).

  • stdev (float) -- the standard deviation of the gaussian noise to add.

  • columns (Sequence[str]) -- the columns to add noise to. If not given, all columns will be used. Defaults to None.

Returns

A new DataFrame with the noise added to the wanted columns.

Example

df = add_noise_to_arc_bpms(df, min_index=8, stdev=1e-6, columns=["DPSI"])
pyhdtoolkit.utils._misc.add_noise_to_ir_bpms(df: pandas.core.frame.DataFrame, max_index: int, stdev: float, columns: Sequence[str] = None) pandas.core.frame.DataFrame[source]

New in version 1.2.0.

Selects the appropriate IR BPMs according to the max index provided, and adds gaussian noise to each relevant column with the provided standard deviation.

Important

The BPM names should be in the index of the dataframe. Selection is case-insensitive.

Parameters
  • df (pandas.DataFrame) -- the dataframe to add noise to.

  • max_index (int) -- the maximum index of the IR BPMs to add noise to. This number is inclusive (i.e. the BPMs with this index will be selected).

  • stdev (float) -- the standard deviation of the gaussian noise to add.

  • columns (Sequence[str]) -- the columns to add noise to. If not given, all columns will be used. Defaults to None.

Returns

A new DataFrame with the noise added to the wanted columns.

Example

df = add_noise_to_ir_bpms(df, max_index=5, stdev=1e-6, columns=["DPSI"])
pyhdtoolkit.utils._misc.apply_colin_corrs_balance(madx: cpymad.madx.Madx) None[source]

New in version 0.20.0.

Applies the SbS local coupling correction settings from the 2022 commissioning as they were in the machine, and tilts of Q3s that would compensate for those settings. This way the bump of each corrector is very local to MQSX3 - MQSXQ3 and other effects can be added and studied, pretending a perfect local coupling correction.

Parameters

madx (cpymad.madx.Madx) -- an instanciated Madx object with your LHC setup.

Example

apply_colin_corrs_balance(madx)
pyhdtoolkit.utils._misc.get_betastar_from_opticsfile(opticsfile: Union[pathlib.Path, str]) float[source]

New in version 0.16.0.

Parses the \(\beta^{*}\) value from the opticsfile content, which is in the first lines. This contains a check that ensures the betastar is the same for IP1 and IP5. The values returned are in meters.

Note

For file in acc-models-lhc make sure to point to the strength file (see example below) where the \(\beta^{*}\) is set, as the opticsfile itself only contains call.

Parameters

opticsfile (Union[Path, str]) -- pathlib.Path object to the optics file, or string of the path to the file.

Returns

The \(\beta^{*}\) value parsed from the file.

Raises

AssertionError -- if the \(\beta^{*}\) value for IP1 and IP5 is not the same (in both planes too).

Example

get_betastar_from_opticsfile(
    "acc-models-lhc/strengths/ATS_Nominal/2022/squeeze/ats_30cm.madx"
)
# returns 0.3
pyhdtoolkit.utils._misc.log_runtime_versions() None[source]

New in version 0.17.0.

Issues a CRITICAL-level log stating the runtime versions of both pyhdtoolkit, cpymad and MAD-X.

Example

log_runtime_versions()
# 2022-10-05 15:06:26 | CRITICAL | pyhdtoolkit.utils._misc:39 - Using: pyhdtoolkit 1.0.0rc0 | cpymad 1.10.0  | MAD-X 5.08.01 (2022.02.25)
pyhdtoolkit.utils._misc.split_complex_columns(df: pandas.core.frame.DataFrame, drop: bool = False) pandas.core.frame.DataFrame[source]

New in version 1.2.0.

Find complex valued columns in df and split them into a column for the real and imaginary parts each. New columns will be named like the existing ones, with _REAL or _IMAG appended.

Parameters
  • df (tfs.TfsDataFrame) -- the dataframe to split columns in.

  • drop (bool) -- whether to drop the original complex columns or not. Defaults to False.

Returns

A new DataFrame with the complex columns split into real and imaginary parts, and the original complex columns potentially dropped.

Exemple:
df = split_complex_columns(df, drop=True)

Operations Utilities

A collection classes with utility functions to perform common / convenient operations on the classic Python structures.

Warning

This module contains un-used code and will be removed in a future release.

class pyhdtoolkit.utils.operations.ListOperations[source]

New in version 0.2.0.

A class to group some common / useful operations on lists or sequences.

static all_unique(sequence: Sequence) bool[source]

New in version 0.2.0.

Returns True if all the values in a flat list are unique, False otherwise.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

True if all elements are unique, False otherwise.

Example

ListOperations.all_unique([1, 2, 3, 5, 12, 0])
# returns True
static average_by(sequence: typing.Sequence, function: typing.Callable = <function ListOperations.<lambda>>) float[source]

New in version 0.2.0.

Returns the average of sequence after mapping each element to a value using the provided function. Use map to map each element to the value returned by function. Uses sum to sum all of the mapped values, divided by len.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • function (Callable) -- function to apply to elements of the sequence.

Returns

The average of each element’s result through function.

Example

ListOperations.average_by(
  [{'n': 4}, {'n': 2}, {'n': 8}, {'n': 6}], lambda x: x['n']
)
# returns 5.0
static bifurcate(sequence: Sequence, filters: List[bool]) Sequence[source]

New in version 0.2.0.

Splits values into two groups. If an element in filter is True, the corresponding element in the collection belongs to the first group; otherwise, it belongs to the second group. Uses list comprehension and enumerate to add elements to groups, based on filter.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • filters (List[bool]) -- a list of booleans.

Returns

A list of two lists, one for each boolean output of the filters.

Example

ListOperations.bifurcate(['beep', 'boop', 'foo', 'bar'], [True, True, False, True])
# returns [['beep', 'boop', 'bar'], ['foo']]
static bifurcate_by(sequence: Sequence, function: Callable) list[source]

New in version 0.2.0.

Splits values into two groups according to a function, which specifies which group an element in the input sequence belongs to. If the function returns True, the element belongs to the first group; otherwise it belongs to the second group. Uses list comprehension to add elements to groups, based on function.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on the elements of sequence, that should return a boolean.

Returns

A list of two lists, as groups of elements of sequence classified depending on their result when passed to function.

Example

ListOperations.bifurcate_by(list(range(5)), lambda x: x % 2 == 0)
# returns [[0, 2, 4], [1, 3]]
static chunk_list(sequence: Sequence, size: int) Sequence[source]

New in version 0.2.0.

Chunks a sequence into smaller lists of a specified size. If the size is bigger than that of sequence, return sequence to avoid unnecessary nesting. Uses list and range to create a list of the desired size. Uses map on that list and fills it with splices of sequence. Finally, returns the created list.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • size (int) -- the size of the wanted sublists.

Returns

A list of lists of length size (except maybe the last element), with elements from sequence.

Example

ListOperations.chunk_list(list(range(10)), 3)
# returns [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
static deep_flatten(sequence: Sequence) list[source]

New in version 0.2.0.

Recursively deep flattens sequence, no matter the nesting levels.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

A list with all elements of sequence, but flattened.

Example

ListOperations.deep_flatten([["a", "b"], [1, 2], None, [True, False]])
# returns ["a", "b", 1, 2, None True, False]
static eval_none(sequence: typing.Sequence, function: typing.Callable = <function ListOperations.<lambda>>) bool[source]

New in version 0.2.0.

Returns False if the provided function returns True for at least one element in sequence, True otherwise. Iterates over sequence to test if every element returns False based on function. Omit the seconds argument, function, to check if all elements are False.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on elements of sequence that should return a boolean.

Returns

A boolean. See first line of docstring.

Examples

ListOperations.eval_none([0, 0, 1, 0], lambda x: x >= 2)
# returns True
ListOperations.eval_none([0, 1, 2, 0], lambda x: x >= 2)
# returns False
static eval_some(sequence: typing.Sequence, function: typing.Callable = <function ListOperations.<lambda>>) bool[source]

New in version 0.2.0.

Returns True if the provided function returns True for at least one element in sequence, False otherwise. Iterates over the elements of sequence to test if every element returns True based on function. Omit the seconds argument, function, to check if all elements are True.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on elements of sequence that should return a boolean.

Returns

A boolean. See first line of docstring.

Examples

ListOperations.eval_some([0, 1, 2, 0], lambda x: x >= 2)
# returns True
ListOperations.eval_some([0, 0, 1, 0], lambda x: x >= 2)
# returns False
static get_indices(element, sequence: Sequence) List[int][source]

New in version 0.2.0.

Return all array indices at which element is located.

Parameters
  • element -- any reference element to check.

  • sequence (Sequence) -- a sequence containing objects comparable to elements. A string can be compared to an int in Python, custom objects probably won’t be comparable.

Returns

A list of all indices at which element is found in sequence. Empty list if element is not present in sequence at all.

Example

ListOperations.get_indices(0, [0, 1, 3, 5, 7, 3, 9, 0, 0, 5, 3, 2])
# returns [0, 7, 8]
static group_by(sequence: Sequence, function: Callable) Dict[str, list][source]

New in version 0.2.0.

Groups the elements of sequence based on the given function. Uses list in combination with map and function to map the values of sequence to the keys of an object. Uses list comprehension to map each element to the appropriate key.

Parameters
  • sequence (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on the elements of sequence that should return a boolean.

Returns

A dict with keys True and False, each having as value a list of all elements of sequence that were evaluated to respectively True or False through function.

Example

ListOperations.group_by(list(range(5)), lambda x: x % 2 == 0)
# returns {True: [0, 2, 4], False: [1, 3]}
static has_duplicates(sequence: Sequence) bool[source]

New in version 0.2.0.

Returns True if there are duplicate values in sequence, False otherwise. Uses set on the given sequence to remove duplicates, then compares its length with that of sequence.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

A boolean indicating the presence of duplicates in sequence.

Example

ListOperations.has_duplicates([1, 2, 1])
# returns True
static sample(sequence: Sequence) list[source]

New in version 0.2.0.

Returns a random element from sequence.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

A random element from sequence in a list (to manage potentially nested sequences as input).

Examples

ListOperations.sample(["a", "b", 1, 2, False])
# returns 2
static sanitize_list(sequence: Sequence) list[source]

New in version 0.2.0.

Removes falsey values from a sequence. Uses filter to filter out falsey values (False, None, 0, and "").

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

The sequence without falsey values.

Example

ListOperations.sanitize_list([1, False, "a", 2, "", None, 6, 0])
# returns [1, "a", 2, 6]
static shuffle(sequence: Sequence) Sequence[source]

New in version 0.2.0.

Randomizes the order of the elements in sequence, returning a new list. Uses an improved version of the Fisher-Yates algorithm to reorder the elements.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

A copy of sequence with original elements at a random index.

Examples

ListOperations.shuffle(["a", "b", 1, 2, False])
# returns ['b', 1, False, 2, 'a']
static spread(sequence: Sequence) list[source]

New in version 0.2.0.

Flattens the provided sequence, by spreading its elements into a new list. Loops over elements, uses list.extend if the element is a list, list.append otherwise. This might look like deep_flatten but is a subset of its functionality, and is used in deep_flatten.

Warning

This only works if all elements in sequence are iterables.

Parameters

sequence (Sequence) -- a sequence of elements.

Returns

The sequence flattened, see first docstring sentence.

Example

ListOperations.spread([list(range(5)), list(range(5))])
# returns [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]
static symmetric_difference_by(seq_1: Sequence, seq_2: Sequence, function: Callable) list[source]

New in version 0.2.0.

Returns the symmetric difference of the provided sequences, after applying the provided function to each element of both. Creates a set by applying function to each element in each sequence, then uses list comprehension in combination with function on each one to only keep values not contained in the previously created set of the other.

Parameters
  • lst_1 (Sequence) -- a sequence of elements.

  • lst_2 (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on elements of seq_1 and seq_2.

Returns

A list, see first docstring sentence reference.

Examples

ListOperations.symmetric_difference_by([2.1, 1.2], [2.3, 3.4], math.floor)
# returns [1.2, 3.4]
ListOperations.symmetric_difference_by([2.1, 1.2], [0.5, 1.2], lambda x: x >= 2)
# returns [2.1]
static union_by(seq_1: Sequence, seq_2: Sequence, function: Callable) list[source]

New in version 0.2.0.

Returns every element that exists in any of the two sequences once, after applying the provided function to each element of both. This is the set theory union of the two sequences, but based on the results of applying function to each.

Note

Python’s set is strange in how is gives output, so this function sorts the final list before returning it, in order to give it predictable behavior.

Creates a set by applying function to each element in seq_1, then uses list comprehension in combination with function on seq_2 to only keep values not contained in the previously created set, _lst_1. Finally, create a set from the previous result and _lst_1, and transform it into a list.

Parameters
  • lst_1 (Sequence) -- a sequence of elements.

  • lst_2 (Sequence) -- a sequence of elements.

  • function (Callable) -- a callable on elements of seq_1 and seq_2.

Returns

A list, see first docstring sentence reference.

Example

ListOperations.union_by([2.1], [1.2, 2.3], math.floor)
# returns [1.2, 2.1]
static zipper(*args, fillvalue=None) list[source]

New in version 0.2.0.

Creates a list of lists of elements, each internal list being a grouping based on the position of elements in the original sequences. Essentially, a list containing:

  • a first list with all first elements,

  • then a second list with all second elements, etc.

Uses max combined with list comprehension to get the length of the longest list in the arguments. Loops for max_length times grouping elements. If lengths of sequences vary, uses fill_value to adjust the smaller ones (defaults to None).

Parameters
  • *args -- a number (>= 2) of different iterables.

  • fillvalue -- value to use in case of length mismatch.

Returns

A list with the proper level of nesting, and original elements zipped.

Example

ListOperations.zipper([1, 2, 3], [2, 5, 3, 7], ["a", "b", "c"])
# returns [[1, 2, 'a'], [2, 5, 'b'], [3, 3, 'c'], [None, 7, None]]
class pyhdtoolkit.utils.operations.MiscellaneousOperations[source]

New in version 0.2.0.

A class to group some misc. operations that don’t pertain to classic structures.

static longest_item(*args)[source]

New in version 0.2.0.

Takes any number of iterable objects, or objects with a length property, and returns the longest one. If multiple objects have the same length, the first one will be returned. Usess max with len as the key to return the item with the greatest length.

Parameters

*args -- any number (>= 2) of iterables.

Returns

The longest elements of provided iterables.

Example

MiscellaneousOperations.longest_item(
    list(range(5)), list(range(100)), list(range(50))
)
# returns list(range(100))
static map_values(obj: dict, function: Callable) dict[source]

New in version 0.2.0.

Creates an new dict with the same keys as the provided obj, and values generated by running function on the obj’s values. Uses dict.keys to iterate over the object’s keys, assigning the values produced by function to each key of a new object.

Parameters
  • obj (dict) -- a dictionary.

  • function (Callable) -- a callable on values of obj.

Returns

A new dictionary with the results.

Example

MiscellaneousOperations.map_values(
    {"a": list(range(5)), "b": list(range(10)), "c": list(range(15))},
    lambda x: len(x)
)
# returns {"a": 5, "b": 10, "c": 15}
class pyhdtoolkit.utils.operations.NumberOperations[source]

New in version 0.2.0.

A class to group some common / useful operations on numbers.

static clamp_number(num: Union[int, float], a_val: Union[int, float], b_val: Union[int, float]) Union[int, float][source]

New in version 0.2.0.

Clamps num within the inclusive range specified by the boundary values a_val and b_val. If num falls within the range, returns num. Otherwise, return the nearest number in the range.

Parameters
  • num (Union[int, float]) -- a number (float / int)

  • a_val (Union[int, float]) -- a number (float / int)

  • b_val (Union[int, float]) -- a number (float / int)

Returns

A number (float / int), being the nearest to num in the range [a_val, b_val].

Examples

NumberOperations.clamp_number(17, 4, 5)
# returns 5
NumberOperations.clamp_number(23, 20, 30)
# returns 23
static degrees_to_radians(deg_value: Union[int, float], decompose: bool = False) Union[Tuple[float, str, str], int, float][source]

New in version 0.2.0.

Converts an angle from degrees to radians. Uses math.pi and the degree to radian formula to convert the provided deg_value.

Parameters
  • deg_value (Union[int, float]) -- angle value in degrees.

  • decompose (bool) -- boolean option to return a more verbose result. Defaults to False.

Returns

The angle value in radians.

Examples

NumberOperations.degrees_to_radians(160)
# returns 2.792526803190927
NumberOperations.degrees_to_radians(360, decompose=True)
# returns (2, "pi", "rad")
static greatest_common_divisor(sequence: Sequence) Union[int, float][source]

New in version 0.2.0.

Calculates the greatest common divisor of a sequence of numbers. Uses reduce and math.gcd over the given sequence.

Parameters

sequence (Sequence) -- a sequence of numbers (floats are advised against as this would become a very heavy computation).

Returns

The greatest common divisor of all elements in sequence.

Examples

NumberOperations.greatest_common_divisor([54, 24])
# returns 6
NumberOperations.greatest_common_divisor([30, 132, 378, 582, 738])
# returns 6
static is_divisible_by(dividend: Union[int, float], divisor: Union[int, float]) bool[source]

New in version 0.2.0.

Checks if the first numeric argument is divisible by the second one. Uses the modulo operator (%) to check if the remainder is equal to 0.

Parameters
  • dividend (Union[int, float]) -- a number.

  • divisor (Union[int, float]) -- a number.

Returns

A boolean stating if dividend can be divided by divisor.

Examples

NumberOperations.is_divisible_by(35, 15)
# returns False
static least_common_multiple(*args) int[source]

New in version 0.2.0.

Returns the least common multiple of two or more numbers. Defines a function, spread, that uses either list.extend or list.append on each element of s sequence to flatten it. Uses math.gcd and lcm(x,y) = (x * y) / gcd(x,y) to determine the least common multiple.

Parameters

*args -- any number (>= 2) of numbers (floats are advised against as this would become a very heavy computation).

Returns

The least common multiple of all provided numbers.

Examples

NumberOperations.least_common_multiple(4, 5)
# returns 20
NumberOperations.least_common_multiple(2, 5, 17, 632)
# returns 53720
static radians_to_degrees(rad_value: Union[int, float]) Union[int, float][source]

New in version 0.2.0.

Converts an angle from radians to degrees. Uses math.pi and the radian to degree formula to convert the provided rad_value.

Parameters

rad_value (Union[int, float]) -- angle value in degrees.

Returns

The angle value in degrees.

Examples

NumberOperations.radians_to_degrees(2* math.pi)
# returns 360
NumberOperations.radians_to_degrees(2.710)
# returns 155.2715624804531
class pyhdtoolkit.utils.operations.StringOperations[source]

New in version 0.2.0.

A class to group some common / useful operations on strings.

static camel_case(text: str) str[source]

New in version 0.2.0.

Converts a string to camelCase. Breaks the string into words and combines them capitalizing the first letter of each word, using a regexp, title and lower.

Parameters

text (str) -- a string.

Returns

The same string best adapted to camelCase.

Examples

StringOperations.camel_case("a_snake_case_name")
# returns "aSnakeCaseName"
StringOperations.camel_case("A Title Case Name")
# returns "aTitleCaseName"
static capitalize(text: str, lower_rest: bool = False) str[source]

New in version 0.2.0.

Capitalizes the first letter of a string, eventually lowers the rest of it. Omit the lower_rest parameter to keep the rest of the string intact, or set it to True to convert to lowercase.

Parameters
  • text (str) -- a string.

  • lower_rest (bool) -- boolean option to lower all elements starting from the second.

Returns

The string, capitalized.

Examples

StringOperations.capitalize("astringtocapitalize")
# returns "Astringtocapitalize"
StringOperations.capitalize("astRIngTocApItalizE", lower_rest=True)
# returns "Astringtocapitalize"
static is_anagram(str_1: str, str_2: str) bool[source]

New in version 0.2.0.

Checks if a string is an anagram of another string (case-insensitive, ignores spaces, punctuation and special characters). Uses str.replace to remove spaces from both strings. Compares the lengths of the two strings, return False if they are not equal. Uses sorted on both strings and compares the results.

Parameters
  • str_1 (str) -- a string.

  • str_2 (str) -- a string.

Returns

A boolean stating whether str_1 is an anagram of str_2 or not.

Examples

StringOperations.is_anagram("Tom Marvolo Riddle", "I am Lord Voldemort")
# returns True
StringOperations.is_anagram("A first string", "Definitely not an anagram")
# returns False
static is_palindrome(text: str) bool[source]

New in version 0.2.0.

Returns True if the given string is a palindrome, False otherwise. Uses str.lower and re.sub to convert to lowercase and remove non-alphanumeric characters from the given string. Then compare the new string with its reverse.

Parameters

text (str) -- a string.

Returns

A boolean stating whether text is a palindrome or not.

Examples

StringOperations.is_palindrome("racecar")
# returns True
StringOperations.is_palindrome("definitelynot")
# returns False
static kebab_case(text: str) str[source]

New in version 0.2.0.

Converts a string to kebab-case. Breaks the string into words and combines them adding - as a separator, using a regexp.

Parameters

text (str) -- a string.

Returns

The same string best adapted to kebab_case.

Examples

StringOperations.kebab_case("camel Case")
# returns "camel-case"
StringOperations.kebab_case("snake_case")
# returns "snake-case"
static snake_case(text: str) str[source]

New in version 0.2.0.

Converts a string to snake_case. Breaks the string into words and combines them adding _ as a separator, using a regexp.

Parameters

text (str) -- a string.

Returns

The same string best adapted to snake_case.

Examples

StringOperations.snake_case("A bunch of words")
# returns "a_bunch_of_words"
StringOperations.snake_case("camelCase")
# returns "camelcase"

Printing Utilities

A class utility class to allow me printing text in color, bold, etc.

Warning

This module’s functionality is superseeded by the rich package, and it will be removed in a future release.

class pyhdtoolkit.utils.printutil.Background[source]

New in version 0.2.0.

ANSI color escape sequences for the background of a terminal output.

class pyhdtoolkit.utils.printutil.Foreground[source]

New in version 0.2.0.

ANSI color escape sequences for the foreground of a terminal output.

class pyhdtoolkit.utils.printutil.Styles[source]

New in version 0.2.0.

ANSI style escape sequences for a terminal output.