Utilities¶

Validators, the generic sliding Window, and a timing context manager. These are public because user-defined transforms need the same validation behaviour as the built-in ones.

utils ¶

Small, dependency-free helpers shared across the package.

Three things live here:

Validators (as_floats, require_non_empty, require_positive) that coerce and check inputs once, at the public API boundary, and raise ValidationError with a uniform message.
Window, a generic fixed-capacity sliding window used by the streaming filters in mypackage.smoothing.
timer, a context manager for wall-clock measurements in examples and benchmarks.

Window ¶

Window(capacity: int)

A fixed-capacity sliding window over a stream of values.

Pushing past capacity evicts the oldest element, so the window always holds the most recent capacity items. The element type is generic: the window itself never inspects the values it stores.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	Maximum number of retained elements. Must be positive.	required

Raises:

Type	Description
`ValidationError`	If `capacity` is not a positive integer.

Examples:

>>> window: Window[int] = Window(capacity=3)
>>> for value in (1, 2, 3, 4):
...     _ = window.push(value)
>>> list(window)
[2, 3, 4]
>>> len(window), window.is_full
(3, True)

Pushing returns self, so calls chain:

>>> list(Window[str](capacity=2).push("a").push("b").push("c"))
['b', 'c']

Source code in src/mypackage/utils.py

def __init__(self, capacity: int) -> None:
    self.capacity = require_positive(capacity, name="capacity")
    self._items: deque[T] = deque(maxlen=self.capacity)

is_full `property` ¶

is_full: bool

Whether the window holds exactly capacity elements.

push ¶

push(value: T) -> Window[T]

Append value, evicting the oldest element if the window is full.

Parameters:

Name	Type	Description	Default
`value`	`T`	The item to append.	required

Returns:

Type	Description
`Window[T]`	`self`, so that pushes can be chained.

Source code in src/mypackage/utils.py

def push(self, value: T) -> Window[T]:
    """Append ``value``, evicting the oldest element if the window is full.

    Args:
        value: The item to append.

    Returns:
        ``self``, so that pushes can be chained.
    """
    self._items.append(value)
    return self

clear ¶

clear() -> None

Drop every retained element.

Source code in src/mypackage/utils.py

def clear(self) -> None:
    """Drop every retained element."""
    self._items.clear()

Elapsed `dataclass` ¶

Elapsed(seconds: float = 0.0)

Wall-clock duration recorded by timer.

Attributes:

Name	Type	Description
`seconds`	`float`	Elapsed wall-clock time. Zero until the `with` block exits.

milliseconds `property` ¶

milliseconds: float

Elapsed time in milliseconds.

as_floats ¶

as_floats(series: Series | Sequence[str], *, name: str = 'series') -> list[float]

Coerce a sequence of numbers to a fresh list[float].

Strings that parse as numbers are accepted too, which is what lets mypackage.cli hand raw tokens straight to this function instead of duplicating the validation logic.

Parameters:

Name	Type	Description	Default
`series`	`Series \| Sequence[str]`	The sequence to coerce. Never mutated.	required
`name`	`str`	Argument name used in the error message.	`'series'`

Returns:

Type	Description
`list[float]`	A new list containing every element of `series` as a `float`.

Raises:

Type	Description
`ValidationError`	If any element cannot be converted to `float`, or if `series` is not a sequence.

Examples:

>>> as_floats([1, 2, 3])
[1.0, 2.0, 3.0]
>>> as_floats(range(3))
[0.0, 1.0, 2.0]
>>> as_floats(["1.5", "2"])
[1.5, 2.0]
>>> try:
...     as_floats([1.0, "oops"])
... except ValidationError as err:
...     print(err)
'series' must contain only numbers, got 'oops' at index 1

Source code in src/mypackage/utils.py

def as_floats(series: Series | Sequence[str], *, name: str = "series") -> list[float]:
    """Coerce a sequence of numbers to a fresh `list[float]`.

    Strings that parse as numbers are accepted too, which is what lets
    [`mypackage.cli`][] hand raw tokens straight to this function instead of
    duplicating the validation logic.

    Args:
        series: The sequence to coerce. Never mutated.
        name: Argument name used in the error message.

    Returns:
        A new list containing every element of ``series`` as a `float`.

    Raises:
        ValidationError: If any element cannot be converted to `float`, or if
            ``series`` is not a sequence.

    Examples:
        >>> as_floats([1, 2, 3])
        [1.0, 2.0, 3.0]
        >>> as_floats(range(3))
        [0.0, 1.0, 2.0]
        >>> as_floats(["1.5", "2"])
        [1.5, 2.0]
        >>> try:
        ...     as_floats([1.0, "oops"])
        ... except ValidationError as err:
        ...     print(err)
        'series' must contain only numbers, got 'oops' at index 1
    """
    try:
        values = list(series)
    except TypeError as err:  # pragma: no cover - defensive
        msg = f"{name!r} must be a sequence of numbers, got {type(series).__name__}"
        raise ValidationError(msg) from err

    out: list[float] = []
    for index, value in enumerate(values):
        try:
            out.append(float(value))
        except (TypeError, ValueError) as err:
            msg = f"{name!r} must contain only numbers, got {value!r} at index {index}"
            raise ValidationError(msg) from err
    return out

