Leonid Pershin
343f741baa
- Updated the project structure to store configuration files (.env, models.yaml, extensions.yaml) in the project root instead of the user's home directory. - Enhanced the setup process to automatically copy example files to the project root on first run. - Implemented a migration function to transfer legacy configuration files from the user's home directory to the new project structure. - Revised documentation to reflect changes in file locations and setup instructions. - Improved code readability and maintainability by refactoring path management functions.
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""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)
|