diff --git a/.gitignore b/.gitignore index 9ccae003..cbff9821 100644 --- a/.gitignore +++ b/.gitignore @@ -161,17 +161,17 @@ cython_debug/ output/ models/ +local/ !healthchain/models/ !healthchain/interop/models/ -scrap/ .DS_Store .vscode/ .ruff_cache/ .python-version .cursor/ -.local/ .keys/ .idea/ +.claude/ # Personal AI context (keep local) CLAUDE.LOCAL.MD diff --git a/README.md b/README.md index 46318128..120e462f 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,25 @@ HealthChain is an open-source SDK for production-ready healthcare AI. Skip month pip install healthchain ``` +## Quick Start + +```bash +# Scaffold a FHIR Gateway project +healthchain new my-app -t fhir-gateway +cd my-app + +# Run locally +healthchain serve +``` + +
+ HealthChain CLI demo +
+ +Edit `app.py` to add your model, and `healthchain.yaml` to configure compliance, security, and deployment settings. + +See the [CLI reference](https://dotimplement.github.io/HealthChain/cli/) for all commands. + ## Core Features HealthChain is the **quickest way for AI/ML engineers to integrate their models with real healthcare systems**. @@ -132,7 +151,7 @@ HealthChain understands healthcare protocols and data formats natively, so you d - **Automatic validation** - Type-safe FHIR models prevent broken healthcare data - **Built-in NLP support** - Extract structured data from clinical notes, output as FHIR - **Developer experience** - Modular and extensible architecture works across any EHR system -- **Production-ready** - OAuth2-based authentication, Dockerized deployment, and structured audit hooks for real-world healthcare environments +- **Production-ready foundations** - Dockerized deployment, configurable security and compliance settings, and an architecture designed for real-world healthcare environments ## 🏆 Recognition & Community @@ -146,19 +165,6 @@ HealthChain understands healthcare protocols and data formats natively, so you d Exploring HealthChain for your product or organization? [Get in touch](mailto:jenniferjiangkells@gmail.com) to discuss integrations, pilots, or collaborations, or join our [Discord](https://discord.gg/UQC6uAepUz) to connect with the community. -## Create a New HealthChain App - -```bash -# Scaffold a new project -healthchain new my-app -cd my-app - -# Run locally -healthchain serve -``` - -See the [CLI reference](https://dotimplement.github.io/HealthChain/cli/) for all commands. - ## Usage Examples ### Building a Pipeline [[Docs](https://dotimplement.github.io/HealthChain/reference/pipeline/pipeline)] diff --git a/docs/assets/images/demo.gif b/docs/assets/images/demo.gif new file mode 100644 index 00000000..3299c5b4 Binary files /dev/null and b/docs/assets/images/demo.gif differ diff --git a/docs/cli.md b/docs/cli.md index de36fa8f..77ab9591 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -1,6 +1,6 @@ # CLI Reference -HealthChain ships with a CLI to help you scaffold, run, and customize projects. +HealthChain ships with a CLI to scaffold, configure, run, and test projects. ```bash healthchain --help @@ -10,19 +10,31 @@ healthchain --help ## `healthchain new` -Scaffold a new project directory with everything you need to get started. +Scaffold a new project directory. ```bash -healthchain new my-app +healthchain new my-app # empty stub +healthchain new my-app --type cds-hooks # working CDS Hooks service +healthchain new my-app --type fhir-gateway # working FHIR Gateway service +healthchain new my-app -t cds-hooks # shorthand ``` +**`--type` / `-t`** options: + +| Type | Description | +|------|-------------| +| *(default)* | Empty `app.py` stub — choose your own starting point | +| `cds-hooks` | Working CDS Hooks service with a `patient-view` hook | +| `fhir-gateway` | Working FHIR Gateway that aggregates conditions from multiple EHR sources | + Creates: ``` my-app/ -├── app.py # your application entry point -├── .env.example # FHIR credential template — copy to .env and fill in -├── requirements.txt # add extra dependencies here +├── app.py # application entry point (populated for cds-hooks / fhir-gateway) +├── healthchain.yaml # project configuration +├── .env.example # credential template — copy to .env and fill in +├── requirements.txt ├── Dockerfile └── .dockerignore ``` @@ -31,16 +43,20 @@ my-app/ ## `healthchain serve` -Start your app locally with uvicorn. +Start your app locally with uvicorn. Reads `healthchain.yaml` automatically if present. ```bash -healthchain serve # defaults to app:app on port 8000 +healthchain serve # uses port from healthchain.yaml (default 8000) healthchain serve app:app -healthchain serve app:app --port 8080 -healthchain serve app:app --host 127.0.0.1 --port 8080 +healthchain serve app:app --port 8080 # CLI flag overrides healthchain.yaml +healthchain serve app:app --host 127.0.0.1 ``` -The `app_module` argument is the Python import path to your FastAPI app instance — `:`. If your app is defined as `app = HealthChainAPI()` in `app.py`, the default `app:app` works as-is. +On startup, HealthChain prints a status banner showing your service name, type, environment, security configuration, and docs URL. + +If `security.tls.enabled: true` in `healthchain.yaml`, cert and key paths are passed to uvicorn automatically. + +**Port resolution order:** `--port` flag → `service.port` in `healthchain.yaml` → `8000` To run in Docker instead: @@ -51,15 +67,113 @@ docker run -p 8000:8000 --env-file .env my-app --- +## `healthchain status` + +Show the current project's configuration state, read from `healthchain.yaml`. + +```bash +healthchain status +``` + +Example output: + +``` +HealthChain — my-sepsis-app v1.0.0 +----------------------------------- + +Service + type cds-hooks + port 8000 + +Site + environment production + name General Hospital NHS Trust + +Security + auth smart-on-fhir + TLS enabled + origins https://fhir.epic.com + +Compliance + HIPAA enabled + audit log ./logs/audit.jsonl + +Eval + provider mlflow + tracking ./mlruns + events model_inference, cds_card_returned, card_feedback +``` + +--- + +## `healthchain sandbox run` + +Fire test requests at a running HealthChain service. Useful for testing and live demos. + +```bash +# Generate synthetic requests (no data files needed) +healthchain sandbox run \ + --url http://localhost:8000/cds/cds-services/my-service + +# Specify workflow and number of requests +healthchain sandbox run \ + --url http://localhost:8000/cds/cds-services/my-service \ + --workflow encounter-discharge \ + --size 5 + +# Load requests from patient files instead of generating +healthchain sandbox run \ + --url http://localhost:8000/cds/cds-services/my-service \ + --from-path ./data/patients/ + +# Quick test — don't save results to disk +healthchain sandbox run \ + --url http://localhost:8000/cds/cds-services/my-service \ + --no-save +``` + +**Options:** + +| Flag | Default | Description | +|------|---------|-------------| +| `--url` | *(required)* | Full service URL | +| `--workflow` | `patient-view` | CDS workflow to simulate (`patient-view`, `order-select`, `order-sign`, `encounter-discharge`) | +| `--size` | `3` | Number of synthetic requests to generate | +| `--from-path` | — | Load requests from a file or directory (`.json` FHIR prefetch or `.xml` CDA) instead of generating synthetic data | +| `--output` | from `healthchain.yaml` or `./output` | Directory to save results | +| `--no-save` | — | Don't save results to disk | + +Example output: + +``` +Sandbox — http://localhost:8000/cds/cds-services/my-service +workflow: patient-view +Generating 3 synthetic request(s)... +Sending 3 request(s) to service... + +Results: 3/3 successful + + [1] 1 card(s) + INFO: Sepsis risk: LOW — no immediate action required + [2] 1 card(s) + WARNING: Sepsis risk: HIGH — review vitals and consider intervention + [3] 1 card(s) + INFO: Sepsis risk: MODERATE — monitor closely + +Saved to ./output/ +``` + +--- + ## `healthchain eject-templates` -Copy the built-in interop templates into your project so you can customize them. +Copy the built-in interop templates into your project for customization. ```bash healthchain eject-templates ./my_configs ``` -Only needed if you're using the [InteropEngine](reference/interop/interop.md) and want to customize FHIR↔CDA conversion beyond the defaults. After ejecting: +Only needed if you're using the [InteropEngine](reference/interop/interop.md) and want to customize FHIR↔CDA conversion beyond the defaults. ```python from healthchain.interop import create_interop @@ -67,28 +181,34 @@ from healthchain.interop import create_interop engine = create_interop(config_dir="./my_configs") ``` -See [Interoperability](reference/interop/interop.md) for details. +--- + +## `healthchain.yaml` + +Generated by `healthchain new` and read automatically by `healthchain serve` and `healthchain status`. See the [Configuration Reference](reference/config.md) for the full schema. --- ## Typical workflow ```bash -# 1. Scaffold a new project -healthchain new my-cds-service +# 1. Scaffold a new CDS Hooks service +healthchain new my-cds-service -t cds-hooks cd my-cds-service -# 2. Build your app in app.py -# See https://dotimplement.github.io/HealthChain/cookbook/ for examples - -# 3. Set credentials -cp .env.example .env -# edit .env with your FHIR_BASE_URL, CLIENT_ID, CLIENT_SECRET +# 2. Build your clinical logic in app.py -# 4. Run locally +# 3. Run locally healthchain serve -# 5. Ship it +# 4. In another terminal — test with synthetic patients +healthchain sandbox run \ + --url http://localhost:8000/cds/cds-services/my-service + +# 5. Check config and compliance state +healthchain status + +# 6. Ship it docker build -t my-cds-service . docker run -p 8000:8000 --env-file .env my-cds-service ``` diff --git a/docs/reference/config.md b/docs/reference/config.md new file mode 100644 index 00000000..0fef4eba --- /dev/null +++ b/docs/reference/config.md @@ -0,0 +1,118 @@ +# Configuration Reference + +`healthchain.yaml` is the project configuration file generated by `healthchain new`. It is read automatically by `healthchain serve` and `healthchain status`. + +```yaml +name: my-app +version: "1.0.0" + +service: + type: cds-hooks + port: 8000 + +data: + patients_dir: ./data + output_dir: ./output + +security: + auth: none + tls: + enabled: false + cert_path: ./certs/server.crt + key_path: ./certs/server.key + allowed_origins: + - "*" + +compliance: + hipaa: false + audit_log: ./logs/audit.jsonl + +eval: + enabled: false + provider: mlflow + tracking_uri: ./mlruns + track: + - model_inference + - cds_card_returned + - card_feedback + +site: + name: "" + environment: development +``` + +--- + +## `service` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `type` | string | `cds-hooks` | Service type — `cds-hooks` or `fhir-gateway` | +| `port` | int | `8000` | Port for `healthchain serve` | + +--- + +## `data` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `patients_dir` | path | `./data` | Directory for patient data files | +| `output_dir` | path | `./output` | Directory for sandbox results | + +--- + +## `security` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `auth` | string | `none` | Authentication method — `none`, `api-key` (planned), or `smart-on-fhir` (planned) | +| `tls.enabled` | bool | `false` | Enable TLS — passes cert/key to uvicorn automatically | +| `tls.cert_path` | path | `./certs/server.crt` | Path to TLS certificate | +| `tls.key_path` | path | `./certs/server.key` | Path to TLS private key | +| `allowed_origins` | list | `["*"]` | CORS allowed origins — passed directly to FastAPI's CORS middleware | + +!!! note "Authentication is planned, not yet active" + Setting `auth: api-key` or `auth: smart-on-fhir` is accepted by the config but not yet enforced at runtime. Authentication middleware is on the roadmap. `allowed_origins` is functional — it controls which origins are permitted by the CORS middleware. + +--- + +## `compliance` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `hipaa` | bool | `false` | Mark service as HIPAA-scoped — displayed in startup banner and `healthchain status` | +| `audit_log` | path | `./logs/audit.jsonl` | Destination for audit log events (planned) | + +!!! note "Audit logging is planned, not yet active" + Setting `hipaa: true` currently marks the service as HIPAA-scoped in the CLI and startup banner. Structured audit logging (PHI access events written to `audit_log`) is on the roadmap but not yet implemented. + +--- + +## `eval` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `enabled` | bool | `false` | Enable model evaluation tracking | +| `provider` | string | `mlflow` | Eval backend — `mlflow`, `langfuse`, or `none` | +| `tracking_uri` | path | `./mlruns` | MLFlow tracking directory | +| `track` | list | see below | Events to capture | + +Default tracked events: + +- `model_inference` — input features and prediction for each request +- `cds_card_returned` — which card was shown to the clinician +- `card_feedback` — whether the clinician accepted, overrode, or ignored the card + +The `card_feedback` event closes the evaluation loop — it provides implicit ground truth for model performance regardless of whether your model is an ML classifier, NLP pipeline, or LLM. + +!!! note + MLFlow integration is on the roadmap. Setting `eval.enabled: true` prepares your project for when it ships. + +--- + +## `site` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `name` | string | `""` | Hospital or organisation name — displayed in `healthchain status` | +| `environment` | string | `development` | Deployment environment — `development`, `staging`, or `production` | diff --git a/healthchain/cli.py b/healthchain/cli.py index f720367e..3846a78c 100644 --- a/healthchain/cli.py +++ b/healthchain/cli.py @@ -3,6 +3,17 @@ import sys from pathlib import Path +# ANSI color helpers — consistent with startup banner palette +_RST = "\033[0m" +_BOLD = "\033[1m" +_DIM = "\033[2m" +_GREEN = "\033[38;2;0;255;135m" +_CYAN = "\033[38;2;0;215;255m" +_INDIGO = "\033[38;2;99;102;241m" +_PINK = "\033[38;2;255;121;198m" +_AMBER = "\033[38;2;255;180;50m" +_RED = "\033[38;2;255;85;85m" + _DOCKERFILE = """\ # HealthChain application Dockerfile # @@ -73,7 +84,36 @@ !README.md """ -_ENV_EXAMPLE = """\ +_ENV_EXAMPLE_CDS_HOOKS = """\ +# CDS Hooks service — no credentials required to run locally. +# Add FHIR source credentials if your hook fetches patient data. + +# FHIR_BASE_URL= +# CLIENT_ID= +# CLIENT_SECRET= + +# For JWT assertion flow (e.g. Epic SMART on FHIR) +# CLIENT_SECRET_PATH=/path/to/private_key.pem +""" + +_ENV_EXAMPLE_FHIR_GATEWAY = """\ +# FHIR source credentials +# Add one block per EHR source. Prefix matches the source name in add_source(). + +# Epic +EPIC_BASE_URL=https://fhir.epic.com/interconnect-fhir-oauth/api/FHIR/R4 +EPIC_CLIENT_ID= +EPIC_CLIENT_SECRET= +# EPIC_CLIENT_SECRET_PATH=/path/to/epic_private_key.pem + +# Cerner +CERNER_BASE_URL=https://fhir-open.cerner.com/r4/ec2458f2-1e24-41c8-b71b-0e701af7583d +# Cerner open sandbox requires no credentials + +# See docs: https://dotimplement.github.io/HealthChain/reference/gateway/fhir_gateway/ +""" + +_ENV_EXAMPLE_DEFAULT = """\ # FHIR source credentials FHIR_BASE_URL= CLIENT_ID= @@ -85,40 +125,214 @@ _REQUIREMENTS = "healthchain\n" -_APP_PY = """\ +_APP_PY_DEFAULT = """\ # Your HealthChain application goes here. # See https://dotimplement.github.io/HealthChain/ for examples. """ +_APP_PY_CDS_HOOKS = """\ +from healthchain.gateway import HealthChainAPI +from healthchain.gateway.cds import CDSHooksService +from healthchain.models.requests.cdsrequest import CDSRequest +from healthchain.models.responses.cdsresponse import CDSResponse + +app = HealthChainAPI( + title="My CDS Service", + description="A CDS Hooks service built with HealthChain", +) + +cds = CDSHooksService() + + +@cds.hook("patient-view", id="my-service", title="My CDS Service") +def patient_view(request: CDSRequest) -> CDSResponse: + # Add your clinical decision support logic here. + # request.context contains patient/encounter context from the EHR. + # request.prefetch contains pre-fetched FHIR resources. + return CDSResponse( + cards=[ + { + "summary": "HealthChain CDS", + "detail": "Add your clinical logic here.", + "indicator": "info", + "source": {"label": "My CDS Service"}, + } + ] + ) + + +app.register_service(cds) +""" + +_APP_PY_FHIR_GATEWAY = """\ +import os +from typing import List + +from fhir.resources.bundle import Bundle +from fhir.resources.condition import Condition + +from healthchain.gateway import FHIRGateway, HealthChainAPI +from healthchain.fhir import merge_bundles +from healthchain.io.containers import Document +from healthchain.pipeline import Pipeline + +# Add FHIR source credentials to .env (see .env.example) +gateway = FHIRGateway() + +epic_url = os.getenv("EPIC_BASE_URL") +cerner_url = os.getenv("CERNER_BASE_URL") + +if epic_url: + gateway.add_source("epic", epic_url) +if cerner_url: + gateway.add_source("cerner", cerner_url) + +# Add your NLP/ML/LLM processing steps here +pipeline = Pipeline[Document]() + + +@pipeline.add_node +def process(doc: Document) -> Document: + return doc + + +@gateway.aggregate(Condition) +def get_patient_conditions(patient_id: str, sources: List[str]) -> Bundle: + \"\"\"Aggregate conditions for a patient from all configured FHIR sources.\"\"\" + bundles = [] + for source in sources: + try: + bundle = gateway.search( + Condition, + {"patient": patient_id}, + source, + add_provenance=True, + ) + bundles.append(bundle) + except Exception as e: + print(f"Error from {source}: {e}") + + merged = merge_bundles(bundles, deduplicate=True) + doc = pipeline(Document(data=merged)) + return doc.fhir.bundle + + +app = HealthChainAPI( + title="My FHIR Gateway", + description="A multi-EHR data aggregation service built with HealthChain", +) +app.register_gateway(gateway) +""" + + +def _make_healthchain_yaml(name: str, service_type: str) -> str: + return f"""\ +# HealthChain application configuration +# https://dotimplement.github.io/HealthChain/reference/config + +name: {name} +version: "1.0.0" + +# Service settings — read by `healthchain serve` +service: + type: {service_type} + port: 8000 + +# Data paths used by your app and sandbox +data: + patients_dir: ./data + output_dir: ./output + +# Security controls +security: + auth: none # none | api-key (planned) | smart-on-fhir (planned) + tls: + enabled: false + cert_path: ./certs/server.crt + key_path: ./certs/server.key + allowed_origins: + - "*" + +# Compliance settings +compliance: + hipaa: false # set true to activate audit logging + audit_log: ./logs/audit.jsonl + +# Evaluation and model monitoring +eval: + enabled: false + provider: mlflow # mlflow | langfuse | none + tracking_uri: ./mlruns + track: + - model_inference + - cds_card_returned + - card_feedback + +# Site / deployment metadata +site: + name: "" + environment: development # development | staging | production +""" + -def new_project(name: str): +def new_project(name: str, template: str): """Scaffold a new HealthChain project.""" project_dir = Path(name) if project_dir.exists(): - print(f"❌ Directory '{name}' already exists.") + print(f"Error: directory '{name}' already exists.") return project_dir.mkdir() - (project_dir / "app.py").write_text(_APP_PY) - (project_dir / ".env.example").write_text(_ENV_EXAMPLE) + + _app_py = { + "cds-hooks": _APP_PY_CDS_HOOKS, + "fhir-gateway": _APP_PY_FHIR_GATEWAY, + "default": _APP_PY_DEFAULT, + } + _env_example = { + "cds-hooks": _ENV_EXAMPLE_CDS_HOOKS, + "fhir-gateway": _ENV_EXAMPLE_FHIR_GATEWAY, + "default": _ENV_EXAMPLE_DEFAULT, + } + service_type = template if template != "default" else "cds-hooks" + + (project_dir / "app.py").write_text(_app_py[template]) + (project_dir / "healthchain.yaml").write_text( + _make_healthchain_yaml(name, service_type) + ) + (project_dir / ".env.example").write_text(_env_example[template]) (project_dir / "requirements.txt").write_text(_REQUIREMENTS) (project_dir / "Dockerfile").write_text(_DOCKERFILE) (project_dir / ".dockerignore").write_text(_DOCKERIGNORE) - print(f"\nCreated project '{name}/'") - print(f" {name}/app.py") - print(f" {name}/.env.example") - print(f" {name}/requirements.txt") - print(f" {name}/Dockerfile") - print("\nNext steps:") - print(f" 1. Build your app in {name}/app.py") - print(" 2. Copy .env.example to .env and fill in your credentials") - print(f" 3. healthchain serve app:app (from inside {name}/)") - print(f" 4. docker build -t {name} {name}/") - print(f" 5. docker run -p 8000:8000 --env-file {name}/.env {name}") - print("\nUsing format conversion? Run: healthchain eject-templates ./configs") - print("See https://dotimplement.github.io/HealthChain/ for examples.") + print(f"\n{_BOLD}{_GREEN}✚ Created project '{name}/'{_RST}") + print(f" {_CYAN}{name}/app.py{_RST}") + print(f" {_CYAN}{name}/healthchain.yaml{_RST}") + print(f" {_CYAN}{name}/.env.example{_RST}") + print(f" {_CYAN}{name}/requirements.txt{_RST}") + print(f" {_CYAN}{name}/Dockerfile{_RST}") + print(f"\n{_BOLD}Next steps:{_RST}") + print(f" {_BOLD}cd {name}{_RST}") + if template == "cds-hooks": + print( + f" {_BOLD}healthchain serve{_RST} {_DIM}# starts the CDS Hooks service{_RST}" + ) + print(f" {_BOLD}open http://localhost:8000/docs{_RST}") + elif template == "fhir-gateway": + print( + f" {_BOLD}cp .env.example .env{_RST} {_DIM}# fill in your FHIR source credentials{_RST}" + ) + print( + f" {_BOLD}healthchain serve{_RST} {_DIM}# starts the FHIR gateway{_RST}" + ) + print(f" {_BOLD}open http://localhost:8000/docs{_RST}") + else: + print(f" {_DIM}# Pick a template to get started:{_RST}") + print(f" {_BOLD}healthchain new my-app --template cds-hooks{_RST}") + print(f" {_BOLD}healthchain new my-app --template fhir-gateway{_RST}") + print(f"\n{_INDIGO}Configure your app in healthchain.yaml{_RST}") + print(f"{_DIM}See https://dotimplement.github.io/HealthChain/ for examples.{_RST}") def eject_templates(target_dir: str): @@ -127,53 +341,236 @@ def eject_templates(target_dir: str): from healthchain.interop import init_config_templates target_path = init_config_templates(target_dir) - print(f"\n✅ Templates ejected to: {target_path}") - print("\nNext steps:") + print(f"\n{_GREEN}✓ Templates ejected to: {_BOLD}{target_path}{_RST}") + print(f"\n{_BOLD}Next steps:{_RST}") print(" 1. Customize the templates in the created directory") print(" 2. Use them in your code:") - print(" from healthchain.interop import create_interop") - print(f" engine = create_interop(config_dir='{target_dir}')") + print(f" {_DIM}from healthchain.interop import create_interop{_RST}") + print(f" {_DIM}engine = create_interop(config_dir='{target_dir}'){_RST}") print( - "\nSee https://dotimplement.github.io/HealthChain/reference/interop/ for details." + f"\n{_DIM}See https://dotimplement.github.io/HealthChain/reference/interop/ for details.{_RST}" ) except FileExistsError as e: - print(f"❌ Error: {str(e)}") - print("💡 Choose a different directory name or remove the existing one.") + print(f"\n{_RED}Error: {str(e)}{_RST}") + print( + f"{_DIM}Choose a different directory name or remove the existing one.{_RST}" + ) except Exception as e: - print(f"❌ Error ejecting templates: {str(e)}") - print("💡 Make sure HealthChain is properly installed.") + print(f"\n{_RED}Error ejecting templates: {str(e)}{_RST}") + print(f"{_DIM}Make sure HealthChain is properly installed.{_RST}") -def serve(app_module: str, host: str, port: int): +def serve(app_module: str, host: str, port: int | None): """Start a HealthChain app with uvicorn.""" + from healthchain.config.appconfig import AppConfig + + config = AppConfig.load() + + # Resolve port: CLI arg > config > default + resolved_port = ( + port if port is not None else (config.service.port if config else 8000) + ) + + cmd = [ + sys.executable, + "-m", + "uvicorn", + app_module, + "--host", + host, + "--port", + str(resolved_port), + ] + + if config and config.security.tls.enabled: + cmd += [ + "--ssl-certfile", + config.security.tls.cert_path, + "--ssl-keyfile", + config.security.tls.key_path, + ] + try: - subprocess.run( - [ - sys.executable, - "-m", - "uvicorn", - app_module, - "--host", - host, - "--port", - str(port), - ], - check=True, - ) + subprocess.run(cmd, check=True) except subprocess.CalledProcessError as e: print(f"❌ Server error: {e}") except KeyboardInterrupt: pass +def sandbox_run( + url: str, + workflow: str, + size: int, + from_path: str | None, + output: str | None, + no_save: bool, +): + """Fire test requests at a running HealthChain service.""" + from healthchain.config.appconfig import AppConfig + from healthchain.sandbox import SandboxClient + + config = AppConfig.load() + resolved_output = output or (config.data.output_dir if config else "./output") + + print(f"\n{_BOLD}{_CYAN}◆ Sandbox{_RST} {_DIM}{url}{_RST}") + print(f" {_CYAN}workflow {_RST}{workflow}") + + try: + client = SandboxClient(url=url, workflow=workflow) + except ValueError as e: + print(f"\n{_RED}Error:{_RST} {e}") + return + + if from_path: + print(f"\n{_DIM}Loading from {from_path}...{_RST}") + try: + client.load_from_path(from_path) + except (FileNotFoundError, ValueError) as e: + print(f"{_RED}Error loading data:{_RST} {e}") + return + else: + print(f"\n{_DIM}Generating {size} synthetic request(s)...{_RST}") + try: + client.load_synthetic(n=size) + except ValueError as e: + print(f"{_RED}Error generating synthetic data:{_RST} {e}") + return + + print(f"{_DIM}Sending {len(client.requests)} request(s) to service...{_RST}\n") + + try: + responses = client.send_requests() + except RuntimeError as e: + print(f"{_RED}Error:{_RST} {e}") + return + + # Print response summary + success = sum(1 for r in responses if r) + result_col = _GREEN if success == len(responses) else _AMBER + print( + f"{_BOLD}Results:{_RST} {result_col}{success}/{len(responses)} successful{_RST}\n" + ) + + indicator_colours = { + "INFO": _CYAN, + "WARNING": _AMBER, + "CRITICAL": _RED, + "SUCCESS": _GREEN, + } + for i, response in enumerate(responses): + cards = response.get("cards", []) + if not cards: + print(f" {_DIM}[{i + 1}] no cards returned{_RST}") + continue + print(f" {_BOLD}[{i + 1}]{_RST} {len(cards)} card(s)") + for card in cards: + indicator = card.get("indicator", "info").upper() + summary = card.get("summary", "") + ind_col = indicator_colours.get(indicator, _DIM) + print(f" {ind_col}{indicator}{_RST} {summary}") + + if not no_save: + client.save_results(resolved_output) + print(f"\n{_GREEN}✓{_RST} Saved to {_BOLD}{resolved_output}/{_RST}") + + +def status(): + """Show current project status from healthchain.yaml.""" + from healthchain.config.appconfig import AppConfig + + config = AppConfig.load() + + if config is None: + print(f"\n{_AMBER}No healthchain.yaml found in current directory.{_RST}") + print(f"Run {_BOLD}healthchain new {_RST} to scaffold a project.") + return + + def _key(k: str) -> str: + return f" {_CYAN}{k}{_RST}" + + def _val_on(v: str) -> str: + return f"{_GREEN}{v}{_RST}" + + def _val_off(v: str) -> str: + return f"{_RED}{v}{_RST}" + + def _val_env(e: str) -> str: + c = { + "production": _GREEN, + "staging": _CYAN, + "development": _AMBER, + }.get(e, _RST) + return f"{c}{e}{_RST}" + + def _section(s: str) -> str: + return f"\n{_BOLD}{_INDIGO}{s}{_RST}" + + print( + f"\n{_BOLD}{_PINK}✚ HealthChain{_RST} " + f"{_BOLD}{config.name}{_RST} {_DIM}v{config.version}{_RST}" + ) + print(f"{_INDIGO}{'─' * 40}{_RST}") + + print(_section("Service")) + print(f"{_key('type ')}{config.service.type}") + print(f"{_key('port ')}{_BOLD}{config.service.port}{_RST}") + + print(_section("Site")) + print(f"{_key('environment ')}{_val_env(config.site.environment)}") + if config.site.name: + print(f"{_key('name ')}{config.site.name}") + + print(_section("Security")) + auth_col = _GREEN if config.security.auth != "none" else _AMBER + print(f"{_key('auth ')}{auth_col}{config.security.auth}{_RST}") + tls_val = ( + _val_on("enabled") if config.security.tls.enabled else _val_off("disabled") + ) + print(f"{_key('TLS ')}{tls_val}") + origins = ", ".join(config.security.allowed_origins) + print(f"{_key('origins ')}{_DIM}{origins}{_RST}") + + print(_section("Compliance")) + hipaa_val = _val_on("enabled") if config.compliance.hipaa else _val_off("disabled") + print(f"{_key('HIPAA ')}{hipaa_val}") + if config.compliance.hipaa: + print(f"{_key('audit log ')}{_BOLD}{config.compliance.audit_log}{_RST}") + + print(_section("Eval")) + if config.eval.enabled: + print(f"{_key('provider ')}{config.eval.provider}") + print(f"{_key('tracking ')}{_BOLD}{config.eval.tracking_uri}{_RST}") + print(f"{_key('events ')}{_DIM}{', '.join(config.eval.track)}{_RST}") + else: + print(f" {_DIM}disabled{_RST}") + + print() + + def main(): - parser = argparse.ArgumentParser(description="HealthChain command-line interface") + parser = argparse.ArgumentParser( + description=( + f"{_BOLD}{_CYAN}✚ HealthChain{_RST} {_DIM}Open-Source Healthcare AI{_RST}\n" + f"{_INDIGO}{'─' * 40}{_RST}" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) subparsers = parser.add_subparsers(dest="command", required=True) # Subparser for the 'new' command new_parser = subparsers.add_parser("new", help="Scaffold a new HealthChain project") new_parser.add_argument("name", type=str, help="Project name (creates a directory)") + new_parser.add_argument( + "--type", + "-t", + type=str, + default="default", + choices=["cds-hooks", "fhir-gateway", "default"], + help="Project type (default: empty stub)", + dest="template", + ) # Subparser for the 'serve' command serve_parser = subparsers.add_parser( @@ -190,8 +587,62 @@ def main(): "--host", type=str, default="0.0.0.0", help="Host (default: 0.0.0.0)" ) serve_parser.add_argument( - "--port", type=int, default=8000, help="Port (default: 8000)" + "--port", + type=int, + default=None, + help="Port (default: from healthchain.yaml or 8000)", + ) + + # Subparser for the 'sandbox' command + sandbox_parser = subparsers.add_parser( + "sandbox", help="Send test requests to a running HealthChain service" + ) + sandbox_subparsers = sandbox_parser.add_subparsers( + dest="sandbox_command", required=True ) + sandbox_run_parser = sandbox_subparsers.add_parser( + "run", help="Fire test requests at a service" + ) + sandbox_run_parser.add_argument( + "--url", + type=str, + required=True, + help="Full service URL (e.g. http://localhost:8000/cds/cds-services/my-service)", + ) + sandbox_run_parser.add_argument( + "--workflow", + type=str, + default="patient-view", + choices=["patient-view", "order-select", "order-sign", "encounter-discharge"], + help="CDS workflow to simulate (default: patient-view)", + ) + sandbox_run_parser.add_argument( + "--size", + type=int, + default=3, + help="Number of synthetic requests to generate (default: 3)", + ) + sandbox_run_parser.add_argument( + "--from-path", + type=str, + default=None, + metavar="PATH", + help="Load requests from a file or directory instead of generating synthetic data", + ) + sandbox_run_parser.add_argument( + "--output", + type=str, + default=None, + help="Directory to save results (default: from healthchain.yaml or ./output)", + ) + sandbox_run_parser.add_argument( + "--no-save", + action="store_true", + help="Don't save results to disk", + ) + + # Subparser for the 'status' command + subparsers.add_parser("status", help="Show project status from healthchain.yaml") # Subparser for the 'eject-templates' command eject_parser = subparsers.add_parser( @@ -209,9 +660,21 @@ def main(): args = parser.parse_args() if args.command == "new": - new_project(args.name) + new_project(args.name, args.template) elif args.command == "serve": serve(args.app_module, args.host, args.port) + elif args.command == "sandbox": + if args.sandbox_command == "run": + sandbox_run( + url=args.url, + workflow=args.workflow, + size=args.size, + from_path=args.from_path, + output=args.output, + no_save=args.no_save, + ) + elif args.command == "status": + status() elif args.command == "eject-templates": eject_templates(args.target_dir) diff --git a/healthchain/config/__init__.py b/healthchain/config/__init__.py index 11e6b739..d6814de6 100644 --- a/healthchain/config/__init__.py +++ b/healthchain/config/__init__.py @@ -6,6 +6,7 @@ from various sources. """ +from healthchain.config.appconfig import AppConfig from healthchain.config.base import ( ConfigManager, ValidationLevel, @@ -18,6 +19,7 @@ ) __all__ = [ + "AppConfig", "ConfigManager", "ValidationLevel", "validate_cda_section_config_model", diff --git a/healthchain/config/appconfig.py b/healthchain/config/appconfig.py new file mode 100644 index 00000000..9fa7bdde --- /dev/null +++ b/healthchain/config/appconfig.py @@ -0,0 +1,104 @@ +""" +App-level configuration model for HealthChain projects. + +Loads and validates healthchain.yaml from the project root. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import List, Optional + +import yaml +from pydantic import BaseModel, field_validator + +logger = logging.getLogger(__name__) + +_CONFIG_FILENAME = "healthchain.yaml" + + +class ServiceConfig(BaseModel): + type: str = "cds-hooks" + port: int = 8000 + + +class DataConfig(BaseModel): + patients_dir: str = "./data" + output_dir: str = "./output" + + +class TLSConfig(BaseModel): + enabled: bool = False + cert_path: str = "./certs/server.crt" + key_path: str = "./certs/server.key" + + +class SecurityConfig(BaseModel): + auth: str = "none" + tls: TLSConfig = TLSConfig() + allowed_origins: List[str] = ["*"] + + @field_validator("auth") + @classmethod + def validate_auth(cls, v: str) -> str: + allowed = {"none", "api-key", "smart-on-fhir"} + if v not in allowed: + raise ValueError(f"auth must be one of: {', '.join(sorted(allowed))}") + return v + + +class ComplianceConfig(BaseModel): + hipaa: bool = False + audit_log: str = "./logs/audit.jsonl" + + +class EvalConfig(BaseModel): + enabled: bool = False + provider: str = "mlflow" + tracking_uri: str = "./mlruns" + track: List[str] = ["model_inference", "cds_card_returned", "card_feedback"] + + +class SiteConfig(BaseModel): + name: str = "" + environment: str = "development" + + @field_validator("environment") + @classmethod + def validate_environment(cls, v: str) -> str: + allowed = {"development", "staging", "production"} + if v not in allowed: + raise ValueError( + f"environment must be one of: {', '.join(sorted(allowed))}" + ) + return v + + +class AppConfig(BaseModel): + name: str = "my-healthchain-app" + version: str = "1.0.0" + service: ServiceConfig = ServiceConfig() + data: DataConfig = DataConfig() + security: SecurityConfig = SecurityConfig() + compliance: ComplianceConfig = ComplianceConfig() + eval: EvalConfig = EvalConfig() + site: SiteConfig = SiteConfig() + + @classmethod + def from_yaml(cls, path: Path) -> "AppConfig": + """Load AppConfig from a YAML file.""" + with open(path) as f: + data = yaml.safe_load(f) or {} + return cls(**data) + + @classmethod + def load(cls) -> Optional["AppConfig"]: + """Load healthchain.yaml from the current working directory if it exists.""" + config_path = Path(_CONFIG_FILENAME) + if config_path.exists(): + try: + return cls.from_yaml(config_path) + except Exception as e: + logger.warning(f"Failed to load {_CONFIG_FILENAME}: {e}") + return None diff --git a/healthchain/gateway/api/app.py b/healthchain/gateway/api/app.py index 1ed8ab86..bfd74573 100644 --- a/healthchain/gateway/api/app.py +++ b/healthchain/gateway/api/app.py @@ -6,6 +6,7 @@ """ import logging +import re from contextlib import asynccontextmanager from datetime import datetime @@ -13,7 +14,6 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse -from termcolor import colored from typing import Dict, Optional, Type, Union @@ -23,6 +23,181 @@ logger = logging.getLogger(__name__) +# ── Half-block pixel font (4 wide × 3 tall, encodes 2 logical rows per char) ── +_HB = { + "H": ["█ █", "█▀▀█", "▀ ▀"], + "E": ["█▀▀▀", "█▀▀ ", "▀▀▀▀"], + "A": ["▄▀▀▄", "█▀▀█", "▀ ▀"], + "L": ["█ ", "█ ", "▀▀▀▀"], + "T": ["▀██▀", " ██ ", " ▀▀ "], + "C": ["▄▀▀▀", "█ ", " ▀▀▀"], + "I": ["▀██▀", " ██ ", "▀▀▀▀"], + "N": ["█▄ █", "█ ▀█", "▀ ▀"], +} + + +def _render_word(word: str) -> list[str]: + rows = [""] * 3 + for j, ch in enumerate(word): + for r in range(3): + rows[r] += _HB[ch][r] + (" " if j < len(word) - 1 else "") + return rows + + +def _gradient(text: str, t0: float, t1: float) -> str: + r0, g0, b0 = 0, 215, 255 + r1, g1, b1 = 192, 132, 252 + visible = [(i, c) for i, c in enumerate(text) if c != " "] + total = max(len(visible) - 1, 1) + chars = list(text) + for idx, (i, c) in enumerate(visible): + t = t0 + (t1 - t0) * (idx / total) + r = int(r0 + (r1 - r0) * t) + g = int(g0 + (g1 - g0) * t) + b = int(b0 + (b1 - b0) * t) + chars[i] = f"\033[38;2;{r};{g};{b}m{c}\033[0m" + return "".join(chars) + + +def _vlen(s: str) -> int: + return len(re.sub(r"\033\[[^m]*m", "", s)) + + +def _pad(s: str, width: int) -> str: + return s + " " * max(0, width - _vlen(s)) + + +def _val_bool(enabled: bool) -> str: + return ( + "\033[38;2;0;255;135m✓ enabled\033[0m" + if enabled + else "\033[38;2;255;85;85m✗ disabled\033[0m" + ) + + +def _val_env(env: str) -> str: + c = {"production": "\033[38;2;0;255;135m", "staging": "\033[38;2;0;215;255m"}.get( + env, "\033[38;2;255;200;50m" + ) + return f"{c}{env}\033[0m" + + +def _val_auth(auth: str) -> str: + if auth == "none": + return "\033[38;2;255;200;50mnone\033[0m" + return f"\033[38;2;0;255;135m{auth}\033[0m" + + +def _val_eval(enabled: bool, provider: str) -> str: + if enabled: + return f"\033[38;2;0;255;135m✓ {provider}\033[0m" + return "\033[2m✗ disabled\033[0m" + + +def _status_row(key: str, value: str) -> str: + return f"\033[1m\033[38;2;0;255;135m{key}\033[0m {value}" + + +def _print_startup_banner( + title: str, + version: str, + gateways: dict, + services: dict, + docs_url: str, + config=None, + config_path: Optional[str] = None, +) -> None: + """Print pixel-wordmark banner with live status panel.""" + health_rows = _render_word("HEALTH") + chain_rows = _render_word("CHAIN") + health_w = len(health_rows[0]) + chain_w = len(chain_rows[0]) + center_pad = (health_w - chain_w) // 2 + LOGO_COL = 38 + + # ── resolve status values from config or sensible defaults ── + svc_type = (config.service.type if config else None) or ( + list({**gateways, **services}.keys())[0] + if {**gateways, **services} + else "unknown" + ) + env = config.site.environment if config else "development" + port = str(config.service.port if config else 8000) + site = config.site.name if config else None + auth = config.security.auth if config else "none" + tls = config.security.tls.enabled if config else False + hipaa = config.compliance.hipaa if config else False + eval_enabled = config.eval.enabled if config else False + eval_provider = config.eval.provider if config else "mlflow" + fhir_configured = any( + hasattr(gw, "sources") and getattr(gw, "sources", None) + for gw in gateways.values() + ) + + status: list[str] = [ + f"\033[1m\033[38;2;255;121;198m{title}\033[0m \033[2mv{version}\033[0m", + "", + _status_row("type: ", svc_type), + _status_row("environment:", _val_env(env)), + _status_row("port: ", f"\033[1m{port}\033[0m"), + ] + if site: + status.append(_status_row("site: ", site)) + status += [ + "", + _status_row("auth: ", _val_auth(auth)), + _status_row( + "fhir creds: ", + "\033[38;2;0;255;135m✓ configured\033[0m" + if fhir_configured + else "\033[38;2;255;200;50m✗ not set\033[0m", + ), + _status_row("tls: ", _val_bool(tls)), + _status_row("hipaa: ", _val_bool(hipaa)), + _status_row("eval: ", _val_eval(eval_enabled, eval_provider)), + "", + _status_row("config: ", f"\033[1m{config_path or '(none)'}\033[0m"), + _status_row("docs: ", f"\033[1mhttp://localhost:{port}{docs_url}\033[0m"), + ] + + # ── build logo lines, vertically centred against status height ── + n = 3 + inner: list[str] = [] + for i in range(n): + inner.append( + _pad( + " " + _gradient(health_rows[i], i / (n * 2), 0.5 + i / (n * 2)), + LOGO_COL, + ) + ) + inner.append(" " * LOGO_COL) + for i in range(n): + inner.append( + _pad( + " " + + " " * center_pad + + _gradient(chain_rows[i], 0.1 + i / (n * 2), 0.6 + i / (n * 2)), + LOGO_COL, + ) + ) + + top = (len(status) - len(inner)) // 2 + bot = len(status) - len(inner) - top + logo_lines = [" " * LOGO_COL] * top + inner + [" " * LOGO_COL] * bot + + max_status_w = max(_vlen(s) for s in status) + inner_w = LOGO_COL + 3 + max_status_w + border = "\033[38;2;99;102;241m" # indigo + rst = "\033[0m" + + print() + print(f"{border}╭{'─' * (inner_w + 2)}╮{rst}") + for logo_line, s in zip(logo_lines, status): + padding = " " * (max_status_w - _vlen(s)) + print(f"{border}│{rst} {logo_line} {s}{padding} {border}│{rst}") + print(f"{border}╰{'─' * (inner_w + 2)}╯{rst}") + print() + class HealthChainAPI(FastAPI): """ @@ -110,9 +285,13 @@ def __init__( # Setup middleware if enable_cors: + from healthchain.config.appconfig import AppConfig + + _config = AppConfig.load() + origins = _config.security.allowed_origins if _config else ["*"] self.add_middleware( CORSMiddleware, - allow_origins=["*"], # Can be configured from settings + allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], @@ -415,22 +594,18 @@ async def _exception_handler( async def _startup(self) -> None: """Display startup information and initialize components.""" - # Display banner - banner = r""" - __ __ ____ __ ________ _ - / / / /__ ____ _/ / /_/ /_ / ____/ /_ ____ _(_)___ - / /_/ / _ \/ __ `/ / __/ __ \/ / / __ \/ __ `/ / __ \ - / __ / __/ /_/ / / /_/ / / / /___/ / / / /_/ / / / / / -/_/ /_/\___/\__,_/_/\__/_/ /_/\____/_/ /_/\__,_/_/_/ /_/ -""" - colors = ["red", "yellow", "green", "cyan", "blue", "magenta"] - for i, line in enumerate(banner.split("\n")): - print(colored(line, colors[i % len(colors)])) - - # Log startup info - logger.info(f"🚀 Starting {self.title} v{self.version}") - logger.info(f"Gateways: {list(self.gateways.keys())}") - logger.info(f"Services: {list(self.services.keys())}") + from healthchain.config.appconfig import AppConfig + + config = AppConfig.load() + _print_startup_banner( + title=config.name if config else self.title, + version=config.version if config else self.version, + gateways=self.gateways, + services=self.services, + docs_url=self.docs_url or "http://localhost:8000/docs", + config=config, + config_path="./healthchain.yaml" if config else None, + ) # Initialize components for name, component in {**self.gateways, **self.services}.items(): @@ -441,13 +616,8 @@ async def _startup(self) -> None: except Exception as e: logger.warning(f"Failed to initialize {name}: {e}") - logger.info(f"📖 Docs: {self.docs_url}") - async def _shutdown(self) -> None: """Handle graceful shutdown.""" - logger.info("🛑 Shutting down...") - - # Shutdown all components for name, component in {**self.services, **self.gateways}.items(): if hasattr(component, "shutdown") and callable(component.shutdown): try: @@ -455,5 +625,3 @@ async def _shutdown(self) -> None: logger.debug(f"Shutdown: {name}") except Exception as e: logger.warning(f"Failed to shutdown {name}: {e}") - - logger.info("✅ Shutdown completed") diff --git a/healthchain/sandbox/sandboxclient.py b/healthchain/sandbox/sandboxclient.py index 6919e8b3..96be54b3 100644 --- a/healthchain/sandbox/sandboxclient.py +++ b/healthchain/sandbox/sandboxclient.py @@ -327,6 +327,46 @@ def _construct_request(self, data: Union[Dict[str, Any], Any]) -> None: self.requests.append(request) + def load_synthetic( + self, + n: int = 1, + random_seed: Optional[int] = None, + ) -> "SandboxClient": + """ + Generate n synthetic CDS requests for the configured workflow. + + Useful for quickly testing a service without real patient data. + Supports patient-view and encounter-discharge workflows. + + Args: + n: Number of synthetic requests to generate (default: 1) + random_seed: Seed for reproducible results. Each request gets + seed + i so they produce different data. + + Returns: + Self for method chaining + + Raises: + ValueError: If the workflow is not supported for synthetic generation + """ + from .generators import CdsDataGenerator + + generator = CdsDataGenerator() + generator.set_workflow(self.workflow) + + for i in range(n): + seed = random_seed + i if random_seed is not None else None + prefetch_data = generator.generate_prefetch( + random_seed=seed, + generate_resources=True, + ) + self._construct_request(prefetch_data) + + log.info( + f"Generated {n} synthetic request(s) for workflow {self.workflow.value}" + ) + return self + def clear_requests(self) -> "SandboxClient": """ Clear all queued requests. diff --git a/mkdocs.yml b/mkdocs.yml index 77d7c3ae..bc87c06e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -33,6 +33,7 @@ nav: - Format Conversion: cookbook/format_conversion.md - Docs: - Welcome: reference/index.md + - Configuration: reference/config.md - Gateway: - Overview: reference/gateway/gateway.md - HealthChainAPI: reference/gateway/api.md diff --git a/tests/config_manager/test_appconfig.py b/tests/config_manager/test_appconfig.py new file mode 100644 index 00000000..5ef01c52 --- /dev/null +++ b/tests/config_manager/test_appconfig.py @@ -0,0 +1,132 @@ +"""Tests for AppConfig — app-level configuration loaded from healthchain.yaml.""" + +import pytest + +from healthchain.config.appconfig import AppConfig + + +def test_appconfig_loads_valid_yaml(tmp_path): + """AppConfig.from_yaml parses a valid healthchain.yaml correctly.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text( + """ +name: my-app +version: "2.0.0" +service: + type: fhir-gateway + port: 9000 +site: + environment: production + name: Test Hospital +""" + ) + config = AppConfig.from_yaml(config_file) + + assert config.name == "my-app" + assert config.version == "2.0.0" + assert config.service.type == "fhir-gateway" + assert config.service.port == 9000 + assert config.site.environment == "production" + assert config.site.name == "Test Hospital" + + +def test_appconfig_missing_fields_use_defaults(tmp_path): + """AppConfig.from_yaml fills in defaults for any omitted fields.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text("name: minimal-app\n") + + config = AppConfig.from_yaml(config_file) + + assert config.name == "minimal-app" + assert config.service.port == 8000 + assert config.service.type == "cds-hooks" + assert config.security.auth == "none" + assert config.security.tls.enabled is False + assert config.compliance.hipaa is False + assert config.eval.enabled is False + assert config.site.environment == "development" + + +def test_appconfig_invalid_auth_raises(tmp_path): + """AppConfig raises ValueError for unrecognised auth method.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text("security:\n auth: magic-token\n") + + with pytest.raises(Exception): + AppConfig.from_yaml(config_file) + + +def test_appconfig_invalid_environment_raises(tmp_path): + """AppConfig raises ValueError for unrecognised environment value.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text("site:\n environment: staging-uat\n") + + with pytest.raises(Exception): + AppConfig.from_yaml(config_file) + + +def test_appconfig_load_returns_none_when_no_file(tmp_path, monkeypatch): + """AppConfig.load returns None when healthchain.yaml is not present.""" + monkeypatch.chdir(tmp_path) + assert AppConfig.load() is None + + +def test_appconfig_load_returns_config_when_file_present(tmp_path, monkeypatch): + """AppConfig.load reads healthchain.yaml from the current directory.""" + (tmp_path / "healthchain.yaml").write_text("name: loaded-app\n") + monkeypatch.chdir(tmp_path) + + config = AppConfig.load() + + assert config is not None + assert config.name == "loaded-app" + + +def test_appconfig_load_returns_none_on_parse_error(tmp_path, monkeypatch): + """AppConfig.load returns None and logs a warning when the file is malformed.""" + (tmp_path / "healthchain.yaml").write_text("security:\n auth: bad-value\n") + monkeypatch.chdir(tmp_path) + + # Should not raise — returns None gracefully + result = AppConfig.load() + assert result is None + + +def test_appconfig_tls_config_parsed(tmp_path): + """AppConfig parses nested TLS config correctly.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text( + """ +security: + tls: + enabled: true + cert_path: ./certs/cert.pem + key_path: ./certs/key.pem +""" + ) + config = AppConfig.from_yaml(config_file) + + assert config.security.tls.enabled is True + assert config.security.tls.cert_path == "./certs/cert.pem" + assert config.security.tls.key_path == "./certs/key.pem" + + +def test_appconfig_eval_track_events_parsed(tmp_path): + """AppConfig parses eval.track list correctly.""" + config_file = tmp_path / "healthchain.yaml" + config_file.write_text( + """ +eval: + enabled: true + provider: langfuse + track: + - model_inference + - card_feedback +""" + ) + config = AppConfig.from_yaml(config_file) + + assert config.eval.enabled is True + assert config.eval.provider == "langfuse" + assert "model_inference" in config.eval.track + assert "card_feedback" in config.eval.track diff --git a/tests/sandbox/test_sandbox_client.py b/tests/sandbox/test_sandbox_client.py index 75ae184d..2a9177f0 100644 --- a/tests/sandbox/test_sandbox_client.py +++ b/tests/sandbox/test_sandbox_client.py @@ -458,6 +458,46 @@ def test_send_requests_soap_success(mock_client_class): assert mock_client.post.called +def test_load_synthetic_generates_correct_number_of_requests(): + """load_synthetic generates the requested number of CDS requests.""" + client = SandboxClient(url="http://localhost:8000/test", workflow="patient-view") + + result = client.load_synthetic(n=3) + + assert len(client.requests) == 3 + assert result is client # method chaining + + +def test_load_synthetic_default_generates_one_request(): + """load_synthetic defaults to n=1.""" + client = SandboxClient(url="http://localhost:8000/test", workflow="patient-view") + + client.load_synthetic() + + assert len(client.requests) == 1 + + +def test_load_synthetic_accepts_random_seed(): + """load_synthetic accepts a random_seed without raising.""" + client = SandboxClient(url="http://localhost:8000/test", workflow="patient-view") + + client.load_synthetic(n=2, random_seed=42) + + assert len(client.requests) == 2 + + +def test_load_synthetic_supports_encounter_discharge_workflow(): + """load_synthetic works with encounter-discharge workflow.""" + client = SandboxClient( + url="http://localhost:8000/test", workflow="encounter-discharge" + ) + + client.load_synthetic(n=2) + + assert len(client.requests) == 2 + assert all(r.hook == "encounter-discharge" for r in client.requests) + + @patch("httpx.Client") def test_send_requests_handles_multiple_requests(mock_client_class): """send_requests processes multiple queued requests sequentially.""" diff --git a/tests/test_cli.py b/tests/test_cli.py index d1448e47..684cd015 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,9 +2,9 @@ import pytest import subprocess -from unittest.mock import patch +from unittest.mock import patch, MagicMock -from healthchain.cli import eject_templates, serve, main +from healthchain.cli import eject_templates, new_project, serve, status, main @pytest.mark.parametrize( @@ -13,15 +13,15 @@ ( FileExistsError("Directory already exists"), [ - "❌ Error: Directory already exists", - "💡 Choose a different directory name or remove the existing one", + "Error: Directory already exists", + "Choose a different directory name or remove the existing one", ], ), ( Exception("Something went wrong"), [ - "❌ Error ejecting templates: Something went wrong", - "💡 Make sure HealthChain is properly installed", + "Error ejecting templates: Something went wrong", + "Make sure HealthChain is properly installed", ], ), ], @@ -36,7 +36,6 @@ def test_eject_templates_error_handling_provides_helpful_guidance( with patch("builtins.print") as mock_print: eject_templates("./test_configs") - # Verify helpful error messages are displayed for expected_msg in expected_messages: assert any(expected_msg in str(call) for call in mock_print.call_args_list) @@ -50,32 +49,205 @@ def test_eject_templates_success_provides_usage_instructions(mock_init_templates with patch("builtins.print") as mock_print: eject_templates(target_dir) - # Verify success message and usage instructions are provided print_output = " ".join(str(call) for call in mock_print.call_args_list) - assert "✅ Templates ejected to:" in print_output + assert "Templates ejected to:" in print_output assert "create_interop(config_dir=" in print_output assert "Next steps:" in print_output @patch("subprocess.run") -def test_serve_handles_execution_errors_gracefully(mock_run): +@patch("healthchain.config.appconfig.AppConfig.load", return_value=None) +def test_serve_handles_execution_errors_gracefully(mock_config, mock_run): """serve provides clear error message when server fails to start.""" mock_run.side_effect = subprocess.CalledProcessError(1, "uvicorn") with patch("builtins.print") as mock_print: - serve("app:app", "0.0.0.0", 8000) + serve("app:app", "0.0.0.0", None) - # Verify error message is informative error_message = mock_print.call_args[0][0] assert "❌ Server error:" in error_message +@patch("subprocess.run") +@patch("healthchain.config.appconfig.AppConfig.load") +def test_serve_reads_port_from_config(mock_load, mock_run): + """serve uses port from healthchain.yaml when --port is not provided.""" + mock_config = MagicMock() + mock_config.service.port = 9000 + mock_config.security.tls.enabled = False + mock_load.return_value = mock_config + + serve("app:app", "0.0.0.0", None) + + call_args = mock_run.call_args[0][0] + assert "--port" in call_args + assert "9000" in call_args + + +@patch("subprocess.run") +@patch("healthchain.config.appconfig.AppConfig.load") +def test_serve_cli_port_overrides_config(mock_load, mock_run): + """serve uses --port flag value over healthchain.yaml port.""" + mock_config = MagicMock() + mock_config.service.port = 9000 + mock_config.security.tls.enabled = False + mock_load.return_value = mock_config + + serve("app:app", "0.0.0.0", 8080) + + call_args = mock_run.call_args[0][0] + assert "8080" in call_args + assert "9000" not in call_args + + +@patch("subprocess.run") +@patch("healthchain.config.appconfig.AppConfig.load", return_value=None) +def test_serve_defaults_to_8000_without_config(mock_load, mock_run): + """serve defaults to port 8000 when no config and no --port flag.""" + serve("app:app", "0.0.0.0", None) + + call_args = mock_run.call_args[0][0] + assert "8000" in call_args + + +@patch("subprocess.run") +@patch("healthchain.config.appconfig.AppConfig.load") +def test_serve_passes_tls_args_when_enabled(mock_load, mock_run): + """serve passes ssl-certfile and ssl-keyfile to uvicorn when TLS is enabled.""" + mock_config = MagicMock() + mock_config.service.port = 8000 + mock_config.security.tls.enabled = True + mock_config.security.tls.cert_path = "./certs/cert.pem" + mock_config.security.tls.key_path = "./certs/key.pem" + mock_load.return_value = mock_config + + serve("app:app", "0.0.0.0", None) + + call_args = mock_run.call_args[0][0] + assert "--ssl-certfile" in call_args + assert "--ssl-keyfile" in call_args + + +def test_new_project_creates_expected_files(tmp_path): + """new_project creates all expected files in a new directory.""" + project_dir = tmp_path / "my-app" + + with patch("builtins.print"): + new_project(str(project_dir), "default") + + assert (project_dir / "app.py").exists() + assert (project_dir / "healthchain.yaml").exists() + assert (project_dir / ".env.example").exists() + assert (project_dir / "requirements.txt").exists() + assert (project_dir / "Dockerfile").exists() + + +def test_new_project_cds_hooks_template_generates_working_app(tmp_path): + """new_project with cds-hooks template generates a non-empty app.py with CDS imports.""" + project_dir = tmp_path / "my-cds-app" + + with patch("builtins.print"): + new_project(str(project_dir), "cds-hooks") + + app_py = (project_dir / "app.py").read_text() + assert "CDSHooksService" in app_py + assert "CDSResponse" in app_py + assert "HealthChainAPI" in app_py + + +def test_new_project_fhir_gateway_template_generates_working_app(tmp_path): + """new_project with fhir-gateway template generates a non-empty app.py with gateway imports.""" + project_dir = tmp_path / "my-fhir-app" + + with patch("builtins.print"): + new_project(str(project_dir), "fhir-gateway") + + app_py = (project_dir / "app.py").read_text() + assert "FHIRGateway" in app_py + assert "HealthChainAPI" in app_py + assert "merge_bundles" in app_py + + +def test_new_project_default_template_generates_stub(tmp_path): + """new_project with default template generates a minimal stub app.py.""" + project_dir = tmp_path / "my-stub-app" + + with patch("builtins.print"): + new_project(str(project_dir), "default") + + app_py = (project_dir / "app.py").read_text() + assert "CDSHooksService" not in app_py + assert "FHIRGateway" not in app_py + + +def test_new_project_yaml_contains_project_name(tmp_path): + """new_project writes the project name into healthchain.yaml.""" + project_dir = tmp_path / "named-app" + + with patch("builtins.print"): + new_project(str(project_dir), "default") + + yaml_content = (project_dir / "healthchain.yaml").read_text() + assert "named-app" in yaml_content + + +def test_new_project_rejects_existing_directory(tmp_path): + """new_project prints an error and does not overwrite an existing directory.""" + project_dir = tmp_path / "existing-app" + project_dir.mkdir() + + with patch("builtins.print") as mock_print: + new_project(str(project_dir), "default") + + print_output = " ".join(str(call) for call in mock_print.call_args_list) + assert "already exists" in print_output + + +@patch("healthchain.config.appconfig.AppConfig.load", return_value=None) +def test_status_with_no_config_prints_helpful_message(mock_load): + """status prints a helpful message when healthchain.yaml is not found.""" + with patch("builtins.print") as mock_print: + status() + + print_output = " ".join(str(call) for call in mock_print.call_args_list) + assert "healthchain.yaml" in print_output + + +@patch("healthchain.config.appconfig.AppConfig.load") +def test_status_displays_config_fields(mock_load): + """status prints key config values from healthchain.yaml.""" + mock_config = MagicMock() + mock_config.name = "test-app" + mock_config.version = "1.0.0" + mock_config.service.type = "cds-hooks" + mock_config.service.port = 8000 + mock_config.site.environment = "production" + mock_config.site.name = "Test Hospital" + mock_config.security.auth = "none" + mock_config.security.tls.enabled = False + mock_config.security.allowed_origins = ["*"] + mock_config.compliance.hipaa = True + mock_config.compliance.audit_log = "./logs/audit.jsonl" + mock_config.eval.enabled = False + mock_load.return_value = mock_config + + with patch("builtins.print") as mock_print: + status() + + print_output = " ".join(str(call) for call in mock_print.call_args_list) + assert "test-app" in print_output + assert "cds-hooks" in print_output + assert "production" in print_output + assert "Test Hospital" in print_output + assert "enabled" in print_output + + @pytest.mark.parametrize( "args,expected_call", [ ( ["healthchain", "serve", "test.py:app"], - ("serve", ("test.py:app", "0.0.0.0", 8000)), + ("serve", ("test.py:app", "0.0.0.0", None)), ), ( ["healthchain", "eject-templates", "my_configs"],