File size: 6,125 Bytes
a3847b2 | 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """
Pydantic models for runtime state: streak store, analysis results, shuffler.
"""
from __future__ import annotations
import datetime
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field, field_validator, model_validator
# ── Module keys (matches original JS constants) ───────────────────────────────
class ModuleKey(str, Enum):
JAM = "jam"
TT = "tt"
IMPROMPTU = "impromptu"
INTERVIEW = "interview"
OTHER = "other"
FREE_MODULE_MAP: dict[str, ModuleKey] = {
"JAM": ModuleKey.JAM,
"Tongue Twister": ModuleKey.TT,
"Impromptu": ModuleKey.IMPROMPTU,
"Interview": ModuleKey.INTERVIEW,
}
class FreeModuleChoice(str, Enum):
JAM = "JAM"
TT = "Tongue Twister"
IMPROMPTU = "Impromptu"
INTERVIEW = "Interview"
# ── Per-module lifetime stats ─────────────────────────────────────────────────
class ModuleStats(BaseModel):
count: int = 0
totalDuration: int = 0 # seconds — matches JS field name for BrowserState compat
@property
def avg_duration(self) -> int:
return round(self.totalDuration / self.count) if self.count else 0
def avg_str(self) -> str:
avg = self.avg_duration
if not avg:
return "–"
return f"{avg // 60}:{avg % 60:02d}"
# ── Per-day entry ─────────────────────────────────────────────────────────────
class DayEntry(BaseModel):
total: int = 0
modules: dict[str, int] = Field(default_factory=dict) # module_key → session count that day
@field_validator("total")
@classmethod
def total_non_negative(cls, v: int) -> int:
if v < 0:
raise ValueError("total must be >= 0")
return v
# ── Full streak store (mirrors localStorage schema v2) ────────────────────────
class StreakStore(BaseModel):
v: int = 2
days: dict[str, DayEntry] = Field(default_factory=dict) # ISO date str → DayEntry
modules: dict[str, ModuleStats] = Field(default_factory=dict) # ModuleKey value → ModuleStats
longest: int = 0
@model_validator(mode="before")
@classmethod
def migrate_v1(cls, data: Any) -> Any:
"""Migrate v1 stores (days[k] = int) to v2 (days[k] = DayEntry)."""
if not isinstance(data, dict):
return data
raw_days = data.get("days", {})
if isinstance(raw_days, dict):
migrated = {}
for k, v in raw_days.items():
if isinstance(v, int):
migrated[k] = {"total": v, "modules": {}}
else:
migrated[k] = v
data["days"] = migrated
return data
def day_total(self, key: str) -> int:
entry = self.days.get(key)
if entry is None:
return 0
return entry.total
def calc_streak(self) -> int:
streak = 0
d = datetime.date.today()
if not self.day_total(d.isoformat()):
d -= datetime.timedelta(days=1)
while self.day_total(d.isoformat()):
streak += 1
d -= datetime.timedelta(days=1)
return streak
def calc_longest(self) -> int:
keys = sorted(k for k in self.days if self.day_total(k))
best = run = 0
prev: datetime.date | None = None
for k in keys:
cur = datetime.date.fromisoformat(k)
if prev:
run = run + 1 if (cur - prev).days == 1 else 1
else:
run = 1
best = max(best, run)
prev = cur
return best
def record_session(self, module_key: ModuleKey, duration_sec: int = 0) -> "StreakStore":
"""Immutable-style: returns updated copy."""
data = self.model_copy(deep=True)
today = datetime.date.today().isoformat()
entry = data.days.get(today) or DayEntry()
entry.total += 1
mk = module_key.value
entry.modules[mk] = entry.modules.get(mk, 0) + 1
data.days[today] = entry
stats = data.modules.get(mk) or ModuleStats()
stats.count += 1
stats.totalDuration += max(0, duration_sec)
data.modules[mk] = stats
data.longest = max(data.longest, data.calc_longest())
# Prune entries older than 400 days
cutoff = (datetime.date.today() - datetime.timedelta(days=400)).isoformat()
data.days = {k: v for k, v in data.days.items() if k >= cutoff}
return data
@classmethod
def from_raw(cls, raw: dict | None) -> "StreakStore":
"""Safely deserialise from BrowserState dict, falling back to empty."""
if not isinstance(raw, dict):
return cls()
try:
return cls.model_validate(raw)
except Exception:
return cls()
def to_raw(self) -> dict:
"""Serialise to plain dict for BrowserState storage."""
return self.model_dump()
# ── No-repeat shuffler state (persisted per-key in BrowserState) ──────────────
class ShufflerState(BaseModel):
seen: list[str] = Field(default_factory=list) # serialised items already shown this cycle
# ── Analysis result ───────────────────────────────────────────────────────────
class AnalysisResult(BaseModel):
transcript: str | None = None
pronunciation: int | None = None # 0-100
fluency: int | None = None # 0-100
error: str | None = None
@field_validator("pronunciation", "fluency", mode="before")
@classmethod
def clamp_score(cls, v: int | None) -> int | None:
if v is None:
return None
return max(0, min(100, v))
|