ccatkidlib package

Subpackages

Submodules

ccatkidlib.io module

Library of helper functions for general directory/file IO operations as well as communication and IO with RFSoC boards.

ccatkidlib.io.add_dir(dir_name: str, data_dir: str, save_root: str | None = None, data_root: str | None = None, sub_dirs: list[str] = [], timestamp: str | None = None) → str

Add a new directory within/mimicing a pre-existing ccatkidlib file tree. The directory will be created in the already existing file tree structure if save_root is not specified, otherwise it will mimic the already existing structure within the specified save_root directory.

Parameters:

  • dir_name – Name of directory to add

  • data_dir – Data directory of ccatkidlib file tree

  • save_root – Root data directory where new directory should be created

  • data_root – Root data directory of ccatkidlib file tree

  • sub_dirs – Sub-directories to create within new directory. If no sub-directories specified, will mimic the sub-directories of the already existing file tree.

  • timestamp – Unix timestamp. If specified, will create a directory with the timestamp as the name within the specified sub_dirs (at the deepest level).

Returns:

Path of the newly added directory

ccatkidlib.io.combine_npy(files: list[str], num: int, com: str | None = None, fname_out: dict | None = None) → list[str]

Combine NumPy into NpzFile zipped archive files.

Parameters:

  • files – List of NumPy files to combine

  • num – Number of NpzFile files to create

  • com

  • fname_out

Returns:

File names of the NpzFile files that were created

ccatkidlib.io.create_dir(dir: str) → None

Create directory at the specified path. Will not overwrite if it already exists.

Parameters:

dir – Path of directory to be created

Raises:

  • FileNotFoundError – If an invalid dir is specified

  • PermissionError – If specified dir does not have suitable write permissions to be created

ccatkidlib.io.create_noise_files(com_to: list[str], tmp_dir: str) → list[str]

Create ccatkidlib /tmp directory and populate with empty noise tone files. Will not overwrite if directory or noise tone files already exist.

Parameters:

  • com_to – List of drones for which to create noise tone files

  • tmp_dir – Temporary directory where noise tone files should be created

Returns:

List of noise tone file paths in order of com_to

ccatkidlib.io.create_tree(com_to: list[str], curr_date: str, sess_id: int, data_dir: str) → tuple[list[str], list[str], list[str], list[str]]

Create file tree for storage of sweep, timestream, comb, and configuration files

Parameters:

  • com_to – List of RFSoC drones to create directories for

  • curr_date – Current date

  • sess_id – Session ID of measurement

  • data_dir – Directory in which to create file tree

Returns:

Tuple of /config directories, /target directories, /timestream directories, and /vna directories in that order.

ccatkidlib.io.edit_config(cfg: dict, key: str, value: Any, append: bool = False) → bool

Update key in cfg configuration file with the specified value.

Parameters:

  • cfg – Configuration file to update

  • key – Key that should be updated

  • value – Value with which to update key

  • append – Whether to add a new key, value pair to configuration file if key is not found

Returns:

True if key was successfully updated or created, otherwise False

Return type:

done

ccatkidlib.io.get_array(src_path: str, dest_path: str, action: Literal['cp', 'mv'] = 'cp', load: bool = True, timestamp: bool = False) → ndarray | str | None

Copy or move a NumPy array and load its contents

Parameters:

  • src_path – Path of the NumPy array

  • dest_path – Destination path where the NumPy array should be copied or moved. If path does not include a file name, the same file name will be used.

  • action – Whether to copy ‘cp’ or move ‘mv’ the NumPy array

  • load – Whether to load and return the contents of the NumPy array

  • timestamp – Whether to remove timestamp from NumPy array file name (Use False if no timestamp)

Returns:

Loaded NumPy array if load is True otherwise the file path where the NumPy array was copied or moved to. Returns None if failed to copy/move/load NumPy array

ccatkidlib.io.get_array_board(c: Connection, ip: str, ssh_key: str, remote_path: str, local_path: str, load: bool = True, timestamp: bool = False) → ndarray | str | None

Load NumPy array from RFSoC board

Parameters:

  • c – Fabric Connection object of RFSoC board

  • ip – IP address of RFSoC board

  • ssh_key – File path of private SSH key corresponding to public key on the RFSoC board

  • remote_path – File path of NumPy array on RFSoC board

  • local_path – Path on local machine where NumPy array should be copied. If it does not contain a file name, the same file name will be used as that on the RFSoC board

  • load – Whether to load and return the contents of the NumPy array

  • timestamp – Whether to remove timestamp from NumPy array file name (Use False if no timestamp)

