STT-Assistant/src/stt_daemon/state.py

"""Persistence of virtual audio sink state."""

import json
import os
import tempfile
from dataclasses import asdict, dataclass, field
from pathlib import Path


@dataclass
class Loopback:
    """A loopback module that connects a source to the sink."""
    module_id: int
    source: str


@dataclass
class AudioState:
    """Complete state of the provisioned virtual audio sink."""
    sink_name: str
    remap_name: str
    null_sink_module: int
    remap_module: int
    loopbacks: list[Loopback] = field(default_factory=list)
    selected_sources: list[str] = field(default_factory=list)


class StateStore:
    """Persist and restore AudioState from filesystem."""

    def __init__(self, path: Path | None = None):
        """Initialize the state store.

        Args:
            path: Path to the state file. If None, uses `$STT_DAEMON_STATE` or
                `audio.json` relative to the current directory.
        """
        if path is None:
            env_path = os.environ.get("STT_DAEMON_STATE")
            path = Path(env_path) if env_path else Path("audio.json")
        self.path = path

    def load(self) -> AudioState | None:
        """Load state from file.

        Returns:
            AudioState if file exists and is valid JSON, None otherwise.
            Never raises; returns None on missing or corrupt files.
        """
        try:
            if not self.path.exists():
                return None

            with open(self.path, "r", encoding="utf-8") as f:
                data = json.load(f)

            # Reconstruct Loopback objects
            loopbacks = [
                Loopback(
                    module_id=lb["module_id"],
                    source=lb["source"],
                )
                for lb in data.get("loopbacks", [])
            ]

            return AudioState(
                sink_name=data["sink_name"],
                remap_name=data["remap_name"],
                null_sink_module=data["null_sink_module"],
                remap_module=data["remap_module"],
                loopbacks=loopbacks,
                selected_sources=data.get("selected_sources", []),
            )
        except (json.JSONDecodeError, KeyError, TypeError, OSError, ValueError):
            return None

    def save(self, state: AudioState) -> None:
        """Save state to file.

        Creates parent directories if needed. Writes via a temp file and
        atomic rename so a crash mid-write cannot corrupt the state file.

        Args:
            state: AudioState to persist.
        """
        self.path.parent.mkdir(parents=True, exist_ok=True)

        data = asdict(state)
        fd, tmp_path = tempfile.mkstemp(
            dir=self.path.parent, prefix=".audio-", suffix=".tmp"
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=2)
            os.replace(tmp_path, self.path)
        except BaseException:
            try:
                os.unlink(tmp_path)
            except OSError:
                pass
            raise

    def clear(self) -> None:
        """Remove the state file if it exists."""
        if self.path.exists():
            self.path.unlink()