require_non_empty ¶

require_non_empty(series: Series, *, name: str = 'series') -> list[float]

Coerce series to floats and require at least one element.

Parameters:

Name	Type	Description	Default
`series`	`Series`	The sequence to validate.	required
`name`	`str`	Argument name used in the error message.	`'series'`

Returns:

Type	Description
`list[float]`	A new `list[float]` with the validated contents.

Raises:

Type	Description
`EmptySeriesError`	If `series` has no elements.
`ValidationError`	If any element is not a number.

Examples:

>>> require_non_empty((1, 2))
[1.0, 2.0]
>>> try:
...     require_non_empty([])
... except EmptySeriesError as err:
...     print(err)
'series' must contain at least one value, got 0

Source code in src/mypackage/utils.py

def require_non_empty(series: Series, *, name: str = "series") -> list[float]:
    """Coerce ``series`` to floats and require at least one element.

    Args:
        series: The sequence to validate.
        name: Argument name used in the error message.

    Returns:
        A new `list[float]` with the validated contents.

    Raises:
        EmptySeriesError: If ``series`` has no elements.
        ValidationError: If any element is not a number.

    Examples:
        >>> require_non_empty((1, 2))
        [1.0, 2.0]
        >>> try:
        ...     require_non_empty([])
        ... except EmptySeriesError as err:
        ...     print(err)
        'series' must contain at least one value, got 0
    """
    values = as_floats(series, name=name)
    if not values:
        msg = f"{name!r} must contain at least one value, got 0"
        raise EmptySeriesError(msg)
    return values

require_positive ¶

require_positive(value: int, *, name: str = 'window') -> int

Require value to be a strictly positive integer.

Parameters:

Name	Type	Description	Default
`value`	`int`	The integer to validate. `bool` is rejected explicitly, since `True` would otherwise silently pass as `1`.	required
`name`	`str`	Argument name used in the error message.	`'window'`

Returns:

Type	Description
`int`	The validated integer.

Raises:

Type	Description
`ValidationError`	If `value` is not a positive `int`.

Examples:

>>> require_positive(3, name="window")
3
>>> try:
...     require_positive(0, name="window")
... except ValidationError as err:
...     print(err)
'window' must be a positive integer, got 0

Source code in src/mypackage/utils.py

def require_positive(value: int, *, name: str = "window") -> int:
    """Require ``value`` to be a strictly positive integer.

    Args:
        value: The integer to validate. `bool` is rejected explicitly, since
            ``True`` would otherwise silently pass as ``1``.
        name: Argument name used in the error message.

    Returns:
        The validated integer.

    Raises:
        ValidationError: If ``value`` is not a positive `int`.

    Examples:
        >>> require_positive(3, name="window")
        3
        >>> try:
        ...     require_positive(0, name="window")
        ... except ValidationError as err:
        ...     print(err)
        'window' must be a positive integer, got 0
    """
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        msg = f"{name!r} must be a positive integer, got {value!r}"
        raise ValidationError(msg)
    return value

timer ¶

timer() -> Iterator[Elapsed]

Measure the wall-clock time spent inside a with block.

The yielded Elapsed is populated when the block exits — including when it exits via an exception.

Yields:

Type	Description
`Elapsed`	An `Elapsed` record that is filled in on exit.

Examples:

>>> with timer() as elapsed:
...     total = sum(range(1_000))
>>> elapsed.seconds >= 0.0
True
>>> elapsed.milliseconds == elapsed.seconds * 1e3
True

Source code in src/mypackage/utils.py

@contextmanager
def timer() -> Iterator[Elapsed]:
    """Measure the wall-clock time spent inside a ``with`` block.

    The yielded [`Elapsed`][mypackage.Elapsed] is populated when the block
    exits — including when it exits via an exception.

    Yields:
        An `Elapsed` record that is filled in on exit.

    Examples:
        >>> with timer() as elapsed:
        ...     total = sum(range(1_000))
        >>> elapsed.seconds >= 0.0
        True
        >>> elapsed.milliseconds == elapsed.seconds * 1e3
        True
    """
    elapsed = Elapsed()
    start = time.perf_counter()
    try:
        yield elapsed
    finally:
        elapsed.seconds = time.perf_counter() - start

Utilities¶

utils ¶

Window ¶

is_full property ¶

push ¶

clear ¶

Elapsed dataclass ¶

milliseconds property ¶

as_floats ¶

require_non_empty ¶

require_positive ¶

timer ¶

is_full `property` ¶

Elapsed `dataclass` ¶

milliseconds `property` ¶