Spaces:

Jindrich3
/

openajaj

Running

openajaj / stt_quality.py

Code-only deploy

42b5eae verified 1 day ago

9.65 kB

"""How well /t sessions went: the page's own report, logged and judged.
The page decides every half second which problem, if any, the visitor sees
(static/transcribe2/index.html, `qualityNow`), and adds up the seconds spent in
each. It sends that tally on page hide and every 5 minutes; each report is the
whole visit so far, so the last one per `sid` is the visit.
Nothing said is in a report: no text, no audio. The client is the same keyed
hash the usage log uses (stt_budget.client_id), never an address.
"""

import json
import logging
import threading
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field, StringConstraints

LOG_FILE = Path("data") / "stt_sessions.jsonl"

# What qualityNow can show; "ok" is time with nothing shown.
Code = Literal["ok", "connect", "off", "mic", "stuck", "slow", "noise", "far", "lang", "music"]
RED = ("off", "mic", "stuck")             # nothing is transcribed, or speech is being lost
AMBER = ("slow", "noise", "far", "lang")  # text comes, but too late or unreliable
# Histogram edges, as the page bins them (LAG_EDGES, CONF_EDGES); the last bin is above
LAG_EDGES_MS = (500, 1000, 2000, 3000, 5000, 10000)
CONF_EDGES = (0.5, 0.6, 0.7, 0.75, 0.8, 0.9)

# Verdict cutoffs. Assumption, not measured: a visit that spent over a fifth
# of its time in a state is described by it. Tune once real visits are in.
SHARE = 0.2
MIN_ACTIVE_S = 15.0  # shorter visits are page checks, kept out of the rates…
# …unless a problem showed this long (3× the page's 1 s hold, Q_HOLD_MS): a visitor
# who leaves over slow or missing text is what the rates are for.
PROBLEM_MIN_S = 3.0
FIRST_CONNECT_S = 10.0  # the page gives a first connect this long (connectWs)

_YEAR_S = 366 * 24 * 3600.0  # a sanity bound: an always-on tab must not hit it
Seconds = Annotated[float, Field(ge=0, le=_YEAR_S)]
Count = Annotated[int, Field(ge=0, le=1_000_000)]
Short = Annotated[str, StringConstraints(max_length=40)]

_log = logging.getLogger(__name__)
_write_lock = threading.Lock()


class SessionReport(BaseModel):
    """One /t page visit so far, as the page reports it."""

    model_config = ConfigDict(extra="forbid")

    sid: Annotated[str, StringConstraints(pattern=r"^[a-z0-9]{8,32}$")]
    secs: dict[Code, Seconds] = Field(max_length=10)
    episodes: dict[Code, Count] = Field(max_length=10)
    off_why: dict[Short, Seconds] = Field(default_factory=dict, max_length=12)
    words: Count = 0
    conf: Annotated[float, Field(ge=0, le=1)] | None = None
    start_ms: Annotated[int, Field(ge=0, le=600_000)] | None = None
    drops: Count = 0
    route: Literal["EU", "US", ""] = ""
    model: Short = ""
    lang: Short = ""
    env: Short = ""
    denoise: bool = True
    display: Literal["cursor", "ball", "off"] = "cursor"
    final: bool = False
    # raw numbers to tune the cutoffs on
    missed_s: Seconds = 0  # mic audio no socket took: drops, restarts, lost queues
    lag_hist: list[Count] = Field(default_factory=lambda: [0] * 7, min_length=7, max_length=7)
    conf_hist: list[Count] = Field(default_factory=lambda: [0] * 7, min_length=7, max_length=7)
    queue_max: Annotated[int, Field(ge=0, le=10**9)] = 0  # bytes stuck in the phone, at worst


def record(report: SessionReport, client: str, vid: str, os_name: str, browser: str) -> None:
    """Append one report. Never raises: a lost report must not break /t.
    `vid` is the visits log's id for the visitor ("" without a captcha token)."""
    line = json.dumps({
        "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "client": client, "vid": vid, "os": os_name, "browser": browser,
        **report.model_dump(),
    })
    try:
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with _write_lock, LOG_FILE.open("a", encoding="utf-8") as file:
            file.write(line + "\n")
    except OSError as error:
        _log.warning("STT session not recorded: %s", error)


