from __future__ import annotations from dataclasses import dataclass from pathlib import Path import re import shutil from typing import Iterable TOKEN_RE = re.compile(r"[A-Za-zÀ-ÿ0-9']+") SENTENCE_RE = re.compile(r'(?<=[.!?])\s+') MOOD_TERMS = { 'tired': 'mood:tired', 'sad': 'mood:sad', 'anxious': 'mood:anxious', 'worried': 'mood:worried', 'calm': 'mood:calm', 'better': 'mood:improving', } SYMPTOM_TERMS = { 'pain': 'symptoms:pain', 'dizzy': 'symptoms:dizziness', 'dizziness': 'symptoms:dizziness', 'cough': 'symptoms:cough', 'fever': 'symptoms:fever', 'nausea': 'symptoms:nausea', 'appetite': 'symptoms:low_appetite', 'breath': 'symptoms:shortness_of_breath', } ACTIVITY_TERMS = { 'walk': 'activity:walking', 'walking': 'activity:walking', 'rest': 'activity:resting', 'sleep': 'activity:sleep', 'slept': 'activity:sleep', 'visit': 'activity:visit', 'appointment': 'activity:appointment', } MEDICATION_TERMS = { 'medication': 'meds:medication', 'meds': 'meds:medication', 'dose': 'meds:dose_change', 'pill': 'meds:pill', 'refill': 'meds:refill', } RISK_TERMS = { 'hurt', 'harm', 'abuse', 'suicide', 'kill', 'overdose', 'emergency', } @dataclass(frozen=True) class JournalSummary: transcript: str family_view: str clinician_view: str tags: list[str] segment_confidences: list[dict[str, object]] safety_tag: str questions_for_doctor: list[str] def tokenize(text: str) -> list[str]: return [token.lower() for token in TOKEN_RE.findall(text or '')] def extract_tags(text: str) -> list[str]: tokens = tokenize(text) tags: list[str] = [] for token in tokens: for mapping in (MOOD_TERMS, SYMPTOM_TERMS, ACTIVITY_TERMS, MEDICATION_TERMS): if token in mapping and mapping[token] not in tags: tags.append(mapping[token]) if not tags: tags.append('care:general') return tags def safety_label(text: str) -> str: lowered = (text or '').lower() return 'needs review' if any(term in lowered for term in RISK_TERMS) else 'ok' def _shorten(text: str, limit: int = 160) -> str: text = ' '.join((text or '').split()) return text if len(text) <= limit else text[: limit - 1].rstrip() + '…' def family_summary(text: str) -> str: sentences = [sentence.strip() for sentence in SENTENCE_RE.split((text or '').strip()) if sentence.strip()] if not sentences: return 'No transcript provided.' first = _shorten(sentences[0], 170) tags = extract_tags(text) return f'Family update: {first}. Tags: {", ".join(tags[:4])}.' def clinician_summary(text: str) -> str: tags = extract_tags(text) first = _shorten((text or '').split('\n', 1)[0], 140) return f'Clinician note: {first}. Relevant tags: {", ".join(tags[:4])}.' def segment_confidences(text: str) -> list[dict[str, object]]: sentences = [sentence.strip() for sentence in SENTENCE_RE.split((text or '').strip()) if sentence.strip()] if not sentences: sentences = [text.strip()] if text and text.strip() else [] if not sentences: return [] confidences = [] for idx, sentence in enumerate(sentences, start=1): token_count = max(1, len(tokenize(sentence))) conf = min(0.99, 0.58 + min(token_count, 18) / 40) confidences.append({ 'segment': idx, 'text': _shorten(sentence, 120), 'confidence': round(conf, 2), }) return confidences def doctor_questions(tags: Iterable[str]) -> list[str]: tag_set = list(tags) questions: list[str] = [] if any(tag.startswith('symptoms:') for tag in tag_set): questions.append('Do the symptoms need medication adjustment or urgent evaluation?') if any(tag.startswith('meds:') for tag in tag_set): questions.append('Was there a missed dose, refill issue, or side effect?') if any(tag.startswith('activity:') for tag in tag_set): questions.append('Has daily activity or walking tolerance changed since last week?') if any(tag.startswith('mood:') for tag in tag_set): questions.append('Is the mood change persistent or linked to sleep and pain?') if not questions: questions.append('Is there anything new that needs a clinician follow-up?') return questions[:3] def summarize_entry(text: str) -> JournalSummary: tags = extract_tags(text) return JournalSummary( transcript=text, family_view=family_summary(text), clinician_view=clinician_summary(text), tags=tags, segment_confidences=segment_confidences(text), safety_tag=safety_label(text), questions_for_doctor=doctor_questions(tags), ) def normalize_audio_file(source_path: str | Path, output_dir: str | Path, *, stem: str | None = None) -> Path: source = Path(source_path) output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) target = output_dir / f'{stem or source.stem}_16k_mono.wav' shutil.copy2(source, target) return target def digest_entries(entries: list[dict[str, object]], *, start_label: str = '', end_label: str = '') -> dict[str, object]: if not entries: return { 'range': {'start': start_label, 'end': end_label}, 'summary': 'No entries found for this date range.', 'key_events': [], 'questions_for_doctor': ['No entries found; record at least one diary clip.'], } tags: list[str] = [] events: list[str] = [] for entry in entries: entry_tags = list(entry.get('tags', [])) for tag in entry_tags: if tag not in tags: tags.append(tag) summary = str(entry.get('family_summary') or entry.get('clinician_summary') or entry.get('transcript') or '') if summary: events.append(_shorten(summary, 100)) clinic_questions = doctor_questions(tags) digest_summary = f"{len(entries)} entries reviewed. Notable themes: {', '.join(tags[:5])}." return { 'range': {'start': start_label, 'end': end_label}, 'summary': digest_summary, 'key_events': events[:5], 'questions_for_doctor': clinic_questions, }