"""Generate or verify the five thin Colab orchestration notebooks.""" from __future__ import annotations import argparse import json from pathlib import Path NOTEBOOKS_DIR = Path(__file__).resolve().parents[1] / "notebooks" BOOTSTRAP = r'''# CarePath bootstrap: local checkout first, otherwise a private GitHub clone. import importlib.util import os import shutil import subprocess import sys import tempfile from pathlib import Path def _find(start): for directory in [start, *start.parents]: if (directory / 'pyproject.toml').exists() and (directory / 'scribe' / 'carepath').exists(): return directory return None def _secret(): for key in ('CAREPATH_GITHUB_TOKEN', 'GITHUB_TOKEN'): if os.environ.get(key): return os.environ[key] try: from google.colab import userdata for key in ('CAREPATH_GITHUB_TOKEN', 'GITHUB_TOKEN'): try: value = userdata.get(key) if value: return value except Exception: pass except Exception: pass return None def _inject_runtime_secrets(): """Copy only approved named Colab secrets into memory for child processes.""" try: from google.colab import userdata except Exception: return for key in ('LLM_API_KEY', 'HF_TOKEN'): if os.environ.get(key): continue try: value = userdata.get(key) except Exception: value = None if value: os.environ[key] = value REPO = _find(Path.cwd().resolve()) if REPO is None and importlib.util.find_spec('google.colab'): target = Path('/content/carepath') REPO = _find(target) if REPO is None: if target.exists(): shutil.rmtree(target) url = os.environ.get('CAREPATH_REPO_URL', 'https://github.com/truong-tt/carepath.git') if '://' in url and '@' in url.split('://', 1)[1].split('/', 1)[0]: raise SystemExit('CAREPATH_REPO_URL must not contain credentials; use a Colab Secret.') token = _secret() clone_env = dict(os.environ) clone_env['GIT_TERMINAL_PROMPT'] = '0' with tempfile.TemporaryDirectory(prefix='carepath_git_') as temp: if token: askpass = Path(temp) / 'askpass.sh' askpass.write_text( '#!/bin/sh\ncase "$1" in *Username*) echo x-access-token ;; *) echo "$GITHUB_TOKEN" ;; esac\n', encoding='utf-8', ) askpass.chmod(0o700) clone_env['GIT_ASKPASS'] = str(askpass) clone_env['GITHUB_TOKEN'] = token result = subprocess.run( ['git', 'clone', url, str(target)], env=clone_env, capture_output=True, text=True, ) if result.returncode: error = result.stderr or result.stdout if token: error = error.replace(token, '***') raise SystemExit( 'Clone failed. Add a Colab Secret named GITHUB_TOKEN with repository read access, ' 'enable Notebook access, then rerun.\n' + error ) REPO = target assert REPO, 'Run the notebook inside CarePath or from Google Colab.' os.chdir(REPO) sys.path[:0] = [str(REPO / 'scribe' / 'training'), str(REPO / 'scribe')] _inject_runtime_secrets() PROFILE = os.environ.get('CAREPATH_PROFILE', 'smoke') CONFIRM_PAID = os.environ.get('CAREPATH_CONFIRM_PAID') == '1' from gec.notebook import init_stage CTX = init_stage(PROFILE, confirm_paid=CONFIRM_PAID) P, PROF = CTX.paths, CTX.profile ''' INSTALL_BASE = r"""# Pin the repository's tested training and combined-app environment. extras = ['dev', 'training'] if PROFILE == 'reproduction': extras.append('training-tts') """ INSTALL_NORMAL = INSTALL_BASE + r"""extra = '.[' + ','.join(extras) + ']' subprocess.run( [sys.executable, '-m', 'pip', 'install', '-q', '-e', extra, '-e', './shared', '-e', './interpreter'], check=True, ) """ INSTALL_FAST = INSTALL_BASE + r"""if PROF.paid: extras.append('training-fast') extra = '.[' + ','.join(extras) + ']' subprocess.run( [sys.executable, '-m', 'pip', 'install', '-q', '-e', extra, '-e', './shared', '-e', './interpreter'], check=True, ) """ NOTEBOOK_SPECS = ( ( "00_data_prep.ipynb", INSTALL_NORMAL, "# CarePath — 00 Data preparation `[CPU]`\n\n" "Validate the selected profile and build the term datastore. Smoke is the default; paid " "profiles require `CAREPATH_CONFIRM_PAID=1` and an approved dataset manifest.", "CTX.run_pipeline('data')\n" "if PROF.paid:\n" " CTX.run_step([\n" " 'scribe/training/scripts/prepare_public_sources.py',\n" " '--manifest', 'scribe/training/manifests/soap-public-v1.json',\n" " '--canonical', 'shared/carepath_shared/terms/medical_terms.json',\n" " '--output-root', '/content/carepath_data',\n" " ])\n" "print('datastore ->', P.datastore)\n", ), ( "01_asr_benchmark.ipynb", INSTALL_NORMAL, "# CarePath — 01 ASR benchmark `[L4 for real data]`\n\n" "Build production-compatible **single-best** Gipformer pairs and a duration-derived " "benchmark. Paid profiles train/evaluate plain PhoWhisper-small LoRA on ViMedCSS train only; " "the reproduction profile fails closed if the plain-LoRA gate makes the still-unimplemented " "contrastive near-miss stage mandatory.", "CTX.run_pipeline('asr')\nprint(P.asr_benchmark.read_text(encoding='utf-8'))\n", ), ( "02_train_gec.ipynb", INSTALL_FAST, "# CarePath — 02 Transcript correction `[L4/A100]`\n\n" "Run optional synthetic/TTS work only for the explicit reproduction profile, then train " "the configured candidate and seed through the consent-gated orchestrator.", "if PROF.enable_synthetic:\n CTX.run_pipeline('synth')\n" "CTX.run_pipeline('train')\nprint('candidate adapter ->', P.candidate_adapter(\n" " PROF.candidate_variant, PROF.candidate_seed, all_variants=PROF.all_variants, " "multi_seed=len(PROF.seeds) > 1))\n", ), ( "03_train_soap.ipynb", INSTALL_FAST, "# CarePath — 03 Grounded SOAP training `[L4/A100]`\n\n" "Call the sibling SOAP orchestrator. Its data, factual filters, and export gates remain " "owned by the SOAP training package.", "CTX.run_soap_pipeline()\n", ), ( "04_evaluate_export_stage.ipynb", INSTALL_NORMAL, "# CarePath — 04 Evaluate, gate, export, and stage `[CPU, then GPU smoke]`\n\n" "Run normal and frozen safety evaluation through the same orchestrator as CI. Export occurs " "only when the selected trained candidate clears both gates; mock bundles are deliberately refused.", "import json\n" "CTX.run_pipeline('eval')\n" "selection = json.loads(P.candidate_selection.read_text(encoding='utf-8'))\n" "print(json.dumps(selection, ensure_ascii=False, indent=2))\n" "asr_component = P.asr_lora_root / f'seed-{PROF.candidate_seed}'\n" "asr_manifest = asr_component / 'asr_component.json'\n" "if asr_manifest.exists():\n" " asr_meta = json.loads(asr_manifest.read_text(encoding='utf-8'))\n" " if asr_meta.get('gate_accepted') and asr_meta.get('selected_for_serving'):\n" " CTX.run_step([\n" " 'scribe/training/scripts/stage_asr_component.py',\n" " '--component', str(asr_component), '--manifest', str(CTX.config_path.parent.parent / 'manifests' / 'vimedcss-v1.json'),\n" " '--predictions', str(P.evaluations / 'asr_staging_predictions.jsonl'),\n" " '--report', str(P.evaluations / 'asr_staging_report.json'), '--confirm-paid'])\n" " else:\n" " print('Direct-ASR staging deliberately skipped: component did not pass and win selection.')\n" "from soap.bundle import validate_bundle\n" "from soap.config import load_config as load_soap_config\n" "soap_profile = 'replicate' if PROFILE == 'reproduction' else PROFILE\n" "soap_cfg_path = Path('scribe/training/configs') / f'soap-{soap_profile}-v1.json'\n" "if not soap_cfg_path.exists() and PROFILE == 'smoke':\n" " soap_cfg_path = Path('scribe/training/configs/soap-smoke-v1.json')\n" "soap_cfg = load_soap_config(soap_cfg_path)\n" "soap_bundle = P.root / soap_cfg.run_id / 'bundle'\n" "assembled = P.root / 'accepted_scribe_bundle'\n" "try:\n" " validate_bundle(soap_bundle)\n" "except ValueError as exc:\n" " print('Staging deliberately refused:', exc)\n" "else:\n" " component_args = None\n" " if selection['selected'] == 'gec_pred' and P.serve_bundle.exists():\n" " component_args = ['--gec-bundle', str(P.serve_bundle)]\n" " elif selection['selected'] == 'phowhisper_pred' and asr_manifest.exists():\n" " component_args = ['--asr-component', str(asr_component)]\n" " if component_args:\n" " CTX.run_step(['scribe/training/scripts/assemble_scribe_bundle.py', '--soap-bundle', str(soap_bundle), *component_args, '--output', str(assembled), '--confirm-stack'])\n" " CTX.run_step(['scribe/training/scripts/stage_scribe_bundle.py', '--bundle', str(assembled)])\n" " else:\n" " print('No assembled bundle staged: raw Gipformer baseline or no accepted transcript component.')\n" "print('Private publishing remains a separate explicit --confirm-private-upload command.')\n", ), ) def _source(text: str) -> list[str]: return text.splitlines(keepends=True) def _notebook(install: str, title: str, body: str) -> str: payload = { "cells": [ {"cell_type": "markdown", "metadata": {}, "source": _source(title)}, { "cell_type": "code", "metadata": {}, "execution_count": None, "outputs": [], "source": _source(BOOTSTRAP), }, { "cell_type": "code", "metadata": {}, "execution_count": None, "outputs": [], "source": _source(install), }, { "cell_type": "code", "metadata": {}, "execution_count": None, "outputs": [], "source": _source(body), }, ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3", }, "language_info": {"name": "python"}, }, "nbformat": 4, "nbformat_minor": 5, } return json.dumps(payload, ensure_ascii=False, indent=1) + "\n" def build(*, check: bool = False) -> None: expected = { name: _notebook(install, title, body) for name, install, title, body in NOTEBOOK_SPECS } actual_names = {path.name for path in NOTEBOOKS_DIR.glob("[0-9][0-9]_*.ipynb")} drift = sorted(actual_names ^ expected.keys()) drift += [ name for name, content in expected.items() if not (NOTEBOOKS_DIR / name).exists() or (NOTEBOOKS_DIR / name).read_text(encoding="utf-8") != content ] if check: if drift: raise SystemExit( "generated notebooks drifted: " + ", ".join(sorted(set(drift))) ) print("Generated notebooks are current.") return NOTEBOOKS_DIR.mkdir(parents=True, exist_ok=True) for stale in NOTEBOOKS_DIR.glob("[0-9][0-9]_*.ipynb"): if stale.name not in expected: stale.unlink() for name, content in expected.items(): (NOTEBOOKS_DIR / name).write_text(content, encoding="utf-8") print("wrote", name) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--check", action="store_true", help="fail if generated notebooks drift" ) args = parser.parse_args() build(check=args.check) if __name__ == "__main__": main()