Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

session_logger

Per-session JSONL event logger.

Each session gets its own JSONL file under the configured log root so post-hoc analytics (enhancer latency, GPU assignment, segment timings) can be recovered without a tracing backend. The internal UI uses this format; keeping the same shape makes log tooling portable.

fastvideo.entrypoints.streaming.session_logger.SessionLogEvent

SessionLogEvent(session_id: str, event: str, payload: dict[str, Any] = dict(), ts: float = time.time())

One line in the session JSONL file.

Source
@dataclass
class SessionLogEvent:
    """One line in the session JSONL file."""

session_id: str
event: str
payload: dict[str, Any] = field(default_factory=dict)
ts: float = field(default_factory=time.time)</code></pre></details>


fastvideo.entrypoints.streaming.session_logger.SessionLogger


SessionLogger(log_dir: str | None)


Append-only JSONL logger keyed by session id.


Thread-safe; the server may be writing from multiple asyncio tasks
(fMP4 encoder thread + control-frame handler) for the same session.


Source
class SessionLogger:
    """Append-only JSONL logger keyed by session id.

Thread-safe; the server may be writing from multiple asyncio tasks
(fMP4 encoder thread + control-frame handler) for the same session.
&quot;&quot;&quot;

def __init__(self, log_dir: str | None) -&gt; None:
    self._log_dir = log_dir
    self._files: dict&#91;str, TextIO&#93; = &#123;&#125;
    self._locks: dict&#91;str, threading.Lock&#93; = &#123;&#125;
    self._registry_lock = threading.Lock()
    self._ensure_dir()

def log(self, event: SessionLogEvent) -&gt; None:
    if self._log_dir is None:
        return
    opened = self._get_file(event.session_id)
    if opened is None:
        return
    handle, lock = opened
    line = json.dumps(&#123;
        &quot;session_id&quot;: event.session_id,
        &quot;event&quot;: event.event,
        &quot;ts&quot;: event.ts,
        &quot;payload&quot;: event.payload,
    &#125;)
    with lock, contextlib.suppress(ValueError):
        handle.write(line + &quot;\n&quot;)
        handle.flush()

def close(self, session_id: str) -&gt; None:
    with self._registry_lock:
        handle = self._files.pop(session_id, None)
        lock = self._locks.pop(session_id, None)
    if handle is None or lock is None:
        return
    with lock, contextlib.suppress(Exception):
        handle.close()

def close_all(self) -&gt; None:
    with self._registry_lock:
        sids = list(self._files)
    for sid in sids:
        self.close(sid)

def _ensure_dir(self) -&gt; None:
    if self._log_dir is None:
        return
    os.makedirs(self._log_dir, exist_ok=True)

def _get_file(self, session_id: str) -&gt; tuple&#91;TextIO, threading.Lock&#93; | None:
    if self._log_dir is None:
        return None
    with self._registry_lock:
        handle = self._files.get(session_id)
        lock = self._locks.get(session_id)
        if handle is not None and lock is not None:
            return handle, lock
        # Defense-in-depth: session_id is server-generated UUID today,
        # but sanitize against path traversal in case future code paths
        # allow client-supplied ids.
        safe_id = _FILENAME_SANITIZE_RE.sub(&quot;_&quot;, session_id) or &quot;unknown&quot;
        path = os.path.join(
            self._log_dir,
            f&quot;session-&#123;safe_id&#125;.jsonl&quot;,
        )
        try:
            handle = open(path, &quot;a&quot;, encoding=&quot;utf-8&quot;)  # noqa: SIM115
        except OSError:
            return None
        lock = threading.Lock()
        self._files&#91;session_id&#93; = handle
        self._locks&#91;session_id&#93; = lock
        return handle, lock</code></pre></details>


fastvideo.entrypoints.streaming.session_logger.SessionLogger.log


log(event: SessionLogEvent) -> None


Source
def log(self, event: SessionLogEvent) -> None:
    if self._log_dir is None:
        return
    opened = self._get_file(event.session_id)
    if opened is None:
        return
    handle, lock = opened
    line = json.dumps({
        "session_id": event.session_id,
        "event": event.event,
        "ts": event.ts,
        "payload": event.payload,
    })
    with lock, contextlib.suppress(ValueError):
        handle.write(line + "\n")
        handle.flush()


fastvideo.entrypoints.streaming.session_logger.SessionLogger.close


close(session_id: str) -> None


Source
def close(self, session_id: str) -> None:
    with self._registry_lock:
        handle = self._files.pop(session_id, None)
        lock = self._locks.pop(session_id, None)
    if handle is None or lock is None:
        return
    with lock, contextlib.suppress(Exception):
        handle.close()


fastvideo.entrypoints.streaming.session_logger.SessionLogger.close_all


close_all() -> None


Source
def close_all(self) -> None:
    with self._registry_lock:
        sids = list(self._files)
    for sid in sids:
        self.close(sid)

Was this page helpful?