Returns:

Loaded NumPy array if load is True otherwise the file path on the local machine where NumPy array was copied to. Returns None if failed to copy/load NumPy array

ccatkidlib.io.get_connection(ip: str, ssh_key: str) → Connection

Create Fabric Connection object to the RFSoC board with the specified ip address

Important

This function requires a SSH key pair to exist between the local machine and the RFSoC board for passwordless login

Parameters:

  • ip – IP address of RFSoC board

  • ssh_key – File path of private SSH key corresponding to public key on the RFSoC board

Returns:

Fabric Connection object to specified RFSoC board

ccatkidlib.io.get_creation_time(path: str) → float

Get the creation time of a file.

Note

The time reported is the ctime of the file, which usually corresponds to the creation time of the file. The ctime can, however, change with certain file modifications so care must be taken when interpreting the time reported.

Parameters:

path – Path of the file to get creation time of

Returns:

Creation time of file or -1 if creation time could not be determined

ccatkidlib.io.get_most_recent_file(dir: str, file_identifier: str | list[str] = '*', time_past: float = 3600, time_ref: float | None = None, ccatkidlib_file: bool = False) → str

Fetch the most recent file in the dir directory

Parameters:

  • dir – Directory from which to get most recent file

  • file_identifier – List of sub-strings to use for identifying/filtering files. A file will be identified as valid if it contains any of the sub-strings in the list.

  • time_past – How old the file can be in seconds. Files older than time_past will be ignored.

  • time_ref – Unix time to reference creation time of files against for determining if file is too old/new.

  • ccatkidlib_file – Whether or not the files in dir are ccatkidlib data files. If True, will use the timestamp in the file name as the creation time.

Returns:

File path of the most recent file if a vaild file is found, otherwise returns “invalid/path”

ccatkidlib.io.get_most_recent_file_board(c: Connection, dir: str, file_identifier: str = '*', time_past: float = 3600) → str

Get most recent file in specified dir directory on RFSoC board

Parameters:

  • c – Fabric Connection object of RFSoC board

  • dir – Directory from which to get most recent file from

  • file_identifier – Sub-string used to identify/filter files

  • time_past – How old the file can be in seconds. Files older than time_past will be ignored.

Returns:

File path of the most recent file if valid file exists otherwise returns “invalid/path”

ccatkidlib.io.get_timestamp(path: str) → int

Extract the timestamp from a ccatkidlib data file name.

Parameters:

path – Path of the ccatkidlib data file

Returns:

Timestamp of the file or -1 if no valid timestamp can be determined

ccatkidlib.io.increment_file(dir: str, file_prefix: str, file_suffix: str, overwrite: bool = False) → tuple[str, int | None]

Increment the file count in a file name. Used when saving files to prevent overwriting files with duplicate names.

Parameters:

  • dir – Directory where file will be saved

  • file_prefix – Sub-string of the file name before the file count

  • file_suffix – Sub-string of the file name after the file count

  • overwrite – Whether overwriting files is allowed. Will not add file count to file name if True

Returns:

The new file name with file count included and the file count of the file. If overwrite is True, will return None instead of the file count

ccatkidlib.io.load_config(cfg_path: str) → dict | list[dict]

Load configuration file from specified cfg_path

Parameters:

cfg_path – File path of configuration file to load

Returns:

Loaded configuration file or list of loaded configuration files if cfg_path contains more than one

ccatkidlib.io.path_exists(c: Connection, path: str) → bool

Check if specified path exists on RFSoC board.

Parameters:

  • c – Fabric Connection object of RFSoC board

  • path – Path to check existence of

Returns:

True if path exists, otherwise False

ccatkidlib.io.save_array_board(ip: str, ssh_key: str, path: str, array: ndarray, tmp_dir: str) → str | None

Save NumPy array to RFSoC board

Parameters:

  • ip – IP address of RFSoC board

  • ssh_key – File path of private SSH key corresponding to public key on the RFSoC board

  • path – File path where NumPy array should be saved on RFSoC board

  • array – NumPy array to save to RFSoC board

  • tmp_dir – Temporary directory where NumPy array is saved on local machine before copying to RFSoC board

Returns:

Returns output string of rsync command if successful, otherwise returns None

ccatkidlib.io.save_config(cfg_path: str, cfg: dict, save: bool = True) → dict

Save cfg configuration file to the specified cfg_path

Parameters:

  • cfg_path – File path where the configuration file should be saved

  • cfg – Configuration file to save

  • save – Whether to save configuration file

Returns:

Configuration file that was saved

ccatkidlib.log module

Library for standardized console and terminal logging across module

