ki
28124c5617
- dashboard.py: plugin-based renderer with 4x4 grid layout - admin.py: web UI with layout editor + plugin configs - layout.py: pack algorithm, item placement, grid system - plugins/: clock, weather, system, spotify, strava, gmail, minimax, hello - network_watchdog.py: WiFi AP/client mode management - waveshare_epd_init.py: vendor driver stub
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""MiniMax Token-Usage Widget - responsive."""
|
|
import os, sys, json
|
|
import urllib.request, urllib.error
|
|
from datetime import datetime
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
from plugins.base import Widget
|
|
from palette import FG, INFO, OK, WARN, ALERT, ORANGE, BG, measure, fit_font, hbar, centered_text, is_small, is_wide, parse_thresholds, DEFAULT_THRESHOLDS
|
|
|
|
|
|
def _fmt_timedelta_short(seconds):
|
|
if seconds is None: return ""
|
|
s = int(seconds)
|
|
if s <= 0: return "now"
|
|
days, rem = divmod(s, 86400)
|
|
h, rem = divmod(rem, 3600)
|
|
m = rem // 60
|
|
if days: return f"{days}d {h}h"
|
|
if h: return f"{h}h {m}m"
|
|
return f"{m}m"
|
|
|
|
|
|
def _deep_get(d, *keys, default=None):
|
|
for k in keys:
|
|
if isinstance(d, dict) and k in d:
|
|
v = d[k]
|
|
if v is not None:
|
|
return v
|
|
return default
|
|
|
|
|
|
def _fetch_remains(subscription_key, base_url="https://api.minimax.io"):
|
|
url = f"{base_url.rstrip('/')}/v1/token_plan/remains"
|
|
req = urllib.request.Request(url, headers={
|
|
"Authorization": f"Bearer {subscription_key}",
|
|
"Content-Type": "application/json", "Accept": "application/json",
|
|
})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
raw = r.read()
|
|
try:
|
|
return {"ok": True, "data": json.loads(raw), "raw": raw.decode("utf-8", "ignore")}
|
|
except json.JSONDecodeError as e:
|
|
return {"ok": False, "error": f"invalid json: {e}",
|
|
"raw": raw.decode("utf-8", "ignore")[:500]}
|
|
except urllib.error.HTTPError as e:
|
|
body = ""
|
|
try: body = e.read().decode("utf-8", "ignore")[:300]
|
|
except Exception: pass
|
|
return {"ok": False, "error": f"HTTP {e.code} {e.reason}: {body}".strip()}
|
|
|
|
|
|
def _check_base_resp(data):
|
|
base = _deep_get(data, "base_resp")
|
|
if not isinstance(base, dict): return None
|
|
code = base.get("status_code", 0)
|
|
if code in (0, None, "0", ""): return None
|
|
return f"{base.get('status_msg') or base.get('message') or 'API-Fehler'} (code {code})"
|
|
|
|
|
|
def _parse_model_remains(items):
|
|
now_ms = int(datetime.now().timestamp() * 1000)
|
|
parsed = []
|
|
for entry in items:
|
|
if not isinstance(entry, dict): continue
|
|
model_name = entry.get("model_name", "model")
|
|
for win_label, rem_pct, end_time_raw, status_raw in [
|
|
("5-Hour", entry.get("current_interval_remaining_percent"),
|
|
entry.get("end_time"), entry.get("current_interval_status")),
|
|
("Weekly", entry.get("current_weekly_remaining_percent"),
|
|
entry.get("weekly_end_time"), entry.get("current_weekly_status")),
|
|
]:
|
|
if rem_pct is None: continue
|
|
try: status = int(status_raw) if status_raw is not None else 1
|
|
except (TypeError, ValueError): status = 1
|
|
if status != 1: continue
|
|
reset_sec = None
|
|
if isinstance(end_time_raw, (int, float)) and end_time_raw > 1e12:
|
|
reset_sec = max(0, int((end_time_raw - now_ms) / 1000))
|
|
elif isinstance(end_time_raw, (int, float)):
|
|
reset_sec = max(0, int(end_time_raw))
|
|
used_pct = max(0.0, min(100.0, 100.0 - float(rem_pct)))
|
|
parsed.append({
|
|
"label": win_label,
|
|
"used_pct": used_pct,
|
|
"remaining_pct": float(rem_pct),
|
|
"reset_seconds": reset_sec,
|
|
"raw": entry,
|
|
"model_name": model_name,
|
|
})
|
|
return parsed
|
|
|
|
|
|
def _parse_windows(data):
|
|
model_remains = data.get("model_remains")
|
|
if isinstance(model_remains, list) and model_remains:
|
|
return _parse_model_remains(model_remains)
|
|
candidates = [
|
|
data.get("remains"),
|
|
_deep_get(data, "data", "remains"),
|
|
_deep_get(data, "data", "windows"),
|
|
data.get("windows"),
|
|
_deep_get(data, "plan", "windows"),
|
|
]
|
|
raw_windows = None
|
|
for c in candidates:
|
|
if isinstance(c, list) and c:
|
|
raw_windows = c; break
|
|
if not isinstance(raw_windows, list):
|
|
return []
|
|
parsed = []
|
|
for w in raw_windows:
|
|
if not isinstance(w, dict): continue
|
|
reset_sec = w.get("reset_in") or w.get("reset_in_seconds")
|
|
if reset_sec is None:
|
|
reset_iso = w.get("reset_at") or w.get("resets_at")
|
|
if isinstance(reset_iso, str):
|
|
try:
|
|
dt = datetime.fromisoformat(reset_iso.replace("Z", "+00:00"))
|
|
reset_sec = (dt - datetime.now(dt.tzinfo)).total_seconds()
|
|
except Exception:
|
|
reset_sec = None
|
|
usage_raw = w.get("usage") or w.get("used_pct") or w.get("utilization")
|
|
used_pct = None
|
|
if isinstance(usage_raw, (int, float)):
|
|
used_pct = float(usage_raw) * 100 if float(usage_raw) <= 1.0 else float(usage_raw)
|
|
if used_pct is None: continue
|
|
name = (w.get("name") or w.get("window") or w.get("label") or "window").lower()
|
|
parsed.append({
|
|
"label": name.upper(),
|
|
"used_pct": min(100.0, max(0.0, used_pct)),
|
|
"remaining_pct": max(0.0, 100.0 - used_pct),
|
|
"reset_seconds": int(reset_sec) if reset_sec else None,
|
|
})
|
|
return parsed
|
|
|
|
|
|
def _parse_credits(data):
|
|
points = _deep_get(data, "points_balance", "credits_balance", "balance", "points", "credits")
|
|
if isinstance(points, (int, float)):
|
|
return {"points": float(points)}
|
|
nested = _deep_get(data, "plan", "credits")
|
|
if isinstance(nested, (int, float)):
|
|
return {"points": float(nested)}
|
|
return None
|
|
|
|
|
|
class Widget(Widget):
|
|
name = "minimax"
|
|
label = "MiniMax Token Usage"
|
|
description = "Quota für MiniMax Token-Plan (5-Hour + Weekly). Responsive."
|
|
category = "info"
|
|
|
|
config_schema = [
|
|
{"key": "subscription_key", "label": "Subscription Key", "type": "secret",
|
|
"help": "Token-Plan Subscription Key. NICHT der pay-as-you-go API-Key."},
|
|
{"key": "show_credits", "label": "Credit-Balance anzeigen", "type": "bool", "default": True},
|
|
{"key": "api_base", "label": "API Base URL", "type": "string", "default": "https://api.minimax.io"},
|
|
{"key": "bar_thresholds", "label": "Quota-Schwellen (ok,warn,alert)",
|
|
"type": "string", "default": "ok@50,warn@80,alert@95",
|
|
"help": "Format: 'ok@50,warn@80,alert@95' oder 'ok,warn,alert' (default 50/80/95)"},
|
|
{"key": "bar_gradient", "label": "Verlaufsmodus",
|
|
"type": "bool", "default": True,
|
|
"help": "Wenn aus, einfarbige Bar in der Farbe der aktuellen Schwelle."},
|
|
]
|
|
default_config = {"subscription_key": "", "show_credits": True,
|
|
"api_base": "https://api.minimax.io",
|
|
"bar_thresholds": "ok@50,warn@80,alert@95",
|
|
"bar_gradient": True}
|
|
|
|
def _bar_args(self):
|
|
return (parse_thresholds(self.cfg("bar_thresholds")),
|
|
bool(self.cfg("bar_gradient", True)))
|
|
|
|
def fetch(self):
|
|
key = self.cfg("subscription_key")
|
|
if not key:
|
|
return {"_error": "Subscription Key fehlt — in der Admin-UI setzen."}
|
|
try:
|
|
r = _fetch_remains(key, self.cfg("api_base", "https://api.minimax.io"))
|
|
if not r.get("ok"):
|
|
return {"_error": r.get("error", "unbekannter Fehler")}
|
|
data = r["data"]
|
|
base_err = _check_base_resp(data)
|
|
if base_err:
|
|
return {"_error": base_err,
|
|
"_raw_keys": list(data.keys()) if isinstance(data, dict) else None}
|
|
windows = _parse_windows(data)
|
|
credits = _parse_credits(data) if self.cfg("show_credits") else None
|
|
if not windows and not credits:
|
|
return {"_error": "Schema unbekannt.", "_raw_keys": list(data.keys()) if isinstance(data, dict) else None}
|
|
return {"windows": windows, "credits": credits}
|
|
except Exception as e:
|
|
return {"_error": str(e)}
|
|
|
|
def _render_window(self, draw, fonts, x, y, w, h, wn, compact=False):
|
|
from palette import measure
|
|
# Label + Reset rechts
|
|
label = wn["label"]
|
|
reset_txt = ("R " + _fmt_timedelta_short(wn["reset_seconds"])) if wn.get("reset_seconds") else ""
|
|
font_label = fit_font(draw, label, fonts, w - 60, 22)
|
|
draw.text((x, y), label, font=font_label, fill=FG)
|
|
if reset_txt and not compact:
|
|
font_r = fit_font(draw, reset_txt, fonts, 60, 20)
|
|
rw, rh = measure(draw, reset_txt, font_r)
|
|
draw.text((x + w - rw - 4, y + 2), reset_txt, font=font_r, fill=FG)
|
|
# Bar
|
|
bar_y = y + (22 if not compact else 18)
|
|
bar_h = max(10, h - (bar_y - y) - 4)
|
|
pct = wn["used_pct"]
|
|
thresholds, gradient = self._bar_args()
|
|
hbar(draw, x, bar_y, w, bar_h, pct,
|
|
thresholds=thresholds, gradient=gradient)
|
|
# Prozent overlay
|
|
if not compact:
|
|
pct_txt = f"{int(pct + 0.5)}%"
|
|
font_pct = fit_font(draw, pct_txt, fonts, 60, bar_h - 4)
|
|
tw, th = measure(draw, pct_txt, font_pct)
|
|
draw.text((x + w - tw - 8, bar_y + 2), pct_txt, font=font_pct, fill=BG)
|
|
|
|
def render(self, draw, fonts, x, y, w, h):
|
|
from palette import measure
|
|
pad = 8
|
|
draw.text((x + pad, y + pad), "MINIMAX", font=fonts.get("24", fonts.get("20")), fill=INFO)
|
|
|
|
d = self.fetch()
|
|
if "_error" in d:
|
|
draw.text((x + pad, y + 60), "Konfiguration fehlt", font=fonts.get("24", fonts.get("20")), fill=WARN)
|
|
err = d["_error"][:50]
|
|
draw.text((x + pad, y + 90), err, font=fonts.get("20", fonts.get("16")), fill=FG)
|
|
if d.get("_raw_keys"):
|
|
draw.text((x + pad, y + 120), f"raw keys: {','.join(d['_raw_keys'][:6])}",
|
|
font=fonts.get("16", fonts.get("default")), fill=FG)
|
|
return
|
|
|
|
windows = d.get("windows", [])
|
|
|
|
if is_small(w, h):
|
|
# mini: nur "5h: 38%" etc kompakt
|
|
line_y = y + 50
|
|
for wn in windows[:2]:
|
|
label = f"{wn['label'][:1]}{wn['remaining_pct']:.0f}%"
|
|
font = fit_font(draw, label, fonts, w - 2 * pad, h // 3)
|
|
draw.text((x + pad, line_y), label, font=font, fill=FG)
|
|
line_y += h // 3
|
|
return
|
|
|
|
if is_wide(w, h):
|
|
# Wide strip: alle windows nebeneinander
|
|
n = max(1, len(windows))
|
|
col_w = w // n
|
|
for i, wn in enumerate(windows):
|
|
cx = x + i * col_w
|
|
self._render_window(draw, fonts, cx + pad // 2, y + 30,
|
|
col_w - pad, h - 35, wn, compact=True)
|
|
return
|
|
|
|
# Standard: vertikale Liste mit Bars
|
|
row_h = max(40, (h - 60) // max(1, len(windows)))
|
|
for i, wn in enumerate(windows):
|
|
row_y = y + 50 + i * row_h
|
|
self._render_window(draw, fonts, x + pad, row_y,
|
|
w - 2 * pad, row_h - 6, wn)
|
|
|
|
# Credits
|
|
credits = d.get("credits")
|
|
if credits and self.cfg("show_credits") and h > 220:
|
|
pts = credits.get("points")
|
|
if isinstance(pts, (int, float)):
|
|
txt = f"Credits: {pts:,.0f}"
|
|
font = fit_font(draw, txt, fonts, w - 2 * pad, 22)
|
|
draw.text((x + pad, y + h - 28), txt, font=font, fill=ORANGE)
|