os_helper package

Submodules

Module contents

OS Helper

A collection of cross-platform utility functions covering: - file and directory handling - system / OS detection and process helpers - string manipulation and ASCII normalization - temporary files and folders (including remote staging) - hashing of strings, files, and folders - configuration loading (JSON / YAML / .env / environment) - URL checks, downloads, zipping - logging and verbosity helpers

Usage Example

>>> import os_helper as osh
>>> osh.verbosity(1)
>>> if osh.unix():
...     osh.info("running on a Unix-based OS")
>>> osh.info(osh.now_string("filename"))
>>> h = osh.hash_string("hello", size=8)

The same helpers are also exposed as an argparse CLI (os-helper) and a click CLI (os-helper-click, install [cli] extra). See the README and GUI.md for the multi-surface story.

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

os_helper.absolute2relative_path(path, base_path=None)[source]

Convert a path to a relative path expressed from base_path.

Parameters:

path (str) – The path to convert (absolute or relative).
base_path (str, optional) – Reference path. Defaults to the current working directory.

Returns:

The relative path from base_path to path.

Return type:

str

Example

>>> absolute2relative_path("/home/user/project/file.txt", "/home/user")
'project/file.txt'

os_helper.asciistring(input_string, replacement_char='-', lower=True, allow_digits=True)[source]

Convert a given string into a “safe” ASCII string by replacing accented and non-ASCII characters.

Non-ASCII characters that cannot be converted will be replaced with a specified character.

Parameters:

input_string (str) – The input string to be converted.
replacement_char (str, optional) – The character to replace non-ASCII or unwanted characters with. Defaults to ‘-‘.
lower (bool, optional) – Whether to convert the string to lowercase. Defaults to True.
allow_digits (bool, optional) – Whether to allow digits in the resulting string. Defaults to True.

Returns:

A “safe” ASCII string with unwanted characters replaced and case adjusted.

Return type:

str

Examples

>>> asciistring("MyFile@2024.txt")
'myfile-2024-txt'

>>> asciistring("Café-Con-Leche!", replacement_char="_")
'cafe_con_leche'

>>> asciistring("Special#File$2024", lower=False)
'Special-File-2024'

os_helper.check(condition, msg='Assertion failed')[source]

Assert a condition, logging an error and raising if it fails.

Parameters:

condition (bool) – The predicate that must hold.
msg (str, optional) – Message logged and attached to the raised error on failure.

Raises:

AssertionError – If condition is falsy.

Return type:

None

os_helper.checkfile(filepath, msg='', check_empty=False)[source]

Assert that a file exists and, optionally, that it is non-empty.

Parameters:

filepath (str) – The path to the file.
msg (str, optional) – Prefix added to the assertion message on failure.
check_empty (bool, optional) – If True, also asserts that the file size is greater than zero.

Raises:

AssertionError – If the file does not exist (or is empty when check_empty is True).

Return type:

None

Example

>>> checkfile("data.csv", msg="Data file missing", check_empty=True)

os_helper.copyfile(source, destination)[source]

Copy a file from source to destination, preserving metadata.

If destination is an existing directory, the source file name is appended to it (mirroring cp semantics).

Parameters:

source (str) – The path to the source file (must exist and be non-empty).
destination (str) – The path to the destination file or existing directory.

Raises:

AssertionError – If the source file does not exist or is empty, or if source and destination resolve to the same path.
OSError – If the underlying shutil.copy2 call fails (propagated as-is).

Return type:

None

Example

>>> copyfile("source.txt", "backup/source_backup.txt")

os_helper.cpu_timer()[source]

Measure CPU time consumed by the current process using time.process_time() (sums user + system CPU across all threads).

Differs from wall_timer() in two important ways:

It excludes time spent blocked on I/O, sleeping, or waiting on the GPU — so it isolates “actual computation done by Python+native code”.
It excludes subprocesses (ffmpeg, etc.). For those, use wall_timer() or os.times() directly.

On a multi-threaded computation it can report more seconds than wall-clock — that’s intentional: it counts the CPU work, not the elapsed time.

Yields:: dict – {"seconds": float, "milliseconds": float}.
Return type:: Generator[dict[str, float], None, None]

Examples

>>> with cpu_timer() as t:
...     total = sum(i * i for i in range(1_000_000))
>>> assert t["seconds"] > 0

os_helper.critical(msg, *args, **kwargs)[source]

Log a message at CRITICAL level via the dedicated os_helper logger.

Parameters:

msg (str) – The (possibly %-style) message template.
*args (Any) – Positional interpolation arguments forwarded to logging.critical.
**kwargs (Any) – Keyword options forwarded to logging.critical.

