Tim Schubert
eb81a1fedf
Some checks are pending
Continuous Integration / Checks (push) Waiting to run
128 lines
3 KiB
Python
Executable file
128 lines
3 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import requests
|
|
import logging
|
|
import subprocess
|
|
|
|
from datetime import datetime
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Status:
|
|
def status(self):
|
|
return None
|
|
|
|
|
|
class Cat(Status):
|
|
index = 0
|
|
|
|
def status(self):
|
|
cat_width = 200
|
|
index = self.index
|
|
catwalk = " " * (cat_width - index) + 1 * "🐈🏳️🌈" + " " * index
|
|
self.index = (index + 1) % cat_width
|
|
|
|
return {"full_text": catwalk}
|
|
|
|
|
|
class Space(Status):
|
|
backoff = 0
|
|
c_status = None
|
|
|
|
def status(self):
|
|
backoff = self.backoff
|
|
if self.backoff == 0:
|
|
self.update()
|
|
|
|
return {"full_text": self.c_status}
|
|
|
|
def update(self):
|
|
spacestatus_url = "https://status.stratum0.org/status.json"
|
|
resp = requests.get(url=spacestatus_url)
|
|
self.backoff = (self.backoff + 1) % 120
|
|
data = resp.json()
|
|
if data["isOpen"]:
|
|
since = datetime.strptime(data["since"], "%Y-%m-%dT%H:%M:%S.%f").strftime("%A at %H:%M")
|
|
spacestatus = f"Space is open since {since}"
|
|
else:
|
|
spacestatus = "Space is closed"
|
|
self.c_status = spacestatus
|
|
|
|
|
|
class Battery(Status):
|
|
capacity_file = open('/sys/class/power_supply/BAT0/capacity', 'r')
|
|
status_file = open('/sys/class/power_supply/BAT0/status', 'r')
|
|
|
|
def status(self):
|
|
self.status_file.seek(0)
|
|
status = self.status_file.read().rstrip()
|
|
|
|
self.capacity_file.seek(0)
|
|
capacity = self.capacity_file.read().rstrip()
|
|
|
|
battery = f"{status} {capacity}%"
|
|
|
|
return {"full_text": battery}
|
|
|
|
|
|
class Time(Status):
|
|
def status(self):
|
|
return {"full_text": datetime.now().strftime("%Vth %A %H:%M") }
|
|
|
|
|
|
class FailedUnits(Status):
|
|
def status(self):
|
|
proc = subprocess.run(["systemctl", "list-units", "--failed"], capture_output = True)
|
|
stdout = proc.stdout.decode('utf-8')
|
|
failed = 0
|
|
for line in stdout:
|
|
if 'failed' in line:
|
|
failed += 1
|
|
if failed is 0:
|
|
return {"full_text": f"No failed units"}
|
|
else:
|
|
return {"full_text": f"There are {failed} failed units", "color": "#ff0000"}
|
|
|
|
|
|
def print_header():
|
|
header = {
|
|
"version": 1,
|
|
"click_events": False,
|
|
}
|
|
print(json.dumps(header))
|
|
print("[")
|
|
|
|
|
|
def run(interval, widgets):
|
|
print_header()
|
|
|
|
while True:
|
|
body = []
|
|
|
|
for widget in widgets:
|
|
try:
|
|
status = widget.status()
|
|
except Exception as e:
|
|
logger.error(e)
|
|
if status:
|
|
body += status,
|
|
|
|
print(json.dumps(body), ",", flush=True)
|
|
|
|
ts = interval - (time.time() % interval)
|
|
time.sleep(ts)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
# Interval in seconds
|
|
interval = 1.0
|
|
|
|
widgets = [Cat(), FailedUnits(), Space(), Battery(), Time()]
|
|
|
|
run(interval, widgets)
|