|
|
@@ -1,57 +1,39 @@
|
|
|
"""
|
|
|
Transcript Processor — deduplicates, stitches overlapping ASR windows,
|
|
|
detects utterance boundaries, and formats streaming output for the client.
|
|
|
-
|
|
|
-Flow:
|
|
|
- raw ASR transcript (sliding window, every ~2 s)
|
|
|
- │
|
|
|
- ▼
|
|
|
- _stitch() — overlap-merge with accumulated buffer
|
|
|
- │
|
|
|
- ▼
|
|
|
- _detect_boundary() — same text repeated = silence = utterance end
|
|
|
- │
|
|
|
- ├── partial → {"type": "partial", "text": accumulated, "seq": N}
|
|
|
- │
|
|
|
- └── (after 3× same text + 3 s silence)
|
|
|
- utterance → {"type": "utterance", "text": full, "seq": N}
|
|
|
"""
|
|
|
|
|
|
import time
|
|
|
from dataclasses import dataclass, field
|
|
|
-from typing import Optional, Callable, Awaitable
|
|
|
+from typing import Optional, List, Callable, Awaitable
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class _Utterance:
|
|
|
- """Tracks a single utterance being built."""
|
|
|
accumulated: str = ""
|
|
|
last_raw: str = ""
|
|
|
same_count: int = 0
|
|
|
silence_since: float = 0.0
|
|
|
- finalized: bool = False
|
|
|
- seq: int = 0 # monotonic counter for this utterance
|
|
|
|
|
|
|
|
|
class TranscriptProcessor:
|
|
|
"""
|
|
|
Processes streaming ASR transcripts into clean, segmented utterances.
|
|
|
|
|
|
- Pipeline (extensible):
|
|
|
- 1. stitch — overlap-based deduplication
|
|
|
- 2. boundary — silence-based utterance segmentation
|
|
|
- 3. post — (future) LLM-based punctuation/correction
|
|
|
+ Each call to feed() returns a list of zero or more message dicts:
|
|
|
+ {"type": "partial", "text": "...", "seq": N}
|
|
|
+ {"type": "utterance", "text": "...", "seq": N}
|
|
|
"""
|
|
|
|
|
|
def __init__(
|
|
|
self,
|
|
|
- silence_repeats: int = 3, # same-raw-text count before silence kicks in
|
|
|
- silence_seconds: float = 3.0, # timer after silence detection
|
|
|
+ silence_repeats: int = 3,
|
|
|
+ silence_seconds: float = 3.0,
|
|
|
post_hook: Optional[Callable[[str], Awaitable[str]]] = None,
|
|
|
):
|
|
|
self.silence_repeats = silence_repeats
|
|
|
self.silence_seconds = silence_seconds
|
|
|
- self._post_hook = post_hook # LLM correction hook (future)
|
|
|
+ self._post_hook = post_hook
|
|
|
|
|
|
self._current: Optional[_Utterance] = None
|
|
|
self._global_seq = 0
|
|
|
@@ -60,21 +42,28 @@ class TranscriptProcessor:
|
|
|
# Public API
|
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
- def feed(self, raw_text: str) -> dict | None:
|
|
|
- """
|
|
|
- Feed a raw ASR transcript. Returns a message dict to send to the
|
|
|
- client, or None if nothing should be sent this tick.
|
|
|
-
|
|
|
- Message shapes:
|
|
|
- {"type": "partial", "text": "...", "seq": N}
|
|
|
- {"type": "utterance", "text": "...", "seq": N}
|
|
|
- """
|
|
|
+ def feed(self, raw_text: str) -> List[dict]:
|
|
|
+ """Feed a raw ASR transcript. Returns 0..N messages to send."""
|
|
|
raw_text = raw_text.strip()
|
|
|
- self._ensure_utterance()
|
|
|
+ if not raw_text:
|
|
|
+ return []
|
|
|
+
|
|
|
+ messages: List[dict] = []
|
|
|
+
|
|
|
+ # ---- ensure we have an utterance context ----
|
|
|
+ if self._current is None:
|
|
|
+ self._current = _Utterance()
|
|
|
|
|
|
# ---- 1. stitch ----
|
|
|
if self._current.accumulated:
|
|
|
- merged = self._stitch(self._current.accumulated, raw_text)
|
|
|
+ merged, flush_msg = self._stitch(
|
|
|
+ self._current.accumulated, raw_text
|
|
|
+ )
|
|
|
+ if flush_msg is not None:
|
|
|
+ messages.append(flush_msg)
|
|
|
+ # _stitch may have flushed (self._current = None). Restore.
|
|
|
+ if self._current is None:
|
|
|
+ self._current = _Utterance()
|
|
|
else:
|
|
|
merged = raw_text
|
|
|
|
|
|
@@ -89,62 +78,74 @@ class TranscriptProcessor:
|
|
|
else:
|
|
|
self._current.same_count = 0
|
|
|
self._current.silence_since = 0.0
|
|
|
- self._global_seq += 1
|
|
|
self._current.last_raw = raw_text
|
|
|
- return {"type": "partial", "text": merged, "seq": self._global_seq}
|
|
|
+ self._global_seq += 1
|
|
|
+ messages.append({
|
|
|
+ "type": "partial",
|
|
|
+ "text": merged,
|
|
|
+ "seq": self._global_seq,
|
|
|
+ })
|
|
|
|
|
|
# ---- 3. silence timer ----
|
|
|
if (
|
|
|
self._current.silence_since > 0
|
|
|
- and time.monotonic() - self._current.silence_since >= self.silence_seconds
|
|
|
+ and time.monotonic() - self._current.silence_since
|
|
|
+ >= self.silence_seconds
|
|
|
):
|
|
|
- return self._finalize()
|
|
|
+ msg = self._flush()
|
|
|
+ if msg:
|
|
|
+ messages.append(msg)
|
|
|
|
|
|
- return None
|
|
|
+ return messages
|
|
|
|
|
|
- def finalize(self) -> dict | None:
|
|
|
- """Force-finalize the current utterance (e.g. on pause/end)."""
|
|
|
- self._ensure_utterance()
|
|
|
- return self._finalize()
|
|
|
+ def finalize(self) -> List[dict]:
|
|
|
+ """Force-finalize (pause/end). Returns 0..1 messages."""
|
|
|
+ return self._finish()
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
|
# Internals
|
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
- def _ensure_utterance(self):
|
|
|
- if self._current is None:
|
|
|
- self._current = _Utterance(seq=self._global_seq)
|
|
|
-
|
|
|
- def _stitch(self, previous: str, next_text: str) -> str:
|
|
|
+ def _stitch(
|
|
|
+ self, previous: str, next_text: str
|
|
|
+ ) -> tuple[str, Optional[dict]]:
|
|
|
"""
|
|
|
- Stitch two overlapping transcripts by finding the longest suffix of
|
|
|
- *previous* that matches a prefix of *next_text*, then appending only
|
|
|
- the non-overlapping suffix. If there is no overlap, *next_text* is a
|
|
|
- fresh utterance — finalize current and return *next_text* alone.
|
|
|
+ Merge next_text into previous using longest suffix–prefix overlap.
|
|
|
+ Returns (merged_text, pending_utterance_message_or_None).
|
|
|
+ The pending message contains the OLD utterance if there is no
|
|
|
+ overlap with previous — i.e. the ASR has started a new utterance.
|
|
|
"""
|
|
|
if not next_text:
|
|
|
- return previous
|
|
|
+ return previous, None
|
|
|
if next_text.startswith(previous):
|
|
|
- return next_text
|
|
|
+ return next_text, None
|
|
|
|
|
|
# Longest suffix–prefix overlap
|
|
|
for n in range(len(next_text), 0, -1):
|
|
|
needle = next_text[:n]
|
|
|
if previous.endswith(needle):
|
|
|
- return previous + next_text[n:]
|
|
|
+ return previous + next_text[n:], None
|
|
|
|
|
|
- # No overlap → new utterance
|
|
|
- self._finalize()
|
|
|
- self._current = _Utterance(seq=self._global_seq)
|
|
|
- return next_text
|
|
|
+ # No overlap at all — previous utterance is finished.
|
|
|
+ flush = self._flush()
|
|
|
+ return next_text, flush
|
|
|
|
|
|
- def _finalize(self) -> dict | None:
|
|
|
- """Finalize current utterance and return the utterance message."""
|
|
|
+ def _flush(self) -> Optional[dict]:
|
|
|
+ """Flush current utterance. Returns utterance message or None."""
|
|
|
if self._current is None:
|
|
|
return None
|
|
|
text = self._current.accumulated.strip()
|
|
|
self._current = None
|
|
|
- self._global_seq += 1
|
|
|
if not text:
|
|
|
return None
|
|
|
+ self._global_seq += 1
|
|
|
return {"type": "utterance", "text": text, "seq": self._global_seq}
|
|
|
+
|
|
|
+ def _finish(self) -> List[dict]:
|
|
|
+ """Finalize and clean up. Returns 0..1 messages."""
|
|
|
+ messages: List[dict] = []
|
|
|
+ msg = self._flush()
|
|
|
+ if msg:
|
|
|
+ messages.append(msg)
|
|
|
+ self._current = None
|
|
|
+ return messages
|