Return type:

None

os_helper.debug(msg, *args, **kwargs)[source]

Log a message at DEBUG level via the dedicated os_helper logger.

Parameters:

msg (str) – The (possibly %-style) message template.
*args (Any) – Positional interpolation arguments forwarded to logging.debug.
**kwargs (Any) – Keyword options (e.g. exc_info) forwarded to logging.debug.

Return type:

None

os_helper.dir_exists(path, check_empty=False)[source]

Check if a directory exists, with an option to verify it’s not empty.

Parameters:

path (str) – The path to the directory.
check_empty (bool, optional) – If True, also checks that the directory is not empty (excluding hidden files). Defaults to False.

Returns:

True if the directory exists (and is not empty if check_empty is True), False otherwise.

Return type:

bool

Example

>>> dir_exists("/path/to/folder")
True
>>> dir_exists("/path/to/empty_folder", check_empty=True)
False

os_helper.download_file(url, file_path='', *, chunk_size=None, progress=True, check_url=True, resume=True, retries=3, sha256=None, overwrite=False)[source]

Download a URL to a local file: streamed, resumable, retried, atomic, verified.

This is the suite-wide download primitive; every helper that fetches bytes from HTTP (model mirrors, media, templates, catalogs) routes through it so the smart behaviour lives in one place:

Streamed block-by-block, so the payload is never held in memory (a multi-GB model downloads with a flat footprint). The block size adapts to the payload via _adaptive_chunk_size() (~500 progress updates, clamped [64 KiB, 4 MiB]); pass chunk_size to override.
Resumable (resume=True): bytes land in a <file>.part sidecar, and an interrupted transfer continues with an HTTP Range request instead of restarting. If the server ignores Range (answers 200 instead of 206), the sidecar is restarted from zero automatically.
Retried (retries attempts) on transient network errors, with exponential backoff, each retry resuming from the current sidecar size.
Atomic: the finished sidecar is os.replace-d onto file_path in one step, so file_path only ever exists as a complete download; a crash leaves a .part, never a truncated final file.
Verified (optional sha256): the finished file’s digest is checked and a mismatch raises (the bad sidecar is discarded).
Idempotent (overwrite=False): an existing, complete file_path is reused without re-downloading (its digest re-checked when sha256 is given), so callers can invoke it unconditionally.

A tqdm progress bar is shown on an interactive terminal (auto-suppressed off-TTY). Progress and completion are logged via info(), retries via warning(), hard failures via error(). If file_path is empty, the name is derived from the URL’s last segment.

Parameters:

url (str) – The URL to download from.
file_path (str, optional) – Destination path. Defaults to the URL’s last path segment.
chunk_size (int or None, optional) – Streaming block size in bytes. None (default) adapts to the size.
progress (bool, optional) – Show a progress bar on an interactive terminal (default True).
check_url (bool, optional) – Pre-validate the URL with a HEAD request (default True). Set False when the server rejects HEAD (405/403).
resume (bool, optional) – Continue a partial <file>.part via HTTP Range (default True). False discards any sidecar and downloads from scratch.
retries (int, optional) – Extra attempts on transient network errors, with exponential backoff (default 3, so up to four tries total).
sha256 (str or None, optional) – Expected lowercase hex SHA-256. When given, the finished file is verified and a mismatch raises ValueError.
overwrite (bool, optional) – Re-download even if file_path already exists (default False: reuse a complete destination, re-checking sha256 when supplied).

Returns:

{"path": <destination>, "content_type": <server MIME or "">, "bytes": <size on disk>, "sha256": <hex or "">, "resumed": <bool>}.

Return type:

dict of str to object

Raises:

AssertionError – If the URL fails the is_working_url() precondition.
ValueError – If sha256 is given and the finished file’s digest does not match.
requests.RequestException – If every attempt fails or the server returns a non-2xx status.

os_helper.emptystring(s)[source]

Return True if s is None, not a string, or only whitespace.

Convenient for input validation where "", None and " " should all be treated the same way.

Parameters:: s (Optional[str]) – The value to check.
Returns:: True when s is None or contains only whitespace; False otherwise.
Return type:: bool

Examples

>>> emptystring("")
True
>>> emptystring("   ")
True
>>> emptystring(None)
True
>>> emptystring("hello")
False

os_helper.error(msg, *args, **kwargs)[source]

Log a message at ERROR level via the dedicated os_helper logger.

Note: this is a non-raising logger call. It will not terminate the program. Use check() (assertion-style) or raise an exception explicitly if you need failure semantics.

Parameters:

