Rhino Health SDK — Workflow Planner & Code Expert

Plan-first skill for the rhino-health Python SDK (v2.1.x). Takes high-level research and analytics goals, decomposes them into phased execution plans, and generates complete runnable Python code.

Context Loading

Before responding, read ALL reference files — planning requires the full SDK picture:

1. API Reference — references/sdk_reference.md Endpoint classes, methods, enums, CreateInput summaries, dataclass fields, import paths.

2. Patterns & Gotchas — references/patterns_and_gotchas.md Auth patterns, resource lookup, metrics execution, filtering, code objects, async, and pitfalls.

3. Metrics Reference — references/metrics_reference.md All 40+ federated metric classes with parameters, import paths, and decision guide.

4. Example Index — references/examples/INDEX.md Mapping of use cases to working example files with key methods and difficulty levels.

For SDK questions that don't require planning, you may selectively load only the relevant files.

Request Routing

Determine what the user needs and follow the appropriate workflow:

Planning Process

Follow these four steps for any multi-step goal:

Step 1: Analyze the Goal

Extract from the user's request:

• Data: What data sources? Do datasets already exist, or need ingestion/creation?
• Analysis: What computation? Metrics, custom code, harmonization, or a combination?
• Output: What does the user want? Numbers, transformed datasets, trained models, exported files?
• Constraints: Filters (age > 50, gender = F), specific sites, time ranges, target data models (OMOP/FHIR)?

If any of these are unclear, ask the user before producing the plan.

Step 2: Select Workflow Templates

Match the goal to one or more composable SDK pipeline templates:

Template A: Federated Analytics

Run statistical metrics across one or more sites without moving data.

Auth → Project → Datasets → Metric Config → Execute → Results

Use when: descriptive stats, survival analysis, hypothesis tests, or any metric-based analysis.

Template B: Code Object Execution

Run custom containerized or Python code across federated sites.

Auth → Project → Data Schema → Code Object Create → Build → Run → Wait → Output Datasets

Use when: custom computation — train/test splits, feature engineering, model training, any logic that metrics alone cannot express.

Template C: Data Harmonization

Transform source data into a target data model (OMOP, FHIR, custom).

Auth → Project → Vocabulary → Semantic Mapping → Syntactic Mapping → Config → Run → Output

Key harmonization types: TransformationType.SPECIFIC_VALUE, SOURCE_DATA_VALUE, ROW_PYTHON, TABLE_PYTHON, SEMANTIC_MAPPING, VLOOKUP, DATE, SECURE_UUID.

Target models: SyntacticMappingDataModel.OMOP, .FHIR, .CUSTOM.

Use when: source data needs transformation before analysis — different column names, value encodings, or target standards like OMOP/FHIR.

Template D: SQL Data Ingestion

Pull data from an on-prem database into the Rhino platform.

Auth → Project → Connection Details → SQL Query → Import as Dataset → Verify

Use when: data lives in a relational database and needs to be brought into the platform as a Dataset.

Template E: Model Training + Inference

Train a federated model, then run inference on new data. This is Template B applied twice:

1. Train phase: Code Object with training logic → produces model artifacts
2. Inference phase: session.code_run.run_inference() using the trained model

Use when: federated ML model training and validation.

Template F: Multi-Pipeline Composition

Chain 2+ templates when a single template cannot satisfy the goal:

Chaining rule: the output datasets of one phase become the input datasets of the next. Use result.output_dataset_uids.root[0].root[0].root[0] to extract UIDs and pass them forward.

Step 3: Compose the Plan

1. Authentication is always Phase 0 — shared across all phases. Include project and workgroup discovery.
2. One template per phase — if the goal requires Templates C → A → B, that is three phases plus Phase 0.
3. Chain outputs to inputs — explicitly state which output from Phase N feeds into Phase N+1.
4. Add checkpoints — after each phase, include a verification step (print status, check dataset count, verify output exists).
5. Surface prerequisites — list what must already exist vs. what will be created.
6. Note alternatives — if there are multiple valid approaches, briefly state why you chose one.

Step 4: Generate Implementation

After presenting the plan, generate the complete runnable code following ALL validation rules in Section 6.

Plan Output Format

Structure every planning response as:

## Goal
[1-2 sentence restatement]

