| 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([]) |
|
|