msg (str) – The (possibly %-style) message template.
*args (Any) – Positional interpolation arguments forwarded to logging.error.
**kwargs (Any) – Keyword options forwarded to logging.error.

Return type:

None

os_helper.file_exists(file_path, check_empty=False)[source]

Check if a file exists, with an option to verify it’s not empty.

Parameters:

file_path (str) – The path to the file.
check_empty (bool, optional) – If True, also checks that the file is not empty. Defaults to False.

Returns:

True if the file exists (and is not empty if check_empty is True), False otherwise.

Return type:

bool

Example

>>> file_exists("example.txt")
True
>>> file_exists("empty.txt", check_empty=True)
False

os_helper.folder_description(path, recursive=True, index_html=True, with_size=True, description_json=True)[source]

List a folder’s files (with sizes) and optionally emit an HTML / JSON index.

Walks path (recursively by default), skips hidden entries (names starting with "."), and returns a mapping from each file’s path relative to path to its size in bytes. When index_html is True, a Bootstrap-styled index.html is written into path; when description_json is True, a description.json companion file is written as well.

Parameters:

path (str) – Path to the folder to describe (must exist).
recursive (bool, optional) – If True, descend into subdirectories. Defaults to True.
index_html (bool, optional) – If True, write path/index.html linking each entry. Defaults to True.
with_size (bool, optional) – If True, include a human-readable size column in the HTML index. Defaults to True.
description_json (bool, optional) – If True, write path/description.json with the raw mapping. Defaults to True.

Returns:

Mapping of relative_file_path -> size_in_bytes for every non-hidden file found.

Return type:

dict

Raises:

AssertionError – If path does not exist or is not a directory.

os_helper.folder_name_ext(path, checkpath=False)[source]

Decompose a file or folder path into (folder, basename, extension).

The split happens at the last dot in the basename, so multi-part suffixes like .tar.gz are not collapsed into one extension. Use "basename.extension" to recover the original file name.

Returns an empty extension for: - directories, - files whose basename contains no dot.

Parameters:

path (str) – The path to decompose. Resolved to an absolute path internally.
checkpath (bool, optional) – If True, asserts that the path exists on disk.

Returns:

(folder, basename, extension) where extension excludes the leading dot.

Return type:

tuple of (str, str, str)

Examples

>>> folder_name_ext("/path/to/file.txt")
('/path/to', 'file', 'txt')
>>> folder_name_ext("/path/to/archive.tar.gz")
('/path/to', 'archive.tar', 'gz')
>>> folder_name_ext("/path/to/folder")          # existing directory
('/path/to', 'folder', '')

os_helper.format_size(size)[source]

Convert a byte count into a short human-readable string.

Picks an appropriate SI-style unit (B, KB, MB, GB, TB) using decimal thresholds (1000-based). Useful for log lines and progress reporting.

Parameters:: size (int) – Size in bytes.
Returns:: Formatted size, e.g. "1.23 MB", "456.00 KB", "42 B".
Return type:: str

os_helper.get_config(keys, config_type, path=None, env_files=None)[source]

Load configuration settings using a fixed fallback order.

Precedence (first match wins): 1. path pointing to a JSON/YAML file (or, if a directory, the first

.json, .yaml or .yml file in it that contains all keys);

one or more .env files merged into os.environ;
the current process environment.

Parameters:

keys (List[str]) – Keys required to be present in the resolved configuration.
config_type (str) – Human-readable label used only in log messages.
path (Optional[str], optional) – Path to a configuration file or a directory containing one. If None or empty, this step is skipped.
env_files (Optional[List[str]], optional) – .env files to load into os.environ before reading variables. Defaults to [".env"].

Returns:

Mapping with one entry per requested key.

Return type:

Dict[str, Union[str, int, float]]

Raises:

RuntimeError – If none of the sources provide all required keys.

Example

>>> config = get_config(["host", "port"], "database", path="config.yaml")
>>> config
{'host': 'localhost', 'port': 5432}

os_helper.get_nb_workers(workers=-1)[source]

Resolve a worker count, following scikit-learn’s n_jobs convention.

The default pool size is os.cpu_count() (or 1 if that returns None), overridable via the NB_WORKERS environment variable.

Parameters:

workers (int, optional) –

0 : use the full pool size.
> 0: use exactly that many workers.
< 0: use pool_size + workers + 1 (e.g. -1 → all CPUs, -2 → all but one), clamped to at least 1.

Returns:

The resolved worker count (always >= 1).

Return type:

int

Example

>>> get_nb_workers()  # -1 → all available CPU cores
4

os_helper.get_user_ip()[source]

Fetch the caller’s public IPv4 and IPv6 addresses via the ipify API.

