soniclue-basekit 0.1.1

A foundational toolkit of pure Python utilities and helpers.

soniclue-basekit

A foundational toolkit of pure Python utilities and helpers.

Installation

pip install soniclue-basekit
# or
uv add soniclue-basekit

Modules

ExecTimer

Times a block of code using wall-clock elapsed time. Works as a context manager or function decorator. Optionally logs elapsed time and fires a post-action callback.

from soniclue_basekit import ExecTimer

# Context manager
with ExecTimer("loading data") as t:
    load_data()
print(t.interval)  # seconds elapsed

# Decorator
@ExecTimer("train step")
def train():
    ...

---

ExecTimeProfiler

Profiles execution time across named stages within a single thread. Supports nested stages, inner-time exclusion, and summary statistics (mean, std dev).

from soniclue_basekit import ExecTimeProfiler

profiler = ExecTimeProfiler()
with profiler.stage("preprocess"):
    ...
with profiler.stage("inference"):
    ...
profiler.print_report()

---

Attempter

Wraps a callable and re-invokes it on failure according to a pluggable retry strategy. Raise NoMoreAttempt inside the callable to abort retrying immediately.

from soniclue_basekit import Attempter, ExponentialBackoff, NoMoreAttempt

def fetch():
    resp = requests.get(url)
    if resp.status_code == 404:
        raise NoMoreAttempt()  # no point retrying
    resp.raise_for_status()
    return resp.json()

result = Attempter(fetch).execute(ExponentialBackoff(base=1.0, multiplier=2.0), max_attempts=5)

Retry strategies:

Class	Behaviour
ConstantPause(interval)	Fixed wait between retries
RandomPause(low, high)	Uniform random wait in [low, high]
ExponentialBackoff(base, multiplier, jitter)	base × multiplierᵃᵗᵗᵉᵐᵖᵗ + optional random jitter

---

LockProtector

Wraps an object with a threading.Lock. Supports enforcing lock acquisition order across protectors and optional profiling of contention.

from soniclue_basekit import LockProtector

state = LockProtector({"count": 0})
with state.acquire() as s:
    s["count"] += 1

---

DaemonThread / TerminableThread

Thread base classes with graceful shutdown support. TerminableThread exposes is_terminating() / set_terminating(). DaemonThread adds periodic-task scaffolding and optional OS-level thread naming (Linux).

from soniclue_basekit import DaemonThread

class Poller(DaemonThread):
    def _run_once(self):
        check_queue()

poller = Poller(interval_sec=5, name="poller")
poller.start()
...
poller.set_terminating()
poller.join()

---

ScriptRunner

Runs shell scripts (written as Python strings) in a subprocess with optional timeout, stdout/stderr capture, and streaming line callbacks.

from soniclue_basekit import ScriptRunner

runner = ScriptRunner("echo hello && echo world")
runner.run()

---

CliFormatter

Fluent builder for CLI argument strings passed to subprocess. Handles shell-quoting, key prefixes, and multi-value flags.

from soniclue_basekit import CliFormatter

cmd = (
    CliFormatter("ffmpeg")
    .add("-i", input_path)
    .add("-vf", "scale=1280:720")
    .add("-y", output_path)
    .build()
)

---

run_in_process / run_in_process_decorator

Run a callable in an isolated subprocess and return its result (or re-raise its exception) in the calling process. Useful for isolating GPU memory, avoiding fork-safety issues, or enforcing timeouts.

from soniclue_basekit import run_in_process, run_in_process_decorator

result = run_in_process(my_fn, timeout_sec=30)

@run_in_process_decorator
def heavy_job(x):
    return x * 2

---

is_pytest_env

Returns True when running inside a pytest session — handy for guards that should only activate during tests.

from soniclue_basekit import is_pytest_env

Development

Requires uv.

Run tests

uv run pytest

Lint

uv run ruff check src tests

Format

uv run ruff format src tests

Changelog

0.1.1

• Lowered minimum Python requirement from 3.11 to 3.10 (for ROS2 project).

0.1.0

• Initial release of featured libraries.

License

Apache 2.0 — see LICENSE.