Module `livekit.agents.utils.hw`

Functions

def get_cpu_monitor() ‑> livekit.agents.utils.hw.cpu.CPUMonitor

Expand source code

def get_cpu_monitor() -> CPUMonitor:
    if _is_cgroup_v2():
        return CGroupV2CPUMonitor()
    if _is_cgroup_v1():
        return CGroupV1CPUMonitor()
    return DefaultCPUMonitor()

Classes

class CGroupV2CPUMonitor

Expand source code

class CGroupV2CPUMonitor(CPUMonitor):
    def cpu_count(self) -> float:
        # quota: The maximum CPU time in microseconds that the cgroup can use within a given period.
        # period: The period of time in microseconds over which the quota applies.
        # If the quota is set to "max", it means the cgroup is allowed to use all available CPUs without restriction.  # noqa: E501
        # Otherwise, the quota is a number that represents the maximum CPU time in microseconds that the cgroup can use within a given period.  # noqa: E501
        env_cpus = _cpu_count_from_env()
        if env_cpus is not None:
            return env_cpus
        quota, period = self._read_cpu_max()
        if quota == "max":
            return psutil.cpu_count() or 1.0
        return 1.0 * int(quota) / period

    def cpu_percent(self, interval: float = 0.5) -> float:
        cpu_usage_start = self._read_cpu_usage()
        time.sleep(interval)
        cpu_usage_end = self._read_cpu_usage()
        cpu_usage_diff = cpu_usage_end - cpu_usage_start

        # microseconds to seconds
        cpu_usage_seconds = cpu_usage_diff / 1_000_000

        num_cpus = self.cpu_count()
        cpu_usage_percent = cpu_usage_seconds / (interval * num_cpus)

        return min(cpu_usage_percent, 1)

    def _read_cpu_max(self) -> tuple[str, int]:
        try:
            with open("/sys/fs/cgroup/cpu.max") as f:
                data = f.read().strip().split()
            quota = data[0]
            period = int(data[1])
        except FileNotFoundError:
            quota = "max"
            period = 100000
        return quota, period

    def _read_cpu_usage(self) -> int:
        with open("/sys/fs/cgroup/cpu.stat") as f:
            for line in f:
                if line.startswith("usage_usec"):
                    return int(line.split()[1])
        raise RuntimeError("Failed to read CPU usage")

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

livekit.agents.utils.hw.cpu.CPUMonitor
abc.ABC

Methods

def cpu_count(self) ‑> float

Expand source code

def cpu_count(self) -> float:
    # quota: The maximum CPU time in microseconds that the cgroup can use within a given period.
    # period: The period of time in microseconds over which the quota applies.
    # If the quota is set to "max", it means the cgroup is allowed to use all available CPUs without restriction.  # noqa: E501
    # Otherwise, the quota is a number that represents the maximum CPU time in microseconds that the cgroup can use within a given period.  # noqa: E501
    env_cpus = _cpu_count_from_env()
    if env_cpus is not None:
        return env_cpus
    quota, period = self._read_cpu_max()
    if quota == "max":
        return psutil.cpu_count() or 1.0
    return 1.0 * int(quota) / period

Number of logical CPUs.

Returns a float to allow for fractional CPUs (in the case of cgroups).

def cpu_percent(self, interval: float = 0.5) ‑> float

Expand source code

def cpu_percent(self, interval: float = 0.5) -> float:
    cpu_usage_start = self._read_cpu_usage()
    time.sleep(interval)
    cpu_usage_end = self._read_cpu_usage()
    cpu_usage_diff = cpu_usage_end - cpu_usage_start

    # microseconds to seconds
    cpu_usage_seconds = cpu_usage_diff / 1_000_000

    num_cpus = self.cpu_count()
    cpu_usage_percent = cpu_usage_seconds / (interval * num_cpus)

    return min(cpu_usage_percent, 1)

CPU usage percentage between 0 and 1

class CPUMonitor

Expand source code

class CPUMonitor(ABC):
    @abstractmethod
    def cpu_count(self) -> float:
        """Number of logical CPUs.

        Returns a float to allow for fractional CPUs (in the case of cgroups)."""
        pass

    @abstractmethod
    def cpu_percent(self, interval: float = 0.5) -> float:
        """CPU usage percentage between 0 and 1"""
        pass

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

abc.ABC

Subclasses

livekit.agents.utils.hw.cpu.CGroupV1CPUMonitor
livekit.agents.utils.hw.cpu.CGroupV2CPUMonitor
livekit.agents.utils.hw.cpu.DefaultCPUMonitor

Methods

def cpu_count(self) ‑> float

Expand source code

@abstractmethod
def cpu_count(self) -> float:
    """Number of logical CPUs.

    Returns a float to allow for fractional CPUs (in the case of cgroups)."""
    pass

Number of logical CPUs.

Returns a float to allow for fractional CPUs (in the case of cgroups).

def cpu_percent(self, interval: float = 0.5) ‑> float

Expand source code

@abstractmethod
def cpu_percent(self, interval: float = 0.5) -> float:
    """CPU usage percentage between 0 and 1"""
    pass

CPU usage percentage between 0 and 1

class DefaultCPUMonitor

Expand source code

class DefaultCPUMonitor(CPUMonitor):
    def cpu_count(self) -> float:
        return _cpu_count_from_env() or psutil.cpu_count() or 1.0

    def cpu_percent(self, interval: float = 0.5) -> float:
        return psutil.cpu_percent(interval) / 100.0

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

livekit.agents.utils.hw.cpu.CPUMonitor
abc.ABC

Methods

def cpu_count(self) ‑> float

Expand source code

def cpu_count(self) -> float:
    return _cpu_count_from_env() or psutil.cpu_count() or 1.0

Number of logical CPUs.

Returns a float to allow for fractional CPUs (in the case of cgroups).

def cpu_percent(self, interval: float = 0.5) ‑> float

Expand source code

def cpu_percent(self, interval: float = 0.5) -> float:
    return psutil.cpu_percent(interval) / 100.0

CPU usage percentage between 0 and 1