FM / utils

/opt/imh-python/lib/python3.9/site-packages/celery/utils/

/opt/imh-python/lib/python3.9/site-packages/celery/utils

Name	Size	Mode	Actions
dispatch/	-	0755	rm
static/	-	0755	rm
__pycache__/	-	0755	rm
abstract.py	2874	0644	edit dl rm
collections.py	25432	0644	edit dl rm
debug.py	4709	0644	edit dl rm
deprecated.py	3620	0644	edit dl rm
functional.py	12017	0644	edit dl rm
graph.py	9041	0644	edit dl rm
imports.py	5126	0644	edit dl rm
iso8601.py	2916	0644	edit dl rm
log.py	8757	0644	edit dl rm
nodenames.py	3163	0644	edit dl rm
objects.py	4215	0644	edit dl rm
saferepr.py	9020	0644	edit dl rm
serialization.py	8209	0644	edit dl rm
sysinfo.py	1264	0644	edit dl rm
term.py	5135	0644	edit dl rm
text.py	5844	0644	edit dl rm
threads.py	9552	0644	edit dl rm
time.py	15039	0644	edit dl rm
timer2.py	4813	0644	edit dl rm
__init__.py	935	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/celery/utils/sysinfo.py (1264B)

"""System information utilities."""
from __future__ import annotations

import os
from math import ceil

from kombu.utils.objects import cached_property

__all__ = ('load_average', 'df')

if hasattr(os, 'getloadavg'):

def _load_average() -> tuple[float, ...]:
        return tuple(ceil(l * 1e2) / 1e2 for l in os.getloadavg())

else:  # pragma: no cover
    # Windows doesn't have getloadavg
    def _load_average() -> tuple[float, ...]:
        return 0.0, 0.0, 0.0,

def load_average() -> tuple[float, ...]:
    """Return system load average as a triple."""
    return _load_average()

class df:
    """Disk information."""

def __init__(self, path: str | bytes | os.PathLike) -> None:
        self.path = path

@property
    def total_blocks(self) -> float:
        return self.stat.f_blocks * self.stat.f_frsize / 1024

@property
    def available(self) -> float:
        return self.stat.f_bavail * self.stat.f_frsize / 1024

@property
    def capacity(self) -> int:
        avail = self.stat.f_bavail
        used = self.stat.f_blocks - self.stat.f_bfree
        return int(ceil(used * 100.0 / (used + avail) + 0.5))

@cached_property
    def stat(self) -> os.statvfs_result:
        return os.statvfs(os.path.abspath(self.path))

"""System information utilities.""" from __future__ import annotations import os from math import ceil from kombu.utils.objects import cached_property __all__ = ('load_average', 'df') if hasattr(os, 'getloadavg'): def _load_average() -> tuple[float, ...]: return tuple(ceil(l * 1e2) / 1e2 for l in os.getloadavg()) else: # pragma: no cover # Windows doesn't have getloadavg def _load_average() -> tuple[float, ...]: return 0.0, 0.0, 0.0, def load_average() -> tuple[float, ...]: """Return system load average as a triple.""" return _load_average() class df: """Disk information.""" def __init__(self, path: str | bytes | os.PathLike) -> None: self.path = path @property def total_blocks(self) -> float: return self.stat.f_blocks * self.stat.f_frsize / 1024 @property def available(self) -> float: return self.stat.f_bavail * self.stat.f_frsize / 1024 @property def capacity(self) -> int: avail = self.stat.f_bavail used = self.stat.f_blocks - self.stat.f_bfree return int(ceil(used * 100.0 / (used + avail) + 0.5)) @cached_property def stat(self) -> os.statvfs_result: return os.statvfs(os.path.abspath(self.path))