## Prerequisites
- **Must exist:** [project, datasets, schemas, workgroup access]
- **Created by this plan:** [new code objects, schemas, harmonized datasets]

## Plan

### Phase 0: Setup
- Authenticate and discover project/workgroup/datasets
- Checkpoint: print project name and dataset count

### Phase 1: [Name] — Template [X]
- Step 1.1: [description] — `session.X.method()`
- Step 1.2: [description] — `session.Y.method()`
- Checkpoint: [how to verify]

### Phase 2: [Name] — Template [Y]
- Depends on: Phase 1 output datasets
- Step 2.1: ...
- Checkpoint: [how to verify]

## Alternatives Considered
[Other approaches and why this plan is preferred]

## Implementation
[Complete, runnable Python script]

Decision Guidance

When the goal is ambiguous, use this table:

Validation Checklist

Apply every item to ALL generated code — plans and standalone scripts alike.

Endpoint Accessors

Environment

• Default rh.login() connects to production. For dev/QA/staging, pass rhino_api_url: rh.login(..., rhino_api_url=ApiEnvironment.DEV1_AWS_URL)
• Import: from rhino_health.lib.constants import ApiEnvironment
• If user mentions dev1/dev2/QA/staging environment, ALWAYS add rhino_api_url parameter

Import Paths

Metric Calls

• aggregate_dataset_metric takes List[str] of UIDs: [str(d.uid) for d in datasets]
• get_dataset_metric takes a single dataset_uid: str
• joined_dataset_metric takes query_datasets and optional filter_datasets as List[str]
• Metric config objects require data_column (not column or field)
• FilterVariable uses keys: data_column, filter_column, filter_value, filter_type

CreateInput Alias Fields

Nested Structures & RootModels

• CodeObjectRunInput.input_dataset_uids is List[List[str]]: [[uid1, uid2]]
• output_dataset_uids is triply nested RootModel: access via .root[0].root[0].root[0]
• DataSchema.schema_fields is a SchemaFields RootModel: access list via .root, names via .field_names
• group_by format: {"groupings": [{"data_column": "col"}]}
• data_filters list: [FilterVariable(data_column="col", filter_column="col", filter_value="val", filter_type=FilterType.EQUALS)]
• Enum display: use .value for clean strings (e.g. status.value → 'Approved')

Async Operations

• Call wait_for_build() after creating Generalized Compute code objects
• Call wait_for_completion() after run_code_object(), run_data_harmonization(), run_sql_query()

None Checks

Every get_*_by_name() call must be followed by a None check:

dataset = project.get_dataset_by_name("Name")
if dataset is None:
    raise ValueError("Dataset not found")

Code Template

Every generated script must follow this structure:

import rhino_health as rh
from getpass import getpass
# ... additional imports ...

# For non-production environments, add rhino_api_url:
# from rhino_health.lib.constants import ApiEnvironment
# session = rh.login(username="my_email@example.com", password=getpass(),
#                    rhino_api_url=ApiEnvironment.DEV1_AWS_URL)

session = rh.login(username="my_email@example.com", password=getpass())

PROJECT_NAME = "My Project"
# ... constants ...

project = session.project.get_project_by_name(PROJECT_NAME)
if project is None:
    raise ValueError(f"Project '{PROJECT_NAME}' not found")

# ... core logic ...
print(result)

Metric Selection Tree

Map natural language to the right metric class:

All metrics: from rhino_health.lib.metrics import ClassName

Execution modes

Filtering example

from rhino_health.lib.metrics import Mean, FilterType, FilterVariable

config = Mean(
    variable="Height",
    data_filters=[
        FilterVariable(
            data_column="Gender",
            filter_column="Gender",
            filter_value="Female",
            filter_type=FilterType.EQUALS,
        )
    ],
    group_by={"groupings": ["Gender"]},
)

Error-to-Fix Reference

When the user encounters an error, diagnose using this table:

Diagnostic process: identify exception class → locate failing SDK call → cross-reference correct signature in references/sdk_reference.md → check for compound errors.

Question Routing

For non-planning SDK questions, locate the right context section:

Working Examples

Match the user's goal to verified working examples from references/examples/INDEX.md:

Read the relevant example file before generating code to follow its proven patterns.