stepping it back
This commit is contained in:
@@ -1,12 +1,12 @@
|
||||
import os
|
||||
import time
|
||||
import json
|
||||
import socket
|
||||
import struct
|
||||
import subprocess
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from warehouse import Vpn_cache, Media_flag
|
||||
import pulsectl
|
||||
import psutil
|
||||
|
||||
from warehouse import Vpn_cache, Media_flag, http_get
|
||||
|
||||
class Cached:
|
||||
def __init__(self, ttl):
|
||||
@@ -25,105 +25,122 @@ class Cached:
|
||||
return self.value
|
||||
|
||||
|
||||
def http_get(url, timeout=2):
|
||||
try:
|
||||
with urllib.request.urlopen(url, timeout=timeout) as r:
|
||||
return r.read().decode().strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
class StatusBar:
|
||||
def __init__(self):
|
||||
self.cpu = Cached(5)
|
||||
self.bat = Cached(5)
|
||||
self.vpn = Cached(60)
|
||||
self.ram = Cached(5)
|
||||
self.bat = Cached(1)
|
||||
self.vpn = Cached(1)
|
||||
self.rec = Cached(1)
|
||||
|
||||
def run(self, cmd):
|
||||
return subprocess.run(cmd, capture_output=True, text=True).stdout.strip()
|
||||
|
||||
def get_volume(self):
|
||||
return self.run(["pamixer", "--get-volume-human"])
|
||||
with pulsectl.Pulse("statusbar") as pulse:
|
||||
sink = pulse.get_sink_by_name(
|
||||
pulse.server_info().default_sink_name
|
||||
)
|
||||
|
||||
volume = sink.volume.value_flat * 100
|
||||
|
||||
if sink.mute:
|
||||
return "Muted"
|
||||
|
||||
return f"{volume:.0f}%"
|
||||
|
||||
def get_ram(self):
|
||||
out = self.run(["free", "-m"]).splitlines()
|
||||
mem = next((l for l in out if l.startswith("Mem:")), "")
|
||||
parts = mem.split()
|
||||
if len(parts) < 7:
|
||||
return "?"
|
||||
total = int(parts[1])
|
||||
used = int(parts[2])
|
||||
return f"{used//1000}GB/{total//1000}GB"
|
||||
def compute():
|
||||
mem = psutil.virtual_memory()
|
||||
used = mem.total - mem.available
|
||||
|
||||
return f"{used // 1024**3}GB/{mem.total // 1024**3}GB"
|
||||
|
||||
return self.ram.get(compute)
|
||||
|
||||
def get_cpu(self):
|
||||
def compute():
|
||||
try:
|
||||
out = self.run(["grep", "cpu ", "/proc/stat"])
|
||||
p = out.split()
|
||||
if len(p) < 8:
|
||||
return "0.0%"
|
||||
user, nice, system, idle = map(int, p[1:5])
|
||||
total = user + nice + system + idle
|
||||
busy = total - idle
|
||||
return f"{(busy / total) * 100:.1f}%"
|
||||
except Exception:
|
||||
return "0.0%"
|
||||
return f"{psutil.cpu_percent(interval=0.2):.1f}%"
|
||||
|
||||
return self.cpu.get(compute)
|
||||
|
||||
def get_battery(self):
|
||||
def compute():
|
||||
bat = Path("/sys/class/power_supply/BAT0")
|
||||
if not bat.exists():
|
||||
return ""
|
||||
try:
|
||||
cap = (bat / "capacity").read_text().strip()
|
||||
status = (bat / "status").read_text().strip()
|
||||
return f"{cap}% ({status})"
|
||||
except Exception:
|
||||
bat = psutil.sensors_battery()
|
||||
|
||||
if bat is None:
|
||||
return ""
|
||||
|
||||
if bat.power_plugged:
|
||||
status = "Charging"
|
||||
else:
|
||||
status = "Discharging"
|
||||
|
||||
return f"{bat.percent:.0f}% ({status})"
|
||||
|
||||
return self.bat.get(compute)
|
||||
|
||||
def get_vpn(self):
|
||||
def compute():
|
||||
if not self.run(["wg", "show", "interfaces"]):
|
||||
try:
|
||||
with open("/proc/net/dev") as f:
|
||||
interfaces = [
|
||||
line.split(":")[0].strip()
|
||||
for line in f.readlines()[2:]
|
||||
]
|
||||
|
||||
vpn = any(
|
||||
iface.startswith(("wg", "tun", "tap", "ppp"))
|
||||
for iface in interfaces
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return ""
|
||||
|
||||
if not vpn:
|
||||
if Vpn_cache.exists():
|
||||
Vpn_cache.unlink()
|
||||
return ""
|
||||
|
||||
country = ""
|
||||
if Vpn_cache.exists():
|
||||
country = Vpn_cache.read_text().strip()
|
||||
|
||||
if not country:
|
||||
ip = http_get("https://ifconfig.me")
|
||||
geo = http_get(f"https://iplookup.stab.ing/api/v1/lookup?ip={ip}")
|
||||
try:
|
||||
if Vpn_cache.exists():
|
||||
country = Vpn_cache.read_text().strip()
|
||||
|
||||
if '"country"' in geo:
|
||||
try:
|
||||
country = geo.split('"country":"')[1].split('"')[0]
|
||||
except Exception:
|
||||
country = ""
|
||||
if not country:
|
||||
ip = http_get("https://ifconfig.me")
|
||||
geo = http_get(
|
||||
f"https://iplookup.stab.ing/api/v1/lookup?ip={ip}"
|
||||
)
|
||||
|
||||
if country:
|
||||
Vpn_cache.write_text(country)
|
||||
country = json.loads(geo).get("country", "")
|
||||
|
||||
return f"VPN: {country or 'Unknown'}"
|
||||
if country:
|
||||
Vpn_cache.write_text(country)
|
||||
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
return f"VPN: {country or 'NAY!'}"
|
||||
|
||||
return self.vpn.get(compute)
|
||||
|
||||
def get_recording(self):
|
||||
def compute():
|
||||
return "RECORDING!" if self.run(["pgrep", "-x", "wf-recorder"]) else ""
|
||||
for proc in psutil.process_iter(["name"]):
|
||||
try:
|
||||
if proc.info["name"] == "wf-recorder":
|
||||
return "RECORDING!"
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||||
pass
|
||||
|
||||
return ""
|
||||
|
||||
return self.rec.get(compute)
|
||||
|
||||
def get_time(self):
|
||||
if Media_flag.exists():
|
||||
return "0001-01-01 12:00 AM"
|
||||
return self.run(["date", "+%Y-%m-%d %I:%M %p"])
|
||||
|
||||
return datetime.now().strftime("%Y-%m-%d %I:%M %p")
|
||||
|
||||
def render(self):
|
||||
vol = self.get_volume()
|
||||
@@ -138,14 +155,15 @@ class StatusBar:
|
||||
|
||||
if extra:
|
||||
return f"CPU: {cpu} / RAM: {ram} / VOL: {vol} / {extra} / {t}"
|
||||
|
||||
return f"CPU: {cpu} / RAM: {ram} / VOL: {vol} / {t}"
|
||||
|
||||
|
||||
def main():
|
||||
bar = StatusBar()
|
||||
|
||||
while True:
|
||||
print(bar.render(), flush=True)
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user