Returns:: {"ipv4": <str or None>, "ipv6": <str or None>}. An individual entry is None when its endpoint fails or returns no address.
Return type:: dict
Raises:: AssertionError – If both endpoints fail and no address could be retrieved.

os_helper.getpid()[source]

Return the current process ID as a string.

Returns:: str(os.getpid()).
Return type:: str

os_helper.gpu_timer(backend='auto')[source]

Measure GPU execution time, synchronizing before and after the block.

Backends

"cuda" — uses torch.cuda.Event(enable_timing=True) pairs, which give microsecond-level GPU-side timing.
"mps" — Apple Silicon. PyTorch’s MPS backend does not expose timing events, so this falls back to torch.mps.synchronize() + time.perf_counter() around the block. Accuracy ~1 ms.
"auto" — pick CUDA if available, else MPS, else raise.

Both paths synchronize before and after the block so the measured duration corresponds to actual GPU work, not just kernel-queue submission. Without synchronization, GPU ops are asynchronous and the timer would understate the cost dramatically.

param backend:: "auto" (default), "cuda", or "mps".
type backend:: str, optional
Yields:: dict – {"seconds": float, "milliseconds": float}.
raises RuntimeError:: If PyTorch is not installed, or if the requested backend is unavailable on this machine.
raises ValueError:: If backend is not one of "auto", "cuda", "mps".

Examples

>>> import torch
>>> if torch.cuda.is_available():
...     x = torch.randn(2048, 2048, device="cuda")
...     with gpu_timer() as t:
...         y = x @ x
...     print(t["milliseconds"])

Parameters:: backend (str)
Return type:: Generator[dict[str, float], None, None]

os_helper.hash_string(s, size=-1)[source]

Generate a hash of a given string and optionally returns a truncated version.

Parameters:

s (str) – The input string to hash.
size (int, optional) – If positive, truncates the hash to the specified length. Defaults to -1 (no truncation).

Returns:

The hashed string, optionally truncated.

Return type:

str

Example

>>> isinstance(hash_string("example"), str)
True
>>> len(hash_string("example"))
40
>>> len(hash_string("example", size=8))
8

Note

The exact digest depends on the underlying hash engine (RIPEMD-160 when available, BLAKE2b truncated to 20 bytes otherwise). The output length stays 40 hex characters either way.

os_helper.hashfile(path, hash_content=True, date=False)[source]

Generate a hash for a file’s content and/or its last modification date.

Parameters:

path (str) – The path to the file to hash.
hash_content (bool, optional) – If True, includes the file’s content in the hash (default: True).
date (bool, optional) – If True, includes the current date in the hash (default: False).

Returns:

The resulting hash of the file as a 40-character hex string.

Return type:

str

os_helper.hashfolder(path, hash_content=True, hash_path=False, date=False)[source]

Generate a hash for the contents of a folder and/or its path.

Parameters:

path (str) – The path to the folder to hash.
hash_content (bool, optional) – If True, includes the folder’s contents in the hash (default: True).
hash_path (bool, optional) – If True, includes the folder’s path in the hash (default: False).
date (bool, optional) – If True, includes the current date in the hash (default: False).

Returns:

The resulting hash of the folder and/or its contents as a 40-character hex string.

Return type:

str

os_helper.info(msg, *args, **kwargs)[source]

Log a message at INFO level via the dedicated os_helper logger.

Parameters:

msg (str) – The (possibly %-style) message template.
*args (Any) – Positional interpolation arguments forwarded to logging.info.
**kwargs (Any) – Keyword options forwarded to logging.info.

Return type:

None

os_helper.init_logging(*, level=20, stdout=True, log_format='%(asctime)s | %(levelname)s | %(name)s | %(message)s', date_format='%Y-%m-%d %H:%M:%S', filename=None, capture_warnings=True, reset=True, use_colors=True, propagate=False, name=None, live_stream=False)[source]

Initialize application-wide logging.

This function configures the root logger with a console handler and, optionally, a file handler. It is designed for applications, scripts, notebooks, and machine learning experiments where deterministic logging setup is useful.

Parameters:

level (int, optional) – Logging level applied to the root logger and its handlers. Typical values are logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, and logging.CRITICAL.
stdout (bool, optional) – If True, console logs are sent to sys.stdout. Otherwise, they are sent to sys.stderr.
log_format (str, optional) – Format string used for log records.
date_format (str, optional) – Format string used for timestamps in log records.
filename (str | pathlib.Path | None, optional) – Optional path to a log file. If provided, logs are also written to this file using UTF-8 encoding.
capture_warnings (bool, optional) – If True, warnings emitted through the warnings module are redirected to the logging system.
reset (bool, optional) – If True, existing handlers attached to the root logger are removed before adding new ones. This is often desirable in notebooks and interactive sessions to avoid duplicated messages.
use_colors (bool, optional) – If True, colorize console log levels when ANSI colors are supported. File logs are never colorized.
propagate (bool, optional) – Value assigned to the target logger’s propagation flag. False avoids duplicates for the root logger; a named logger (see name) often wants True so its records still reach a host’s / pytest’s root handlers (e.g. caplog).
name (str | None, optional) – Configure this named logger instead of the root. When set, the reset step only removes handlers this function installed (so a host’s / pytest’s handlers on that logger survive), and repeated calls are idempotent — a second call does not stack a duplicate console handler. This is the CLI-friendly mode: configure "mytool" once, keep propagate=True, and every logging.getLogger("mytool.*") inherits the handler + level.
live_stream (bool, optional) – If True, the console handler re-resolves sys.stdout/sys.stderr on every emit instead of binding the stream once. This keeps output flowing to wherever the stream currently points — surviving pytest’s capsys (which swaps the streams per test) and any post-config redirection. The default keeps the classic bound-stream handler.

Returns:

The configured logger (the root logger, or the one named by name).

Return type:

logging.Logger

Notes

This function is intended for top-level applications, notebooks, and experimentation code. Reusable libraries should generally avoid configuring global logging and should instead use:

logger = logging.getLogger(__name__)

Examples

>>> logger = init_logging(level=logging.DEBUG, filename="experiment.log")
>>> logger.info("Logging is configured.")

>>> logger = init_logging(use_colors=True, reset=True)
>>> logger.warning("This is a warning.")

os_helper.is_working_url(url)[source]

Return True if url is syntactically valid and answers 200 to a HEAD.

Parameters:: url (str) – The URL to check.
Returns:: True only when validation succeeds and the HEAD request returns HTTP 200 within five seconds. Network errors and non-200 responses both yield False.
Return type:: bool

os_helper.join(*args)[source]

Join multiple path components into a single absolute, normalized path.

This is the canonical path-construction helper exposed by os_helper and replaces the older os_path_constructor function (removed in v1.1.0). It accepts either positional components or a single iterable.

Parameters:: *args (str) – The path components to join, or a single iterable of path components.
Returns:: The absolute, normalized path.
Return type:: str

Example

>>> join("folder1", "subfolder2", "file.txt")
'/home/user/project/folder1/subfolder2/file.txt'
>>> join(["folder1", "subfolder2", "file.txt"])
'/home/user/project/folder1/subfolder2/file.txt'

os_helper.linux()[source]

Determine if the current operating system is Linux.

Returns:: True if the operating system is Linux, False otherwise.
Return type:: bool

os_helper.macos()[source]

Determine if the current operating system is macOS.

Returns:: True if the operating system is macOS, False otherwise.
Return type:: bool

os_helper.make_directory(folder_path, exist_ok=True)[source]

Create a directory (and missing parents), optionally tolerating prior existence.

Parameters:

folder_path (str) – The path to the directory to create.
exist_ok (bool, optional) – If True (the default), succeed silently when the directory already exists. If False, FileExistsError is raised in that case.

Raises:

OSError – If the directory cannot be created (propagated from os.makedirs).
AssertionError – If os.makedirs returned without error but the directory is still not visible on disk afterwards.

Return type:

None

Example

>>> make_directory("/path/to/new_folder")

os_helper.make_temporary_directory(prefix='', directory=None)[source]

Create a temporary directory and return its path — caller owns cleanup.

The non-context-manager companion to temporary_folder(). Use this when the directory must outlive a with block: a request handler that schedules deletion after streaming a response, a process-lifetime scratch dir cleaned at exit, or any case where the created path is stored and removed later. It is the suite’s tempfile.mkdtemp() — the caller is responsible for removing the tree (e.g. via remove_directory() / shutil.rmtree / atexit).

Parameters:

prefix (str, optional) – Prefix for the directory name, aiding recognition in /tmp.
directory (str or None, optional) – Parent directory to create it in (default: the system temp location).

Returns:

Absolute path to the freshly created directory.

Return type:

str

Example

>>> work = make_temporary_directory(prefix="myjob-")
>>> # ... use `work`, then clean up when you are done ...
>>> remove_directory(work)

os_helper.now_string(fmt='log')[source]

Get the current timestamp as a formatted string.

Parameters:: fmt (str, optional) – The format of the timestamp. “log” (default) => YYYY/MM/DD-HH:MM:SS “filename” => YYYY-MM-DD-HH-MM-SS (file-system safe)
Returns:: The formatted date-time string.
Return type:: str

