Ports

Pipeline ports implemented by source and destination adapters.

class acmad_uploader.ports.pipeline.Extractor(*args, **kwargs)

Bases: Protocol, Generic

Extract source data without applying API-specific mapping.

extract(source)

Extract source into an infrastructure-independent value.

Return type:: StageResult[TypeVar(ExtractedT)]

class acmad_uploader.ports.pipeline.Transformer(*args, **kwargs)

Bases: Protocol, Generic

Transform extracted data into the canonical ACMAD representation.

transform(extracted)

Validate and transform an extracted value.

Return type:: StageResult[TypeVar(CanonicalT)]

class acmad_uploader.ports.pipeline.Loader(*args, **kwargs)

Bases: Protocol, Generic

Persist canonical data through a destination adapter.

load(canonical)

Persist a canonical value and return a destination result.

Return type:: StageResult[TypeVar(LoadedT)]

Destination port for patient persistence.

class acmad_uploader.ports.patient.PatientLoadAction(*values)

Bases: StrEnum

Change made to a patient at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.patient.PatientUpsertResult(site_patient_id, patient_url, action)

Bases: object

Result returned by a patient destination adapter.

site_patient_id: str

patient_url: str

action: PatientLoadAction

exception acmad_uploader.ports.patient.PatientRepositoryError

Bases: RuntimeError

Expected patient persistence failure suitable for user reporting.

class acmad_uploader.ports.patient.PatientRepository(*args, **kwargs)

Bases: Protocol

Create or update patient records without exposing HTTP concepts.

upsert(patient)

Create a missing patient or update the existing identified patient.

Return type:: PatientUpsertResult

Destination port for diagnosis persistence.

class acmad_uploader.ports.diagnosis.DiagnosisLoadAction(*values)

Bases: StrEnum

Change made to a diagnosis at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.diagnosis.DiagnosisUpsertResult(site_patient_id, diagnosis_type, age_at_diagnosis_days, diagnosis_url, action)

Bases: object

Result returned by a diagnosis destination adapter.

site_patient_id: str

diagnosis_type: DiagnosisType

age_at_diagnosis_days: int

diagnosis_url: str

action: DiagnosisLoadAction

exception acmad_uploader.ports.diagnosis.DiagnosisRepositoryError

Bases: RuntimeError

Expected diagnosis persistence failure suitable for user reporting.

class acmad_uploader.ports.diagnosis.DiagnosisRepository(*args, **kwargs)

Bases: Protocol

Create or update diagnosis records without exposing HTTP concepts.

Diagnoses are matched by (patient, diagnosis_type, age_at_diagnosis_days). Multiple primary diagnoses can exist for one patient at different ages; the destination determines which is “latest” from that history, so this port never sets or reasons about that flag.

upsert(diagnosis)

Create a missing diagnosis or update the existing matched one.

Return type:: DiagnosisUpsertResult

Destination port for gait assessment persistence.

class acmad_uploader.ports.gait_assessment.GaitAssessmentLoadAction(*values)

Bases: StrEnum

Change made to a gait assessment event at the destination.

CREATED = 'created'

FOUND = 'found'

class acmad_uploader.ports.gait_assessment.GaitAssessmentUpsertResult(site_patient_id, age_in_days, gait_assessment_url, action)

Bases: object

Result returned by a gait assessment destination adapter.

site_patient_id: str

age_in_days: int

gait_assessment_url: str

action: GaitAssessmentLoadAction

exception acmad_uploader.ports.gait_assessment.GaitAssessmentRepositoryError

Bases: RuntimeError

Expected gait assessment persistence failure suitable for user reporting.

class acmad_uploader.ports.gait_assessment.GaitAssessmentRepository(*args, **kwargs)

Bases: Protocol

Create or find the patient event and gait assessment for one record.

Unlike patient upsert, a gait assessment has no writable fields beyond its event linkage, so a repeat run performs no write once the event and its assessment already exist; it only resolves and returns them.

upsert(assessment)

Create a missing gait assessment or return the existing one.

Return type:: GaitAssessmentUpsertResult

Destination port for function assessment persistence.

class acmad_uploader.ports.function_assessment.FunctionAssessmentLoadAction(*values)

Bases: StrEnum

Change made to a function assessment at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.function_assessment.FunctionAssessmentUpsertResult(site_patient_id, age_in_days, function_assessment_url, action)

Bases: object

Result returned by a function assessment destination adapter.

site_patient_id: str

age_in_days: int

function_assessment_url: str

action: FunctionAssessmentLoadAction

exception acmad_uploader.ports.function_assessment.FunctionAssessmentRepositoryError

Bases: RuntimeError

Expected function assessment persistence failure suitable for reporting.

class acmad_uploader.ports.function_assessment.FunctionAssessmentRepository(*args, **kwargs)

Bases: Protocol

Create or update a function assessment without exposing HTTP concepts.

A function assessment is matched by its owning gait_assessment (a one-to-one relationship), resolved from (site_patient_id, age_in_days).

upsert(assessment)

Create a missing function assessment or update the existing one.

Return type:: FunctionAssessmentUpsertResult

Destination port for mocap session persistence.

class acmad_uploader.ports.mocap_data.MocapDataLoadAction(*values)

Bases: StrEnum

Change made to a mocap session at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.mocap_data.MocapDataUpsertResult(site_patient_id, age_in_days, mocap_data_url, action)

Bases: object

Result returned by a mocap session destination adapter.

site_patient_id: str

age_in_days: int

mocap_data_url: str

action: MocapDataLoadAction

exception acmad_uploader.ports.mocap_data.MocapDataRepositoryError

Bases: RuntimeError

Expected mocap session persistence failure suitable for reporting.

