Stop EWW network statistics changing width
This commit is contained in:
parent
abbd3df78b
commit
3630cc2bf5
2 changed files with 6 additions and 4 deletions
119
eww/scripts/network_statistics.py
Executable file
119
eww/scripts/network_statistics.py
Executable file
|
@ -0,0 +1,119 @@
|
|||
#!/bin/env python
|
||||
"""Script that outputs network activity"""
|
||||
|
||||
from pathlib import Path
|
||||
from json import dumps
|
||||
from sys import argv
|
||||
from time import sleep
|
||||
from typing import Literal, NoReturn
|
||||
|
||||
|
||||
def get_interfaces() -> list[Path]:
|
||||
return list(Path("/sys/class/net/").iterdir())
|
||||
|
||||
|
||||
class TransferredBytes:
|
||||
def __init__(self, interface: Path, statistic: Literal["tx"] | Literal["rx"]) -> None:
|
||||
self._statistic: Path = interface / f"statistics/{statistic}_bytes"
|
||||
self._current: int = int(self._statistic.read_text())
|
||||
self._previous: int = self._current
|
||||
|
||||
def update(self) -> None:
|
||||
self._previous = self._current
|
||||
self._current = int(self._statistic.read_text())
|
||||
|
||||
def value(self, interval: float) -> float:
|
||||
return (self._current - self._previous) / interval
|
||||
|
||||
|
||||
class Status:
|
||||
def __init__(self, interface: Path) -> None:
|
||||
self._statistic: Path = interface / "operstate"
|
||||
|
||||
def up(self) -> bool:
|
||||
return self._statistic.read_text().strip() == "up"
|
||||
|
||||
|
||||
def format_4_significant_digits(num: float) -> str:
|
||||
assert num >= 1.0, "Doesn't properly handle numbers below 1"
|
||||
if num < 999:
|
||||
return f"{num:04.3g}"
|
||||
if num < 1000:
|
||||
# Above doesn't nicely handle the special case in [999.5 1000)
|
||||
return "0999"
|
||||
return f"{num:.4g}"
|
||||
|
||||
|
||||
EBI_UNITS = ("B", "kiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB", "RiB", "QiB")
|
||||
SHORT_UNITS = ("B", "k", "M", "G", "T", "P", "E", "Z", "Y", "R", "Q")
|
||||
|
||||
|
||||
def format_bytes(num: float) -> str:
|
||||
for prefix in SHORT_UNITS:
|
||||
if num < 1024:
|
||||
return f"{format_4_significant_digits(num)}{prefix}"
|
||||
num /= 1024
|
||||
return "what the fuck"
|
||||
|
||||
|
||||
class Interface:
|
||||
def __init__(self, location: Path) -> None:
|
||||
self._location: Path = location
|
||||
self._tx: TransferredBytes = TransferredBytes(location, "tx")
|
||||
self._rx: TransferredBytes = TransferredBytes(location, "rx")
|
||||
self._status: Status = Status(location)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._location.name
|
||||
|
||||
def update(self) -> None:
|
||||
self._tx.update()
|
||||
self._rx.update()
|
||||
|
||||
def json(self, interval: float) -> dict[str, str | bool | float]:
|
||||
interval_tx = self._tx.value(interval)
|
||||
interval_rx = self._rx.value(interval)
|
||||
|
||||
return {
|
||||
"tx": format_bytes(interval_tx),
|
||||
"rx": format_bytes(interval_rx),
|
||||
"combined": format_bytes(interval_tx + interval_rx),
|
||||
"combined_raw": interval_tx + interval_rx,
|
||||
"up": self._status.up(),
|
||||
}
|
||||
|
||||
|
||||
class Statistics:
|
||||
def __init__(self) -> None:
|
||||
self._interfaces: list[Interface] = [Interface(location) for location in get_interfaces()]
|
||||
|
||||
def update(self) -> None:
|
||||
for interface in self._interfaces:
|
||||
interface.update()
|
||||
|
||||
def json(self, interval: float) -> str:
|
||||
return dumps(
|
||||
{
|
||||
interface.name: interface.json(interval) for interface in self._interfaces
|
||||
},
|
||||
separators=(',', ':'),
|
||||
)
|
||||
|
||||
|
||||
def main(interval: float) -> NoReturn:
|
||||
stats = Statistics()
|
||||
sleep(0.5)
|
||||
stats.update()
|
||||
print(stats.json(interval))
|
||||
|
||||
while True:
|
||||
sleep(interval)
|
||||
stats.update()
|
||||
print(stats.json(interval), flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(argv) != 2:
|
||||
exit("Must provide 1 arg: Polling interval")
|
||||
main(float(argv[1]))
|
Loading…
Add table
Add a link
Reference in a new issue