util.py

import logging
import math
import time
from collections.abc import MutableMapping
from typing import Any, TypeVar

from roborock.diagnostics import redact_device_uid

T = TypeVar("T")


def unpack_list(value: list[T], size: int) -> list[T | None]:
    return (value + [None] * size)[:size]  # type: ignore


class RoborockLoggerAdapter(logging.LoggerAdapter):
    def __init__(
        self,
        duid: str | None = None,
        name: str | None = None,
        logger: logging.Logger | None = None,
    ) -> None:
        super().__init__(logger or logging.getLogger(__name__), {})
        if name is not None:
            self.prefix = name
        elif duid is not None:
            self.prefix = redact_device_uid(duid)
        else:
            raise ValueError("Either duid or name must be provided")

    def process(self, msg: str, kwargs: MutableMapping[str, Any]) -> tuple[str, MutableMapping[str, Any]]:
        return f"[{self.prefix}] {msg}", kwargs


counter_map: dict[tuple[int, int], int] = {}


def get_next_int(min_val: int, max_val: int) -> int:
    """Gets a random int in the range, precached to help keep it fast."""
    if (min_val, max_val) not in counter_map:
        # If we have never seen this range, or if the cache is getting low, make a bunch of preshuffled values.
        counter_map[(min_val, max_val)] = min_val
    counter_map[(min_val, max_val)] += 1
    return counter_map[(min_val, max_val)] % max_val + min_val


def get_timestamp() -> int:
    """Get the current timestamp in seconds since epoch.

    This is separated out to allow for easier mocking in tests.
    """
    return math.floor(time.time())