class acmad_uploader.ports.mocap_data.MocapDataRepository(*args, **kwargs)

Bases: Protocol

Create or update a mocap session without exposing HTTP concepts.

MocapData has no database-level natural key at all (many rows are allowed per gait assessment), so this port matches by (gait_assessment, session_condition, left_condition_detail, right_condition_detail) client-side — confirmed with the user that this combination is unique per day/assessment in practice.

upsert(session)

Create a missing mocap session or update the matched one.

Return type:: MocapDataUpsertResult

Destination port for biomechanics artifact persistence.

class acmad_uploader.ports.biomechanics_artifact.BiomechanicsArtifactLoadAction(*values)

Bases: StrEnum

Change made to a biomechanics artifact at the destination.

CREATED = 'created'

UPDATED = 'updated'

SKIPPED = 'skipped'

class acmad_uploader.ports.biomechanics_artifact.BiomechanicsArtifactUpsertResult(site_patient_id, age_in_days, original_path, artifact_url, action)

Bases: object

Result returned by a biomechanics artifact destination adapter.

site_patient_id: str

age_in_days: int

original_path: str

artifact_url: str

action: BiomechanicsArtifactLoadAction

exception acmad_uploader.ports.biomechanics_artifact.BiomechanicsArtifactRepositoryError

Bases: RuntimeError

Expected biomechanics artifact persistence failure suitable for reporting.

class acmad_uploader.ports.biomechanics_artifact.BiomechanicsArtifactRepository(*args, **kwargs)

Bases: Protocol

Create or update a biomechanics artifact file and its metadata.

Matched by (mocap_data, original_path) — the destination’s own natural key.

upsert(artifact)

Create a missing artifact, update it, or skip an unchanged upload.

Return type:: BiomechanicsArtifactUpsertResult

Destination port for physical examination + anthropometrics persistence.

class acmad_uploader.ports.physical_examination.PhysicalExaminationLoadAction(*values)

Bases: StrEnum

Change made to a physical examination at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.physical_examination.PhysicalExaminationUpsertResult(site_patient_id, age_in_days, physical_examination_url, anthropometrics_url, action, detail_row_count)

Bases: object

Result returned by a physical examination destination adapter.

site_patient_id: str

age_in_days: int

physical_examination_url: str

anthropometrics_url: str

action: PhysicalExaminationLoadAction

detail_row_count: int

exception acmad_uploader.ports.physical_examination.PhysicalExaminationRepositoryError

Bases: RuntimeError

Expected physical examination persistence failure suitable for reporting.

class acmad_uploader.ports.physical_examination.PhysicalExaminationRepository(*args, **kwargs)

Bases: Protocol

Create or update a physical examination and its detail rows.

PhysicalExamination is matched by its owning gait_assessment (one-to-one), resolved from (site_patient_id, age_in_days). Anthropometrics is matched by its owning physical_examination (also one-to-one). The side-keyed detail models (AnthropometricSide/RangeOfMotion/Strength/Tone/ FootStructure) are each matched by (physical_examination, side).

upsert(examination)

Create a missing physical examination/detail rows, or update them.

Return type:: PhysicalExaminationUpsertResult

Destination port for PROMs family persistence.

class acmad_uploader.ports.proms.PromsLoadAction(*values)

Bases: StrEnum

Change made to the PROMs parent row at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.proms.PromsUpsertResult(site_patient_id, age_in_days, proms_url, action, faq_written, ofaq_count, podci_count, mgait_efficiency_count, walk12_count, goal_assessment_count)

Bases: object

Result returned by a PROMs destination adapter.

site_patient_id: str

age_in_days: int

proms_url: str

action: PromsLoadAction

faq_written: bool

ofaq_count: int

podci_count: int

mgait_efficiency_count: int

walk12_count: int

goal_assessment_count: int

exception acmad_uploader.ports.proms.PromsRepositoryError

Bases: RuntimeError

Expected PROMs persistence failure suitable for reporting.

class acmad_uploader.ports.proms.PromsRepository(*args, **kwargs)

Bases: Protocol

Create or update a PROMs parent row and each child questionnaire.

PROMs is matched by its owning gait_assessment (one-to-one), resolved from (site_patient_id, age_in_days). FAQ is matched by its owning proms (also one-to-one). OFAQ/PODCI/ MGaitEfficiency/Walk12/GoalAssessment are each matched by (proms, completed_by).

upsert(proms)

Create missing PROMs rows or update the matched ones.

Return type:: PromsUpsertResult

Destination port for surgery/intervention episode persistence.

class acmad_uploader.ports.surgery.SurgeryLoadAction(*values)

Bases: StrEnum

Change made to a surgery episode at the destination.

CREATED = 'created'

UPDATED = 'updated'

class acmad_uploader.ports.surgery.SurgeryUpsertResult(site_patient_id, age_in_days, surgery_url, action, procedure_count)

Bases: object

Result returned by a surgery destination adapter.

site_patient_id: str

age_in_days: int

surgery_url: str

action: SurgeryLoadAction

procedure_count: int

exception acmad_uploader.ports.surgery.SurgeryRepositoryError

Bases: RuntimeError

Expected surgery persistence failure suitable for user reporting.

class acmad_uploader.ports.surgery.SurgeryRepository(*args, **kwargs)

Bases: Protocol

Create or update a surgery episode and its procedures.

Surgery is matched by its owning patient_event, resolved from (patient, event_type="surgery", age_in_days). Each SurgicalProcedure has no natural key at the destination, so this port matches procedures client-side by (laterality, surgery_type, surgery_region, surgery_intervention).

upsert(surgery)

Create a missing surgery episode or update the matched one.

Return type:: SurgeryUpsertResult