class ccatkidlib.log.Style(ITALICS: str = '\x1b[3m', INVERT: str = '\x1b[7m', PCS: str = '\x1b[92m', DEBUG: str = '\x1b[34m', INFO: str = '\x1b[32m', HEADER: str = '\x1b[48;5;177m', FOOTER: str = '\x1b[48;5;177m', WARNING: str = '\x1b[93m', ERROR: str = '\x1b[31m', CRITICAL: str = '\x1b[91m', DEFAULT: str = '\x1b[0m', WAIT: str = '\x1b[94;7m', TIMER: str = '\x1b[96;7m', LONGEST_DESC: int = 8)

Bases: object

CRITICAL: str = '\x1b[91m'
DEBUG: str = '\x1b[34m'
DEFAULT: str = '\x1b[0m'
ERROR: str = '\x1b[31m'
FOOTER: str = '\x1b[48;5;177m'
HEADER: str = '\x1b[48;5;177m'
INFO: str = '\x1b[32m'
INVERT: str = '\x1b[7m'
ITALICS: str = '\x1b[3m'
LONGEST_DESC: int = 8
PCS: str = '\x1b[92m'
TIMER: str = '\x1b[96;7m'
WAIT: str = '\x1b[94;7m'
WARNING: str = '\x1b[93m'
style_level() → str

Apply style to specified log level

Parameters:

level – Log level to applying styling to

Returns:

String with styling applied to log level

style_name() → str

Apply italics style to specified function/method name

Parameters:

name – Function/method name to apply styling to

Returns:

String with styling applied to function/method name

ccatkidlib.log.function_timer(func: Callable) → Any

Decorator that logs function execution time

Parameters:

func – Function to decorate

Returns:

The return of func

ccatkidlib.log.header(method: Callable) → Any

Decorator that logs HEADER and FOOTER messages and implements error handling for the wrapped method

Parameters:

method – Method to decorate

Returns:

The return of method if executed successfully otherwise None

ccatkidlib.log.log(level: Literal['PCS', 'DEBUG', 'TIMER', 'INFO', 'HEADER', 'WARNING', 'ERROR', 'CRITICAL'], msg: str, *args, name: str = 'ccatkidlib.log') → None

Log message to logger of the specified name with custom formatting

Parameters:

  • level – Level of message at which to log

  • msg – Message to log

  • args – Positional arguments to pass to logger.log method for print-f style message formatting

  • name – Name of logger

ccatkidlib.log.method_timer(method: Callable) → Any

Decorator that logs method execution time

Parameters:

method – Method to decorate

Returns:

The return of method

ccatkidlib.log.setup_logging(log_path: str, file_level: str, terminal_level: str, max_file_size: int = 100, name: str = 'ccatkidlib.log') → None

Setup levels and handlers for logger of the specified name

Parameters:

  • log_path – Path of file that logger should log to

  • file_level – Level at which to log messages to log file

  • terminal_level – Level at which to log messages to terminal

  • max_file_size – Maximum size of log file in Megabytes. If exceeded, a new log file will be created.

  • name – Name of logger

ccatkidlib.log.wait(t: float, desc: str = '', interval: float = 0.01) → None

Sleep for t seconds with a tqdm progress bar. This is done by monitoring the difference between the current and start time every interval seconds.

Parameters:

  • t – Number of seconds to sleep

  • desc – Description to add to progress bar

  • interval – Number of seconds to sleep before checking the difference between the current and start time

ccatkidlib.utils module

General utility functions

ccatkidlib.utils.dict_get(dic: dict, keys: str | list[str]) → Any | None

Fetch value from a nested dictionary corresponding to the last key specified in keys. An attempt will be made to reduce the dictionary using the keys in order until the final key is found or a KeyError is encountered. If a KeyError is encountered, the reduced dictionary will be recursively searched for the final key.

Parameters:

  • dic – Dictionary to get value from

  • keys – List of dictionary keys (in dictionary nesting order)

Returns:

Value of dic corresponding to the final key in keys or None if final key is not found

ccatkidlib.utils.dict_set(dic: dict, keys: str | list[str], value: Any) → bool

Set value of the last key specified in keys in the specified dictionary. An attempt will be made to reduce the dictionary using the keys in order until the final key is found or a KeyError is encountered. If a KeyError is encountered, the reduced dictionary will be recursively searched for the final key.

Parameters:

  • dic – Dictionary to set value in

  • keys – List of dictionary keys (in dictionary nesting order)

  • value – Value to set

Returns:

True if final key in keys was successfully set with the specified value otherwise False

Module contents