os_helper.openfile(filename)[source]

Open a file in the platform’s default application.

Uses os.startfile on Windows, open on macOS, and xdg-open on Linux. Exceptions from the underlying call are propagated as-is so the caller can react to them.

Parameters:: filename (str) – The path to the file to open.
Raises:: OSError – If the platform is unsupported or the underlying open call fails.
Return type:: None

os_helper.path_without_home(path)[source]

Convert an absolute path to be relative to the user’s home directory by replacing the home path with ‘~’.

Parameters:: path (str) – The absolute path to convert.
Returns:: The path with the home directory replaced by ‘~’, if applicable.
Return type:: str

Example

>>> path_without_home("/home/user/project/file.txt")
'~/project/file.txt'

os_helper.progress_bar(total=None, *, desc='', disable=None, unit='B')[source]

Create a tqdm progress bar configured for file transfers.

The suite-wide byte-transfer bar, so every helper that moves bytes (HTTP download, S3 up / download, SFTP put / get) shows the same progress UI: byte-scaled units (KiB / MiB / GiB), a known total when available (for a percentage + ETA), and — crucially — auto-suppression when ``stderr`` is not a TTY, so CI logs and piped output are never flooded with control characters. Each caller simply wraps its transfer library’s progress hook (requests chunks, boto3 Callback, paramiko callback) around one of these bars.

Parameters:

total (int or None, optional) – Total number of bytes when known (enables percentage + ETA); None leaves the bar open-ended.
desc (str, optional) – Short label shown at the left of the bar (e.g. the file / key name).
disable (bool or None, optional) – Force the bar on / off. None (default) auto-disables when stderr is not an interactive terminal.
unit (str, optional) – Progress unit (default "B"; unit_scale renders raw byte counts as KiB / MiB / GiB).

Returns:

A ready progress bar; call .update(n) per transferred chunk and .close() when done.

Return type:

tqdm.tqdm

Examples

>>> bar = progress_bar(total=10, desc="demo", disable=True)
>>> bar.total
10
>>> bar.disable
True
>>> bar.close()

os_helper.recursive_glob(root_dir, pattern)[source]

Recursively search for files matching a glob pattern under root_dir.

Each subdirectory is walked and the pattern applied in turn, so patterns like "*.txt" match in nested folders as well as at the top level.

Parameters:

root_dir (str) – The root directory to start searching from.
pattern (str) – The glob pattern to match against file names (e.g., "*.txt").

Returns:

File paths matching the pattern, in walk order.

Return type:

List[str]

Example

>>> recursive_glob("/home/user", "*.txt")
['/home/user/file1.txt', '/home/user/docs/file2.txt']

os_helper.relative2absolute_path(path, checkpath=False)[source]

Convert a relative path to an absolute path.

Parameters:

path (str) – The relative or absolute path to convert.
checkpath (bool, optional) – If True, verifies that the resulting absolute path exists (as a file or directory). Defaults to False.

Returns:

The absolute path.

Return type:

str

Raises:

FileNotFoundError – If checkpath is True and the path does not exist.

Example

>>> relative2absolute_path("docs/readme.md")
'/home/user/project/docs/readme.md'

os_helper.remove_directory(folder_path)[source]

Remove a directory and all its contents.

A missing directory is treated as a no-op (logged at INFO); any other failure from shutil.rmtree is propagated to the caller.

Parameters:: folder_path (str) – The path to the directory to remove.
Raises:: OSError – If the directory exists but cannot be removed (propagated from shutil.rmtree).
Return type:: None

Example

>>> remove_directory("/path/to/temp_folder")

os_helper.remove_files(files_list)[source]

Remove a list of files on a best-effort basis.

Missing entries are skipped and individual removal failures are logged at ERROR level without aborting the rest of the batch. This function does not raise — if you need hard-fail-on-first semantics, call pathlib.Path(p).unlink() yourself.

Parameters:: files_list (List[str]) – A list of file paths to remove.
Return type:: None

Example

>>> remove_files(["temp1.txt", "temp2.log"])

os_helper.size_file(filepath)[source]

Get the size of a file in bytes.

Parameters:: filepath (str) – The path to the file.
Returns:: The size of the file in bytes, or -1 if the file does not exist.
Return type:: int

Example

>>> size_file("example.txt")
1024

os_helper.str2time(input_string)[source]

Parse a time string into seconds.

Parameters:: input_string (str) – A string representing time.
Returns:: Number of seconds parsed from the string.
Return type:: float

Examples

