Real-Time Quality & Compliance Monitoring

GNANI_API_KEY=your-api-key-here
ALERT_WEBHOOK_URL=https://supervisor.internal/alerts
LANG_CODE=hi-IN
SAMPLE_RATE=16000

import asyncio, os
from gnani.stt import GnaniSTTStreamClient

async def open_stream():
    async with GnaniSTTStreamClient(
        api_key=os.getenv("GNANI_API_KEY"),
        language_code="hi-IN",     # or comma-separated for auto-detect
        sample_rate=16000,
    ) as stream:
        async for event in stream:
            await handle_event(event)

import asyncio

FRAME_SIZE   = 1024   # bytes — exactly 512 x 16-bit samples
FRAME_MS_16K = 0.032  # 32ms per frame at 16kHz
FRAME_MS_8K  = 0.064  # 64ms per frame at 8kHz

async def stream_audio_producer(stream, audio_source, sample_rate=16000, stop_event=None):
    frame_interval = FRAME_MS_16K if sample_rate == 16000 else FRAME_MS_8K
    buffer = bytearray()

    async for chunk in audio_source:
        if stop_event and stop_event.is_set(): break
        buffer.extend(chunk)
        while len(buffer) >= FRAME_SIZE:
            await stream.send_audio(bytes(buffer[:FRAME_SIZE]))
            buffer = buffer[FRAME_SIZE:]
            await asyncio.sleep(frame_interval)  # enforce real-time cadence

    # Flush remaining partial frame padded with silence
    if buffer:
        await stream.send_audio(bytes(buffer) + b"\x00" * (FRAME_SIZE - len(buffer)))

{
  "type": "transcript",
  "timestamp": "2024-01-15T10:30:05.987Z",
  "text": "guaranteed returns milenge, bilkul risk-free hai",
  "audio_duration_ms": 2340,
  "segment_id": "seg_7f3a92",
  "segment_index": 4,
  "latency": 318
}

{
  "prohibited_keywords": [
    {
      "rule_id": "PROH_001", "severity": "CRITICAL",
      "keywords": ["guaranteed returns", "guaranteed profit", "no risk", "risk-free"],
      "description": "SEBI-prohibited investment language"
    },
    {
      "rule_id": "PROH_002", "severity": "CRITICAL",
      "keywords": ["personal account", "off the books", "my account"],
      "description": "Agent directing customer to off-channel transaction"
    }
  ],
  "risk_phrases": [
    {
      "rule_id": "RISK_001", "severity": "WARNING",
      "phrases": ["cancel my policy", "close my account", "policy cancel"],
      "description": "Churn risk signal"
    },
    {
      "rule_id": "RISK_002", "severity": "WARNING",
      "phrases": ["legal action", "consumer forum", "RBI complaint", "complaint"],
      "description": "Regulatory complaint intent"
    }
  ],
  "required_disclosures": [
    {
      "rule_id": "DISC_001", "severity": "CRITICAL",
      "must_contain_one_of": ["this call is being recorded", "call recording", "recorded for quality"],
      "check_within_segments": 3,
      "description": "Recording disclosure required within first 3 segments"
    }
  ]
}

import json
from pathlib import Path
from typing import List, Dict

class ComplianceEngine:
    def __init__(self, rules_path="rules/compliance.json"):
        rules = json.loads(Path(rules_path).read_text())
        self.prohibited   = rules.get("prohibited_keywords", [])
        self.risk_phrases = rules.get("risk_phrases", [])
        self.disclosures  = rules.get("required_disclosures", [])
        self._disclosed   = set()

    def check(self, segment) -> List[Dict]:
        text = segment.text.lower()
        hits = []

        for rule in self.prohibited:
            for kw in rule["keywords"]:
                if kw in text:
                    hits.append({"rule_id": rule["rule_id"], "severity": rule["severity"],
                                  "matched": kw, "description": rule["description"],
                                  "segment_idx": segment.segment_index, "text": segment.text})
                    break

        for rule in self.risk_phrases:
            for phrase in rule["phrases"]:
                if phrase in text:
                    hits.append({"rule_id": rule["rule_id"], "severity": rule["severity"],
                                  "matched": phrase, "description": rule["description"],
                                  "segment_idx": segment.segment_index, "text": segment.text})
                    break

        for rule in self.disclosures:
            rid = rule["rule_id"]
            if rid in self._disclosed: continue
            if any(p in text for p in rule["must_contain_one_of"]):
                self._disclosed.add(rid)
            elif segment.segment_index >= rule["check_within_segments"]:
                hits.append({"rule_id": rid, "severity": rule["severity"],
                              "matched": "MISSING_DISCLOSURE", "description": rule["description"],
                              "segment_idx": segment.segment_index, "text": ""})
                self._disclosed.add(rid)

        return hits

{
  "silence": { "threshold_seconds": 8 },
  "escalation_phrases": [
    "transfer to supervisor", "let me escalate", "i will get my supervisor"
  ],
  "interruption": { "min_duration_ms": 300 },
  "short_segment_ms": 500
}