def load() -> list[dict]:
    """The last report of each visit, newest visits last."""
    visits: dict[str, dict] = {}
    try:
        with LOG_FILE.open(encoding="utf-8") as file:
            for line in file:
                try:
                    entry = json.loads(line)
                    visits[entry["sid"]] = entry
                except (ValueError, KeyError, TypeError):
                    continue  # a half-written last line, or junk
    except FileNotFoundError:
        return []
    return list(visits.values())


def verdict(visit: dict) -> tuple[str, str]:
    """(verdict, main reason) for one visit.
    failed: Deepgram never answered. broken: over SHARE of the time not
    listening. degraded: over SHARE of it late or unreliable. too_short: under
    MIN_ACTIVE_S with under PROBLEM_MIN_S of problems. The reason is the longest problem of the kind that decided
    the verdict; for ok and too_short, the longest problem of any kind.
    """
    secs = visit.get("secs") or {}
    total = sum(secs.values())

    def longest(codes: tuple[str, ...]) -> str:
        found = {c: secs[c] for c in codes if secs.get(c)}
        return max(found, key=found.__getitem__) if found else ""

    if visit.get("start_ms") is None:
        # left before the page's own first-connect timeout: not yet a failure
        if total < FIRST_CONNECT_S:
            return "too_short", ""
        return "failed", longest(RED) or "off"
    if total < MIN_ACTIVE_S and sum(secs.get(c, 0) for c in RED + AMBER) < PROBLEM_MIN_S:
        return "too_short", longest(RED + AMBER)
    if sum(secs.get(c, 0) for c in RED) > SHARE * total:
        return "broken", longest(RED)
    if sum(secs.get(c, 0) for c in AMBER) > SHARE * total:
        return "degraded", longest(AMBER)
    return "ok", longest(RED + AMBER)


def summarize(visits: list[dict]) -> dict:
    """What /stats shows: verdicts, why, and where problems cluster."""
    verdicts: Counter[str] = Counter()
    reasons: Counter[str] = Counter()
    problem_secs: Counter[str] = Counter()
    total_secs = 0.0
    off_why: Counter[str] = Counter()
    lag_hist, conf_hist = [0] * 7, [0] * 7
    missed = 0.0
    for visit in visits:
        name, reason = verdict(visit)
        verdicts[name] += 1
        if name in ("failed", "broken", "degraded"):
            reasons[reason] += 1
        secs = visit.get("secs") or {}
        total_secs += sum(secs.values())
        problem_secs.update({c: s for c, s in secs.items() if c != "ok"})
        off_why.update(visit.get("off_why") or {})
        missed += visit.get("missed_s") or 0
        for total, counts in ((lag_hist, visit.get("lag_hist")), (conf_hist, visit.get("conf_hist"))):
            for i, n in enumerate((counts or [])[:7]):
                total[i] += n
    return {
        "visits": len(visits),
        "verdicts": dict(verdicts),
        "reasons": dict(reasons.most_common()),
        "problem_secs": {c: round(s) for c, s in problem_secs.most_common()},
        "total_s": round(total_secs),
        "off_why": {why: round(s) for why, s in off_why.most_common()},
        "missed_s": round(missed),
        "lag_hist": lag_hist, "lag_edges_ms": LAG_EDGES_MS,
        "conf_hist": conf_hist, "conf_edges": CONF_EDGES,
    }


def times(visit: dict) -> tuple[str, str]:
    """(start, end) of a visit, ISO UTC, "" when unknown. The page ends the sid
    with its start time (Date.now() in base 36); the end is the last report."""
    end = visit.get("ts", "")
    try:
        start = datetime.fromtimestamp(int(visit["sid"][-8:], 36) / 1000, timezone.utc)
    except (KeyError, ValueError, OverflowError, OSError):
        return "", end
    if start.year < 2025:  # not a time: a sid from before it carried one
        return "", end
    start_iso = start.isoformat(timespec="seconds")
    # the page's clock, not ours: never after the report that closed the visit
    return (min(start_iso, end) if end else start_iso), end