>>> str2time("1:30:00")
5400.0
>>> str2time("1 hr 30 min")
5400.0
>>> str2time("120 s")
120.0
>>> str2time("1.5 hours")
5400.0
>>> str2time("1.5 days")
129600.0

os_helper.system(cmd, expected_output='', check_exitcode=True, check_empty=False)[source]

Run a shell-style command via subprocess and capture its output.

The command string is parsed with shlex.split() and executed without spawning an actual shell (shell=False), which avoids shell-injection pitfalls while still accepting a familiar command string.

Parameters:

cmd (str) – Command line to execute (e.g., "ffmpeg -i in.mp4 out.mp3").
expected_output (str, optional) – If non-empty, a file or directory path expected to be present once the command completes successfully.
check_exitcode (bool, optional) – If True, assert that the process exit code is 0.
check_empty (bool, optional) – If True, also assert that expected_output is non-empty (file size > 0 / directory not empty).

Returns:

{"out": <stdout as str>, "err": <stderr as str>}.

Return type:

dict

Raises:

AssertionError – If the exit code check or the expected-output check fails.

os_helper.temporary_filename(suffix='', mode='wt', prefix='', delete=True, directory=None)[source]

Create a temporary file with a unique name that persists even after closing.

This context manager generates a temporary file with a unique name, which is optionally removed after use. It ensures that temporary files are managed safely and are cleaned up to prevent clutter or security issues.

Parameters:

suffix (str, optional) – File suffix (e.g., “.txt”). Defaults to “”.
mode (str, optional) – Mode in which the file is opened (e.g., “wt” for writing text). Defaults to “wt”.
prefix (str, optional) – Prefix for the file name. Defaults to “”.
delete (bool, optional) – Whether to delete the file after exiting the context. Defaults to True.
directory (str or None, optional) – Parent directory to create the file in (default: the system temp dir). Use this when the temp file must sit next to other inputs — e.g. so a tool that resolves paths relative to the file still finds its siblings.

Yields:

str – The name of the temporary file.

Return type:

Generator[str, None, None]

Example

>>> with temporary_filename(suffix=".txt") as temp_file:
...     with open(temp_file, "wt") as fout:
...         fout.write("Temporary content")
...     # temp_file is automatically deleted after the block

os_helper.temporary_folder(prefix='', delete=True)[source]

Create a temporary directory with a unique name that persists during the context.

This context manager generates a temporary directory with a unique name, which is optionally removed after use. It ensures that temporary directories are managed safely and are cleaned up to prevent clutter or security issues.

Parameters:

prefix (str, optional) – Prefix for the folder name. Defaults to “”.
delete (bool, optional) – Whether to delete the directory and its contents after exiting the context. Defaults to True.

Yields:

str – The name of the temporary directory.

Return type:

Generator[str, None, None]

Example

>>> with temporary_folder(prefix="tempdir") as temp_dir:
...     # Use the temporary directory
...     with open(os.path.join(temp_dir, "file.txt"), "w") as f:
...         f.write("Temporary content")
...     # temp_dir and its contents are automatically deleted after the block

os_helper.temporary_remote_file(upload_function, delete_function, *, prefix='', suffix='', from_local_file=None, checkfile_function=None, mode='wb', initial_content=None)[source]

Context manager that uploads a file to a remote location and guarantees deletion of the remote artifact when the context exits.

Two modes:

If from_local_file is given, that existing local file is uploaded as-is and left untouched on cleanup; only the remote copy is deleted.
Otherwise, a uniquely named local temp file is created (optionally pre-populated with initial_content), uploaded, and removed locally when the context manager exits.

Useful for staging files to S3/GCS/SFTP/anywhere when you only need them briefly: pass in the upload + delete callables for your storage backend and the helper handles the lifecycle.

Parameters:

upload_function (Callable[[str], str]) – Called with a local file path; must return the remote path/URI.
delete_function (Callable[[str], None]) – Called with the remote path/URI to remove the remote artifact.
prefix (str, optional) – Prefix for the temporary file name (ignored when from_local_file is provided).
suffix (str, optional) – File extension for the temporary file (with or without leading “.”).
from_local_file (str, optional) – Path to an existing local file to upload instead of creating one.
checkfile_function (Callable[[str], bool], optional) – Optional post-upload sanity check; must return True for success.
mode (str, optional) – Open mode used when writing initial_content to the temp file. Defaults to "wb".
initial_content (bytes | str, optional) – Optional content written into the new temp file before upload. Type must match mode (bytes for "wb", str for "wt").

Yields:

str – The remote file path/URI returned by upload_function.

Raises:

