Skip to content

skill

Classes:

ExtractionSkill 

ExtractionSkill()

Bases: BaseSkill

Looks for file sources in ctx.extras['extraction_sources'] and stores extracted data in ctx.results['intake']. attributes: _report_extractor: description: Value for _report_extractor. _wearable_extractor: description: Value for _wearable_extractor.

Methods:

Source code in src/hiperhealth/skills/extraction/skill.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self) -> None:
    """
    title: Initialize extractors and built-in skill metadata.
    """
    super().__init__(
        SkillMetadata(
            name='hiperhealth.extraction',
            version='0.4.0',
            stages=(Stage.INTAKE,),
            description=(
                'Extract text from medical reports and '
                'wearable data files.'
            ),
        )
    )
    self._report_extractor = MedicalReportFileExtractor()
    self._wearable_extractor = WearableDataFileExtractor()

check_requirements 

check_requirements(
    stage: str, ctx: PipelineContext
) -> list[Inquiry]

Override to return a list of Inquiry objects describing what additional data the skill needs. The default implementation returns an empty list (no extra data needed). Inquiries use three priority levels: - required: must have before this stage can run - supplementary: improves results, available now - deferred: only available after a future pipeline step parameters: stage: type: str ctx: type: PipelineContext returns: type: list[Inquiry]

Source code in src/hiperhealth/pipeline/skill.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def check_requirements(
    self, stage: str, ctx: PipelineContext
) -> list[Inquiry]:
    """
    title: Determine what information is needed before execution.
    summary: |-
      Override to return a list of Inquiry objects describing
      what additional data the skill needs.  The default
      implementation returns an empty list (no extra data needed).
      Inquiries use three priority levels:
      - required: must have before this stage can run
      - supplementary: improves results, available now
      - deferred: only available after a future pipeline step
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: list[Inquiry]
    """
    return []

execute 

execute(
    stage: str, ctx: PipelineContext
) -> PipelineContext
Source code in src/hiperhealth/skills/extraction/skill.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def execute(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Extract data from sources listed in ctx.extras.
    parameters:
      stage:
        type: str
        description: Value for stage.
      ctx:
        type: PipelineContext
        description: Value for ctx.
    returns:
      type: PipelineContext
      description: Return value.
    """
    if stage != Stage.INTAKE:
        return ctx

    sources = ctx.extras.get('extraction_sources', {})
    results: dict[str, Any] = ctx.results.get(Stage.INTAKE, {})

    report_files = sources.get('medical_reports', [])
    for source in report_files:
        report = self._report_extractor.extract_report_data(source)
        results.setdefault('medical_reports', []).append(report)

    wearable_files = sources.get('wearable_data', [])
    for source in wearable_files:
        data = self._wearable_extractor.extract_wearable_data(source)
        results.setdefault('wearable_data', []).append(data)

    ctx.results[Stage.INTAKE] = results
    return ctx

post 

post(stage: str, ctx: PipelineContext) -> PipelineContext
Source code in src/hiperhealth/pipeline/skill.py 
172
173
174
175
176
177
178
179
180
181
182
183
def post(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Called after the stage's main execution.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: PipelineContext
    """
    return ctx

pre 

pre(stage: str, ctx: PipelineContext) -> PipelineContext
Source code in src/hiperhealth/pipeline/skill.py 
146
147
148
149
150
151
152
153
154
155
156
157
def pre(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Called before the stage's main execution.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: PipelineContext
    """
    return ctx