import json
from datetime import datetime, timezone
from pathlib  import Path
from typing   import List, Dict, Optional

class QualityEngine:
    def __init__(self, rules_path="rules/quality.json"):
        rules = json.loads(Path(rules_path).read_text())
        self.silence_threshold  = rules["silence"]["threshold_seconds"]
        self.escalation_phrases = [p.lower() for p in rules["escalation_phrases"]]
        self.interruption_ms    = rules["interruption"]["min_duration_ms"]
        self.short_segment_ms   = rules["short_segment_ms"]
        self._last_processing_ts: Optional[datetime] = None

    def on_processing(self, timestamp_str: str):
        self._last_processing_ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))

    def check(self, session, segment) -> List[Dict]:
        now, text, events = datetime.now(timezone.utc), segment.text.lower(), []

        if session.last_segment_end:
            silence_s = (now - session.last_segment_end).total_seconds() - (segment.audio_duration_ms / 1000)
            if silence_s > self.silence_threshold:
                events.append({"event_type": "SILENCE", "severity": "WARNING",
                                "silence_s": round(silence_s, 1), "segment_idx": segment.segment_index,
                                "description": f"Silence gap of {silence_s:.1f}s detected"})

        for phrase in self.escalation_phrases:
            if phrase in text:
                events.append({"event_type": "ESCALATION", "severity": "WARNING",
                                "matched": phrase, "segment_idx": segment.segment_index,
                                "description": "Supervisor escalation signal"})
                break

        if self._last_processing_ts and segment.audio_duration_ms < self.short_segment_ms:
            gap_ms = (now - self._last_processing_ts).total_seconds() * 1000
            if gap_ms < self.interruption_ms:
                events.append({"event_type": "INTERRUPTION", "severity": "INFO",
                                "gap_ms": round(gap_ms, 1), "segment_idx": segment.segment_index,
                                "description": f"Possible interruption — {gap_ms:.0f}ms gap"})

        return events

import asyncio, random, os
from gnani.stt import GnaniSTTStreamClient, StreamConnectionError, StreamClosedError, StreamError

MAX_RECONNECTS = 5
BASE_BACKOFF_S = 1.0
MAX_BACKOFF_S  = 30.0

async def monitor_call_with_reconnect(session, audio_source, compliance_engine, quality_engine, alert_dispatcher):
    attempt = 0

    while attempt <= MAX_RECONNECTS:
        try:
            async with GnaniSTTStreamClient(
                api_key=os.getenv("GNANI_API_KEY"),
                language_code=session.language_code,
                sample_rate=int(os.getenv("SAMPLE_RATE", "16000")),
            ) as stream:
                if attempt > 0: session.reconnect_count += 1
                attempt = 0  # reset backoff counter on successful connect

                stop_event = asyncio.Event()
                producer   = asyncio.create_task(stream_audio_producer(stream, audio_source, stop_event=stop_event))

                async for event in stream:
                    await handle_event(session, event, compliance_engine, quality_engine, alert_dispatcher)

                stop_event.set()
                await producer
                return  # clean exit

        except StreamConnectionError:
            print(f"[{session.call_id}] Auth failure. Not retrying.")
            raise

        except (StreamClosedError, ConnectionResetError, OSError) as e:
            attempt += 1
            if attempt > MAX_RECONNECTS: raise
            backoff = min(BASE_BACKOFF_S * (2 ** attempt), MAX_BACKOFF_S)
            jitter  = random.uniform(0, backoff * 0.2)
            print(f"[{session.call_id}] Reconnect {attempt}/{MAX_RECONNECTS} in {backoff+jitter:.1f}s")
            await asyncio.sleep(backoff + jitter)

"""
monitor.py — Real-time quality and compliance monitoring pipeline.

Usage:
    GNANI_API_KEY=your-key python monitor.py --audio call.pcm --lang hi-IN
    GNANI_API_KEY=your-key python monitor.py --audio call.pcm --lang en-IN --rate 8000

Install:
    pip install gnani-vachana aiohttp python-dotenv
"""

import asyncio, json, os, random, argparse, aiohttp
from dataclasses import dataclass, field
from datetime   import datetime, timezone
from pathlib    import Path
from typing     import List, Dict, Optional, Set, AsyncIterator
from dotenv     import load_dotenv
from gnani.stt  import (
    GnaniSTTStreamClient,
    StreamConnectedEvent, StreamProcessingEvent,
    StreamTranscriptEvent, StreamErrorEvent,
    StreamConnectionError, StreamClosedError, StreamError,
)
load_dotenv()

FRAME_SIZE     = 1024
MAX_RECONNECTS = 5
BASE_BACKOFF_S = 1.0
MAX_BACKOFF_S  = 30.0
FRAME_MS_16K   = 0.032
FRAME_MS_8K    = 0.064


@dataclass
class TranscriptSegment:
    segment_index:     int
    text:              str
    audio_duration_ms: int
    latency_ms:        int
    timestamp:         datetime
    compliance_flags:  List[str] = field(default_factory=list)
    quality_flags:     List[str] = field(default_factory=list)

