session

Every interaction between System X and hiperhealth is recorded as an event row. Current state (clinical data, results, pending inquiries) is derived by replaying events. System X owns the file lifecycle (storage, deletion, retention).

Classes:

• Inquiry –
• Session –

Inquiry

Bases: BaseModel

Session

Session(path: Path, language: str = 'en')

System X creates or loads a session, provides clinical data, and uses the runner to assess / execute stages. The parquet file is the single source of truth. attributes: path: type: Path _language: type: str _events: type: list[dict[str, Any]]

Methods:

• create –
• load –
• provide_answers –
• record_event –
• set_clinical_data –
• to_context –
• update_from_context –

Attributes:

• clinical_data (dict[str, Any]) –
• events (list[dict[str, Any]]) –
• language (str) –
• pending_inquiries (list[Inquiry]) –
• results (dict[str, Any]) –
• stages_completed (list[str]) –

Source code in src/hiperhealth/pipeline/session.py

def __init__(self, path: Path, language: str = 'en') -> None:
    """
    title: Initialize an in-memory session wrapper for a parquet file.
    parameters:
      path:
        type: Path
      language:
        type: str
    """
    self.path: Path = path
    self._language: str = language
    self._events: list[dict[str, Any]] = []

clinical_data property

clinical_data: dict[str, Any]

events property

events: list[dict[str, Any]]

language property

language: str

pending_inquiries property

pending_inquiries: list[Inquiry]

results property

results: dict[str, Any]

stages_completed property

stages_completed: list[str]

create classmethod

create(path: str | Path, language: str = 'en') -> Session

Source code in src/hiperhealth/pipeline/session.py

@classmethod
def create(
    cls,
    path: str | Path,
    language: str = 'en',
) -> Session:
    """
    title: Create a new session file.
    parameters:
      path:
        type: str | Path
      language:
        type: str
    returns:
      type: Session
    """
    path = Path(path)
    if path.exists():
        msg = f'Session file already exists: {path}'
        raise FileExistsError(msg)
    session = cls(path, language=language)
    session._save()
    return session

load classmethod

load(path: str | Path) -> Session

Source code in src/hiperhealth/pipeline/session.py

@classmethod
def load(cls, path: str | Path) -> Session:
    """
    title: Load an existing session from a parquet file.
    parameters:
      path:
        type: str | Path
    returns:
      type: Session
    """
    path = Path(path)
    if not path.exists():
        msg = f'Session file not found: {path}'
        raise FileNotFoundError(msg)
    session = cls(path)
    session._load()
    return session

provide_answers

provide_answers(answers: dict[str, Any]) -> None

Source code in src/hiperhealth/pipeline/session.py

def provide_answers(self, answers: dict[str, Any]) -> None:
    """
    title: Provide answers to inquiries.
    parameters:
      answers:
        type: dict[str, Any]
    """
    self._append_event(
        'answers_provided',
        data={'fields': answers},
    )

record_event

record_event(
    event_type: str,
    stage: str | None = None,
    skill_name: str | None = None,
    data: dict[str, Any] | None = None,
) -> None

Source code in src/hiperhealth/pipeline/session.py

def record_event(
    self,
    event_type: str,
    stage: str | None = None,
    skill_name: str | None = None,
    data: dict[str, Any] | None = None,
) -> None:
    """
    title: Record an arbitrary event (used by the runner).
    parameters:
      event_type:
        type: str
      stage:
        type: str | None
      skill_name:
        type: str | None
      data:
        type: dict[str, Any] | None
    """
    self._append_event(
        event_type,
        stage=stage,
        skill_name=skill_name,
        data=data,
    )

set_clinical_data

set_clinical_data(fields: dict[str, Any]) -> None

Source code in src/hiperhealth/pipeline/session.py

def set_clinical_data(self, fields: dict[str, Any]) -> None:
    """
    title: Provide clinical information (no PII).
    parameters:
      fields:
        type: dict[str, Any]
    """
    self._append_event(
        'clinical_data_set',
        data={'fields': fields},
    )

to_context

to_context() -> PipelineContext

Source code in src/hiperhealth/pipeline/session.py

def to_context(self) -> PipelineContext:
    """
    title: Build a PipelineContext from current session state.
    returns:
      type: PipelineContext
    """
    return PipelineContext(
        patient=self.clinical_data,
        language=self._language,
        session_id=self.path.stem,
        results=self.results,
        extras={},
    )

update_from_context

update_from_context(
    stage: str, ctx: PipelineContext
) -> None

Source code in src/hiperhealth/pipeline/session.py

def update_from_context(
    self,
    stage: str,
    ctx: PipelineContext,
) -> None:
    """
    title: Capture results after a stage runs.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    """
    stage_result = ctx.results.get(stage)
    result_data: Any
    if stage_result is not None:
        if hasattr(stage_result, 'model_dump'):
            result_data = stage_result.model_dump()
        else:
            result_data = stage_result
    else:
        result_data = {}
    self._append_event(
        'stage_completed',
        stage=stage,
        data={'results': result_data},
    )