"""Local session state. No passwords.""" from __future__ import annotations import json from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any from gpu_rent.paths import runtime_dir, state_path CURRENT_VERSION = 1 @dataclass class SessionState: version: int = CURRENT_VERSION phase: str = "idle" server_id: str | None = None server_name: str | None = None flavor_id: str | None = None flavor_name: str | None = None boot_volume_id: str | None = None data_volume_id: str | None = None floating_ip: str | None = None keypair_name: str | None = None created_at: str | None = None unshelved_at: str | None = None notes: dict[str, Any] = field(default_factory=dict) floating_ip_id: str | None = None network_id: str | None = None security_group_id: str | None = None image_id: str | None = None availability_zone: str | None = None spot: bool = True bootstrapped: bool = False def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: dict[str, Any]) -> SessionState: known = {k: v for k, v in data.items() if k in cls.__dataclass_fields__} return cls(**known) def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def load_state() -> SessionState: path = state_path() if not path.is_file(): return SessionState() raw = json.loads(path.read_text(encoding="utf-8")) if not isinstance(raw, dict): return SessionState() return SessionState.from_dict(raw) def save_state(state: SessionState) -> None: runtime_dir().mkdir(parents=True, exist_ok=True) path = state_path() path.write_text(json.dumps(state.to_dict(), indent=2) + "\n", encoding="utf-8") def preempt_window_end(state: SessionState) -> datetime | None: stamp = state.unshelved_at or state.created_at if not stamp: return None try: start = datetime.fromisoformat(stamp) except ValueError: return None if start.tzinfo is None: start = start.replace(tzinfo=timezone.utc) from datetime import timedelta return start + timedelta(hours=24)