diff options
| -rw-r--r-- | deckit/modules/llm.py | 46 | ||||
| -rw-r--r-- | tests/test_llm.py | 32 |
2 files changed, 78 insertions, 0 deletions
diff --git a/deckit/modules/llm.py b/deckit/modules/llm.py new file mode 100644 index 0000000..2e1cda4 --- /dev/null +++ b/deckit/modules/llm.py @@ -0,0 +1,46 @@ +"""Claude Code and opencode usage, relayed from llm-cost-dashboard's API. + +Everything here is read-only. The dashboard stays the owner of the data and the +pricing; deck-it only renders it. +""" + +import json +import urllib.request + +from ..module import Module + +ENDPOINTS = ("/api/claude_limits", "/api/claude_session", "/api/summary") + + +def _fetch_for(base): + def fetch(endpoint): + with urllib.request.urlopen(base + endpoint, timeout=2) as r: + return json.loads(r.read()) + return fetch + + +def make(cfg: dict) -> Module: + base = cfg.get("llm", {}).get("url", "http://127.0.0.1:8420").rstrip("/") + fetch = cfg.get("fetch") or _fetch_for(base) + + def state(): + try: + data = {ep: fetch(ep) for ep in ENDPOINTS} + except (OSError, ValueError): + return {"offline": True} + limits = data["/api/claude_limits"] + latest = limits[-1] if isinstance(limits, list) and limits else ( + limits if isinstance(limits, dict) else {}) + return { + "offline": False, + "claude": { + "context_pct": latest.get("context_pct", data["/api/claude_session"].get("context_used")), + "five_hour": latest.get("five_hour"), + "seven_day": latest.get("seven_day"), + "monthly_used": latest.get("extra_used_credits"), + "monthly_limit": latest.get("extra_monthly_limit"), + }, + "opencode": data["/api/summary"], + } + + return Module(name="llm", title="LLM usage", state=state) diff --git a/tests/test_llm.py b/tests/test_llm.py new file mode 100644 index 0000000..759666c --- /dev/null +++ b/tests/test_llm.py @@ -0,0 +1,32 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import unittest + +import deckit.modules.llm as llm + + +class Down(unittest.TestCase): + def test_offline_when_dashboard_down(self): + def boom(url, timeout=2): + raise OSError("refused") + st = llm.make({"llm": {"url": "http://x"}, "fetch": boom}).state() + self.assertTrue(st["offline"]) + + +class Up(unittest.TestCase): + def test_merges_endpoints(self): + data = {"/api/claude_limits": [{"context_pct": 12}], + "/api/claude_session": {"context_used": 1200}, + "/api/summary": {"total_cost": 1.5}} + st = llm.make({"llm": {"url": "http://x"}, + "fetch": lambda u, timeout=2: data[u]}).state() + self.assertFalse(st["offline"]) + self.assertEqual(st["claude"]["context_pct"], 12) + self.assertEqual(st["opencode"]["total_cost"], 1.5) + + +if __name__ == "__main__": + unittest.main() |
