os_helper.misc_utils module

Miscellaneous Utilities

A grab-bag, on purpose: each function here answers one small, common question a script needs (what time is it, in a filename-safe format; how big is 12345678 bytes for a human; is this URL actually reachable right now) that is too small to earn its own module, but too common to leave to every caller re-deriving it. Grouped by the question each one answers:

timestamp formatting (now_string)
human-readable byte sizes (format_size)
folder content description with optional HTML / JSON dump (folder_description)
URL liveness checks (is_working_url)
folder zipping (zip_folder)
duration <-> string conversion (time2str, str2time)
simple file downloads (download_file)
public IP lookup (get_user_ip)

Usage example

>>> import os_helper as osh
>>> osh.format_size(12345678)
'12.35 MB'

Author

Warith HARCHAOUI, https://linkedin.com/in/warith-harchaoui

os_helper.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.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.misc_utils.unzip_folder(zip_file_path, folder_path='', *, overwrite=False)[source]

Extract a .zip archive into a folder — the inverse of zip_folder().

Entries that would land outside folder_path are refused rather than written. zipfile.extractall happily honours a member named ../../etc/thing or an absolute path, so an archive from elsewhere can overwrite files anywhere the process can write. Since the point of this function is to unpack archives fetched over the network, that check is the reason it exists rather than a one-line call at each site. Symlink members are refused for the same reason: they are a second way out of the folder.

Parameters:

zip_file_path (str) – The archive to extract (must exist).
folder_path (str, optional) – Destination folder, created if missing. Defaults to the archive path without its .zip suffix.
overwrite (bool, optional) – When False (default), an existing non-empty destination is left untouched and returned as-is, which makes the call idempotent for a download-once-then-reuse cache. True extracts over it.

Returns:

The destination folder.

Return type:

str

Raises:

AssertionError – If the archive does not exist.
ValueError – If a member would escape folder_path, or is a symlink.

os_helper.misc_utils.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