File size: 3,402 Bytes
6754826 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import json
import time
from pathlib import Path
from threading import Lock
from config import ANAM_RATE_PER_MINUTE
_FILE = Path(__file__).parent / "telemetry_data.json"
_lock = Lock()
def _load() -> list[dict]:
if not _FILE.exists():
return []
try:
return json.loads(_FILE.read_text())
except Exception:
return []
def _save(sessions: list[dict]) -> None:
_FILE.write_text(json.dumps(sessions, indent=2))
def _billed_seconds(elapsed: float) -> int:
return int(max(30.0, elapsed))
def _cost(billed_seconds: int) -> float:
return round(billed_seconds / 60 * ANAM_RATE_PER_MINUTE, 4)
def start_session(session_id: str, persona_name: str, plan: str) -> None:
with _lock:
sessions = _load()
sessions.append({
"session_id": session_id,
"persona_name": persona_name,
"plan": plan,
"started_at": time.time(),
"ended_at": None,
"duration_seconds": None,
"billed_seconds": None,
"estimated_cost": None,
"status": "active",
})
_save(sessions)
def end_session(session_id: str) -> dict | None:
with _lock:
sessions = _load()
for s in reversed(sessions):
if s["session_id"] == session_id and s["status"] == "active":
now = time.time()
elapsed = now - s["started_at"]
billed = _billed_seconds(elapsed)
s["ended_at"] = now
s["duration_seconds"] = round(elapsed, 1)
s["billed_seconds"] = billed
s["estimated_cost"] = _cost(billed)
s["status"] = "ended"
_save(sessions)
return s
return None
def summary() -> dict:
sessions = _load()
ended = [s for s in sessions if s["status"] == "ended"]
active = [s for s in sessions if s["status"] == "active"]
total_billed = sum(s["billed_seconds"] for s in ended)
total_cost = sum(s["estimated_cost"] for s in ended)
now = time.time()
live_billed = 0
live_cost = 0.0
for s in active:
b = _billed_seconds(now - s["started_at"])
live_billed += b
live_cost += _cost(b)
return {
"rate_per_minute": ANAM_RATE_PER_MINUTE,
"ended_sessions": len(ended),
"active_sessions": len(active),
"total_billed_seconds": total_billed,
"total_billed_minutes": round(total_billed / 60, 2),
"total_estimated_cost": round(total_cost, 2),
"live_billed_seconds": live_billed,
"live_estimated_cost": round(live_cost, 2),
"grand_total_estimated_cost": round(total_cost + live_cost, 2),
}
def list_sessions() -> list[dict]:
now = time.time()
out = []
for s in _load():
s = dict(s)
if s["status"] == "active":
elapsed = now - s["started_at"]
s["duration_seconds"] = round(elapsed, 1)
s["billed_seconds"] = _billed_seconds(elapsed)
s["estimated_cost"] = _cost(s["billed_seconds"])
out.append(s)
out.sort(key=lambda s: s["started_at"], reverse=True)
return out
def reset() -> None:
with _lock:
_save([])
|