def breakdown(visits: list[dict], numbers: dict[str, int]) -> dict:
    """summarize for all visits, plus "by_os" and "by_visitor": the same per
    group. OSes: most /t time first. Visitors: those with a judged (not
    too_short) visit, most visits first, each with its "sessions" (one per
    visit, newest first, with "start" and "end"). `numbers` maps a visits log
    id to its "Visitor N" on the Visitors tab."""
    # Reports from before "vid" was logged borrow it from a later one of the same client.
    vid_of = {v["client"]: v["vid"] for v in visits if v.get("vid")}
    by_os: dict[str, list[dict]] = defaultdict(list)
    by_who: dict[str, list[dict]] = defaultdict(list)
    for v in visits:
        client = v.get("client", "")
        by_os[v.get("os") or "Unknown"].append(v)
        by_who[v.get("vid") or vid_of.get(client) or client].append(v)
    oses = [{"name": name, **summarize(group)} for name, group in by_os.items()]
    people = [{"name": f"Visitor {numbers[key]}" if key in numbers else f"id {key[:6]}",
               "device": f"{group[-1].get('os', '')} {group[-1].get('browser', '')}".strip(),
               **summarize(group),
               "sessions": [dict(zip(("start", "end"), times(v))) | summarize([v])
                            for v in reversed(group)]}
              for key, group in by_who.items()]
    oses.sort(key=lambda s: -s["total_s"])
    people = sorted((p for p in people if p["visits"] > p["verdicts"].get("too_short", 0)),
                    key=lambda s: (-s["visits"], -s["total_s"]))
    return summarize(visits) | {"by_os": oses, "by_visitor": people}

	"""How well /t sessions went: the page's own report, logged and judged.

	The page decides every half second which problem, if any, the visitor sees
	(static/transcribe2/index.html, `qualityNow`), and adds up the seconds spent in
	each. It sends that tally on page hide and every 5 minutes; each report is the
	whole visit so far, so the last one per `sid` is the visit.

	Nothing said is in a report: no text, no audio. The client is the same keyed
	hash the usage log uses (stt_budget.client_id), never an address.
	"""

	import json
	import logging
	import threading
	from collections import Counter, defaultdict
	from datetime import datetime, timezone
	from pathlib import Path
	from typing import Annotated, Literal

	from pydantic import BaseModel, ConfigDict, Field, StringConstraints

	LOG_FILE = Path("data") / "stt_sessions.jsonl"

	# What qualityNow can show; "ok" is time with nothing shown.
	Code = Literal["ok", "connect", "off", "mic", "stuck", "slow", "noise", "far", "lang", "music"]
	RED = ("off", "mic", "stuck") # nothing is transcribed, or speech is being lost
	AMBER = ("slow", "noise", "far", "lang") # text comes, but too late or unreliable
	# Histogram edges, as the page bins them (LAG_EDGES, CONF_EDGES); the last bin is above
	LAG_EDGES_MS = (500, 1000, 2000, 3000, 5000, 10000)
	CONF_EDGES = (0.5, 0.6, 0.7, 0.75, 0.8, 0.9)

	# Verdict cutoffs. Assumption, not measured: a visit that spent over a fifth
	# of its time in a state is described by it. Tune once real visits are in.
	SHARE = 0.2
	MIN_ACTIVE_S = 15.0 # shorter visits are page checks, kept out of the rates…
	# …unless a problem showed this long (3× the page's 1 s hold, Q_HOLD_MS): a visitor
	# who leaves over slow or missing text is what the rates are for.
	PROBLEM_MIN_S = 3.0
	FIRST_CONNECT_S = 10.0 # the page gives a first connect this long (connectWs)

	_YEAR_S = 366 * 24 * 3600.0 # a sanity bound: an always-on tab must not hit it
	Seconds = Annotated[float, Field(ge=0, le=_YEAR_S)]
	Count = Annotated[int, Field(ge=0, le=1_000_000)]
	Short = Annotated[str, StringConstraints(max_length=40)]

	_log = logging.getLogger(__name__)
	_write_lock = threading.Lock()


	class SessionReport(BaseModel):
	"""One /t page visit so far, as the page reports it."""

	model_config = ConfigDict(extra="forbid")

	sid: Annotated[str, StringConstraints(pattern=r"^[a-z0-9]{8,32}$")]
	secs: dict[Code, Seconds] = Field(max_length=10)
	episodes: dict[Code, Count] = Field(max_length=10)
	off_why: dict[Short, Seconds] = Field(default_factory=dict, max_length=12)
	words: Count = 0
	conf: Annotated[float, Field(ge=0, le=1)] \| None = None
	start_ms: Annotated[int, Field(ge=0, le=600_000)] \| None = None
	drops: Count = 0
	route: Literal["EU", "US", ""] = ""
	model: Short = ""
	lang: Short = ""
	env: Short = ""
	denoise: bool = True
	display: Literal["cursor", "ball", "off"] = "cursor"
	final: bool = False
	# raw numbers to tune the cutoffs on
	missed_s: Seconds = 0 # mic audio no socket took: drops, restarts, lost queues
	lag_hist: list[Count] = Field(default_factory=lambda: [0] * 7, min_length=7, max_length=7)
	conf_hist: list[Count] = Field(default_factory=lambda: [0] * 7, min_length=7, max_length=7)
	queue_max: Annotated[int, Field(ge=0, le=10**9)] = 0 # bytes stuck in the phone, at worst


	def record(report: SessionReport, client: str, vid: str, os_name: str, browser: str) -> None:
	"""Append one report. Never raises: a lost report must not break /t.
	`vid` is the visits log's id for the visitor ("" without a captcha token)."""
	line = json.dumps({
	"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
	"client": client, "vid": vid, "os": os_name, "browser": browser,
	**report.model_dump(),
	})
	try:
	LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
	with _write_lock, LOG_FILE.open("a", encoding="utf-8") as file:
	file.write(line + "\n")
	except OSError as error:
	_log.warning("STT session not recorded: %s", error)


	def load() -> list[dict]:
	"""The last report of each visit, newest visits last."""
	visits: dict[str, dict] = {}
	try:
	with LOG_FILE.open(encoding="utf-8") as file:
	for line in file:
	try:
	entry = json.loads(line)
	visits[entry["sid"]] = entry
	except (ValueError, KeyError, TypeError):
	continue # a half-written last line, or junk
	except FileNotFoundError:
	return []
	return list(visits.values())


	def verdict(visit: dict) -> tuple[str, str]:
	"""(verdict, main reason) for one visit.

	failed: Deepgram never answered. broken: over SHARE of the time not
	listening. degraded: over SHARE of it late or unreliable. too_short: under
	MIN_ACTIVE_S with under PROBLEM_MIN_S of problems. The reason is the longest problem of the kind that decided
	the verdict; for ok and too_short, the longest problem of any kind.
	"""
	secs = visit.get("secs") or {}
	total = sum(secs.values())

	def longest(codes: tuple[str, ...]) -> str:
	found = {c: secs[c] for c in codes if secs.get(c)}
	return max(found, key=found.__getitem__) if found else ""

	if visit.get("start_ms") is None:
	# left before the page's own first-connect timeout: not yet a failure
	if total < FIRST_CONNECT_S:
	return "too_short", ""
	return "failed", longest(RED) or "off"
	if total < MIN_ACTIVE_S and sum(secs.get(c, 0) for c in RED + AMBER) < PROBLEM_MIN_S:
	return "too_short", longest(RED + AMBER)
	if sum(secs.get(c, 0) for c in RED) > SHARE * total:
	return "broken", longest(RED)
	if sum(secs.get(c, 0) for c in AMBER) > SHARE * total:
	return "degraded", longest(AMBER)
	return "ok", longest(RED + AMBER)


	def summarize(visits: list[dict]) -> dict:
	"""What /stats shows: verdicts, why, and where problems cluster."""
	verdicts: Counter[str] = Counter()
	reasons: Counter[str] = Counter()
	problem_secs: Counter[str] = Counter()
	total_secs = 0.0
	off_why: Counter[str] = Counter()
	lag_hist, conf_hist = [0] * 7, [0] * 7
	missed = 0.0
	for visit in visits:
	name, reason = verdict(visit)
	verdicts[name] += 1
	if name in ("failed", "broken", "degraded"):
	reasons[reason] += 1
	secs = visit.get("secs") or {}
	total_secs += sum(secs.values())
	problem_secs.update({c: s for c, s in secs.items() if c != "ok"})
	off_why.update(visit.get("off_why") or {})
	missed += visit.get("missed_s") or 0
	for total, counts in ((lag_hist, visit.get("lag_hist")), (conf_hist, visit.get("conf_hist"))):
	for i, n in enumerate((counts or [])[:7]):
	total[i] += n
	return {
	"visits": len(visits),
	"verdicts": dict(verdicts),
	"reasons": dict(reasons.most_common()),
	"problem_secs": {c: round(s) for c, s in problem_secs.most_common()},
	"total_s": round(total_secs),
	"off_why": {why: round(s) for why, s in off_why.most_common()},
	"missed_s": round(missed),
	"lag_hist": lag_hist, "lag_edges_ms": LAG_EDGES_MS,
	"conf_hist": conf_hist, "conf_edges": CONF_EDGES,
	}


	def times(visit: dict) -> tuple[str, str]:
	"""(start, end) of a visit, ISO UTC, "" when unknown. The page ends the sid
	with its start time (Date.now() in base 36); the end is the last report."""
	end = visit.get("ts", "")
	try:
	start = datetime.fromtimestamp(int(visit["sid"][-8:], 36) / 1000, timezone.utc)
	except (KeyError, ValueError, OverflowError, OSError):
	return "", end
	if start.year < 2025: # not a time: a sid from before it carried one
	return "", end
	start_iso = start.isoformat(timespec="seconds")
	# the page's clock, not ours: never after the report that closed the visit
	return (min(start_iso, end) if end else start_iso), end


	def breakdown(visits: list[dict], numbers: dict[str, int]) -> dict:
	"""summarize for all visits, plus "by_os" and "by_visitor": the same per
	group. OSes: most /t time first. Visitors: those with a judged (not
	too_short) visit, most visits first, each with its "sessions" (one per
	visit, newest first, with "start" and "end"). `numbers` maps a visits log
	id to its "Visitor N" on the Visitors tab."""
	# Reports from before "vid" was logged borrow it from a later one of the same client.
	vid_of = {v["client"]: v["vid"] for v in visits if v.get("vid")}
	by_os: dict[str, list[dict]] = defaultdict(list)
	by_who: dict[str, list[dict]] = defaultdict(list)
	for v in visits:
	client = v.get("client", "")
	by_os[v.get("os") or "Unknown"].append(v)
	by_who[v.get("vid") or vid_of.get(client) or client].append(v)
	oses = [{"name": name, **summarize(group)} for name, group in by_os.items()]
	people = [{"name": f"Visitor {numbers[key]}" if key in numbers else f"id {key[:6]}",
	"device": f"{group[-1].get('os', '')} {group[-1].get('browser', '')}".strip(),
	**summarize(group),
	"sessions": [dict(zip(("start", "end"), times(v))) \| summarize([v])
	for v in reversed(group)]}
	for key, group in by_who.items()]
	oses.sort(key=lambda s: -s["total_s"])
	people = sorted((p for p in people if p["visits"] > p["verdicts"].get("too_short", 0)),
	key=lambda s: (-s["visits"], -s["total_s"]))
	return summarize(visits) \| {"by_os": oses, "by_visitor": people}