pdstools.utils.cdh_utils._io

I/O, logging, working-directory and version-check helpers.

Functions

process_files_to_bytes(→ tuple[bytes, str])

Processes a list of file paths, returning file content as bytes and a corresponding file name.

get_latest_pdstools_version()

setup_logger()

Return the pdstools logger and a log buffer it streams into.

create_working_and_temp_dir(→ tuple[pathlib.Path, ...)

Creates a working directory for saving files and a temp_dir

_read_pdc(pdc_data)

Module Contents

process_files_to_bytes(file_paths: list[str | pathlib.Path], base_file_name: str | pathlib.Path) → tuple[bytes, str]

Processes a list of file paths, returning file content as bytes and a corresponding file name. Useful for zipping muliple model reports and the byte object is used for downloading files in Streamlit app.

This function handles three scenarios: 1. Single file: Returns the file’s content as bytes and the provided base file name. 2. Multiple files: Creates a zip file containing all files, returns the zip file’s content as bytes

and a generated zip file name.

  1. No files: Returns empty bytes and an empty string.

Parameters:

  • file_paths (list[str | Path]) – A list of file paths to process. Can be empty, contain a single path, or multiple paths.

  • base_file_name (str | Path) – The base name to use for the output file. For a single file, this name is returned as is. For multiple files, this is used as part of the generated zip file name.

Returns:

A tuple containing: - bytes: The content of the single file or the created zip file, or empty bytes if no files. - str: The file name (either base_file_name or a generated zip file name), or an empty string if no files.

Return type:

tuple[bytes, str]

get_latest_pdstools_version()
setup_logger()

Return the pdstools logger and a log buffer it streams into.

Targets the named pdstools logger rather than the root logger so we don’t clobber the host application’s logging config (Streamlit, Quarto, Jupyter, etc.). Idempotent: repeated calls return the same buffer instead of stacking new handlers, so re-running a notebook cell or bouncing a Streamlit page doesn’t produce duplicated log lines.

create_working_and_temp_dir(name: str | None = None, working_dir: os.PathLike | None = None) → tuple[pathlib.Path, pathlib.Path]

Creates a working directory for saving files and a temp_dir

Parameters:
Return type:

tuple[pathlib.Path, pathlib.Path]

_read_pdc(pdc_data: polars.LazyFrame)
Parameters:

pdc_data (polars.LazyFrame)