@dataclass
class CallSession:
    call_id:           str
    language_code:     str
    started_at:        datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    segments:          List[TranscriptSegment] = field(default_factory=list)
    last_segment_end:  Optional[datetime] = None
    reconnect_count:   int = 0
    processed_indices: Set[int] = field(default_factory=set)

    def add_segment(self, event) -> TranscriptSegment:
        seg = TranscriptSegment(
            segment_index=event.segment_index, text=event.text,
            audio_duration_ms=event.audio_duration_ms, latency_ms=event.latency,
            timestamp=datetime.now(timezone.utc),
        )
        self.segments.append(seg)
        self.processed_indices.add(event.segment_index)
        self.last_segment_end = datetime.now(timezone.utc)
        return seg


async def stream_audio_producer(stream, audio_source, stop_event: asyncio.Event):
    buffer = bytearray()
    async for chunk in audio_source:
        if stop_event.is_set(): break
        buffer.extend(chunk)
        while len(buffer) >= FRAME_SIZE:
            await stream.send_audio(bytes(buffer[:FRAME_SIZE]))
            buffer = buffer[FRAME_SIZE:]
    if buffer:
        await stream.send_audio(bytes(buffer) + b" " * (FRAME_SIZE - len(buffer)))


async def handle_event(session, event, compliance_engine, quality_engine, alert_dispatcher):
    if isinstance(event, StreamConnectedEvent):
        print(f"[{session.call_id}] Connected  sample_rate={event.sample_rate}")

    elif isinstance(event, StreamProcessingEvent):
        quality_engine.on_processing(event.timestamp)

    elif isinstance(event, StreamTranscriptEvent):
        if event.segment_index in session.processed_indices:
            return  # deduplicate across reconnects
        seg = session.add_segment(event)
        print(f"[{session.call_id}][{seg.segment_index}] {seg.text}  (latency={seg.latency_ms}ms)")
        c_hits   = compliance_engine.check(seg)
        q_events = quality_engine.check(session, seg)
        seg.compliance_flags = [h["rule_id"]     for h in c_hits]
        seg.quality_flags    = [e["event_type"]  for e in q_events]
        if c_hits or q_events:
            asyncio.create_task(alert_dispatcher.send(session, seg, c_hits + q_events))

    elif isinstance(event, StreamErrorEvent):
        raise RuntimeError(f"STT error: {event.message}")


async def monitor_call(session, audio_source, compliance_engine, quality_engine, alert_dispatcher):
    attempt = 0
    while attempt <= MAX_RECONNECTS:
        try:
            async with GnaniSTTStreamClient(
                api_key=os.getenv("GNANI_API_KEY"),
                language_code=session.language_code,
                sample_rate=int(os.getenv("SAMPLE_RATE", "16000")),
            ) as stream:
                if attempt > 0: session.reconnect_count += 1
                attempt = 0
                stop_event = asyncio.Event()
                producer   = asyncio.create_task(stream_audio_producer(stream, audio_source, stop_event))
                async for event in stream:
                    await handle_event(session, event, compliance_engine, quality_engine, alert_dispatcher)
                stop_event.set()
                await producer
                return
        except StreamConnectionError: raise
        except (StreamClosedError, ConnectionResetError, OSError) as e:
            attempt += 1
            if attempt > MAX_RECONNECTS: raise
            backoff = min(BASE_BACKOFF_S * (2 ** attempt), MAX_BACKOFF_S)
            print(f"[{session.call_id}] Reconnect {attempt}/{MAX_RECONNECTS} in {backoff:.1f}s  {e}")
            await asyncio.sleep(backoff + random.uniform(0, backoff * 0.2))
        except StreamError as e:
            attempt += 1
            print(f"[{session.call_id}] Server error: {e.message}")
            await asyncio.sleep(BASE_BACKOFF_S)


async def file_audio_source(path: str, sample_rate=16000) -> AsyncIterator[bytes]:
    frame_interval = FRAME_MS_16K if sample_rate == 16000 else FRAME_MS_8K
    with open(path, "rb") as f:
        while chunk := f.read(FRAME_SIZE):
            yield chunk
            await asyncio.sleep(frame_interval)


async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--audio", required=True)
    parser.add_argument("--lang",  default="hi-IN")
    parser.add_argument("--rate",  default=16000, type=int)
    args = parser.parse_args()
    os.environ["SAMPLE_RATE"] = str(args.rate)

    session = CallSession(call_id="CALL_001", language_code=args.lang)
    audio   = file_audio_source(args.audio, sample_rate=args.rate)
    await monitor_call(session, audio, ComplianceEngine(), QualityEngine(), AlertDispatcher())

    duration = (datetime.now(timezone.utc) - session.started_at).total_seconds()
    print(f"Duration: {duration:.1f}s  |  Segments: {len(session.segments)}")
    print(f"Compliance hits: {sum(len(s.compliance_flags) for s in session.segments)}")
    print(f"Quality events:  {sum(len(s.quality_flags)    for s in session.segments)}")

if __name__ == "__main__":
    asyncio.run(main())