TypeError – If any supplied callable is not actually callable.
FileNotFoundError – If from_local_file is given but the path does not exist.
RuntimeError – If checkfile_function is provided and returns False after upload.

Return type:

Generator[str, None, None]

Examples

>>> storage = {}
>>> def upload(p):
...     with open(p, "rb") as f: storage[p] = f.read()
...     return p
>>> def delete(r):
...     storage.pop(r, None)
>>> with temporary_remote_file(upload, delete, suffix=".bin",
...                            initial_content=b"hello") as remote:
...     assert storage[remote] == b"hello"
>>> # remote artifact is gone after the block

os_helper.tic()[source]

Start (or restart) the implicit global stopwatch.

Returns the start timestamp so callers can pin a handle for nested or interleaved measurements:

>>> t_outer = tic()
>>> # ... work ...
>>> t_inner = tic()      # implicit global now points at t_inner
>>> # ... more work ...
>>> toc(t_inner)         # explicit handle works regardless of which tic() was last
>>> toc(t_outer)

Returns:: time.perf_counter() snapshot, usable as a handle for toc().
Return type:: float

os_helper.time2str(seconds, no_space=False)[source]

Convert a float number of seconds to a readable string, e.g. “1 hr 2 min 5 sec”.

Parameters:

seconds (float) – Time in seconds.
no_space (bool, optional) – If True, removes spaces between number and unit. e.g. “1hr 2min 5sec”.

Returns:

A human-readable time string.

Return type:

str

Examples

>>> time2str(5400.0)
"1 hr 30 min"
>>> time2str(120.0)
"2 min"
>>> time2str(3661.0)
"1 hr 1 min 1 sec"

os_helper.toc(handle=None, *, log=False)[source]

Return seconds elapsed since the matching tic() call.

Parameters:

handle (float, optional) – Handle returned by tic(). If None, the implicit “last tic” timestamp is used.
log (bool, optional) – If True, log the elapsed time at INFO level via the root logger.

Returns:

Seconds elapsed (does not reset the timer — call tic() again to restart).

Return type:

float

Raises:

RuntimeError – If called with no handle and no prior tic().

Examples

>>> tic()
>>> # ... work ...
>>> elapsed = toc()

os_helper.unix()[source]

Determine if the current operating system is Unix-based (Linux or macOS).

Returns:: True if the operating system is Unix-based (Linux or macOS), False otherwise.
Return type:: bool

os_helper.verbosity(level=None)[source]

Get or set the current root logger verbosity.

Called with no argument, returns the current verbosity as an integer. Called with an integer, updates the root logger (and its existing handlers) accordingly and returns the new effective verbosity.

Mapping (higher = more verbose):

>= 2 → DEBUG
1 → INFO
0 → WARNING
-1 → ERROR
<= -2 → CRITICAL

Values outside [-2, 2] are clamped (e.g. verbosity(3) is treated as DEBUG, matching the convenience usage shown in the README).

Parameters:: level (int | None, optional) – New verbosity level to apply, or None to just read the current value.
Returns:: Current (post-update, when setting) verbosity as an integer in the range [-2, 2].
Return type:: int

Examples

>>> verbosity(2)   # turn on DEBUG-level logging
2
>>> verbosity()    # read current level
2

os_helper.wall_timer()[source]

Measure real elapsed wall-clock time using time.perf_counter().

Use this when you want to know “how long did this take to run from the user’s perspective” — it includes I/O, sleeps, GPU waits, and subprocess time.

Yields:: dict – {"seconds": float, "milliseconds": float}, both fields populated when the with block exits.
Return type:: Generator[dict[str, float], None, None]

Examples

>>> with wall_timer() as t:
...     time.sleep(0.05)
>>> assert t["seconds"] >= 0.05

os_helper.warning(msg, *args, **kwargs)[source]

Log a message at WARNING level via the dedicated os_helper logger.

Parameters:

msg (str) – The (possibly %-style) message template.
*args (Any) – Positional interpolation arguments forwarded to logging.warning.
**kwargs (Any) – Keyword options forwarded to logging.warning.

Return type:

None

os_helper.windows()[source]

Determine if the current operating system is Windows.

Returns:: True if the operating system is Windows, False otherwise.
Return type:: bool

os_helper.zip_folder(folder_path, zip_file_path='')[source]

Create a deflated .zip archive of a folder’s contents.

Hidden files (names beginning with .) are skipped. Paths inside the archive are stored relative to folder_path.

Parameters:

folder_path (str) – The path to the folder to zip (must exist).
zip_file_path (str, optional) – Destination archive path. Defaults to folder_path + ".zip".

Raises:

AssertionError – If folder_path does not exist.

Return type:

None