Code / scripts/ml_training/essay_scoring/features/pipeline_helpers.py
scripts/ml_training/essay_scoring/features/pipeline_helpers.py
1
"""Feature-pipeline request preparation helpers.
2
3
Purpose:
4
Keep feature-pipeline orchestration focused on extraction flow while this
5
module owns small request-shaping helpers for candidate feature selections
6
and ledger-derived text references.
7
8
Relationships:
9
- Consumed by `features.pipeline` before local or Hemma extraction.
10
- Uses the shared writing-issue text-reference contract to derive
11
candidate sidecar text streams without changing accepted feature rows.
12
"""
13
14
from __future__ import annotations
15
16
from huleedu_nlp_shared.writing_issues import (
17
FeatureConsumerTextRefV1,
18
)
19
20
from ..features.candidates.candidate_sidecar_text_binding import (
21
lexical_candidate_text_for_ref,
22
)
23
from ..features.parser_text_authority_projection import (
24
ResearchParserTextArtifact,
25
)
26
from ..features.schema import AppliedSpellingFeatureCounts
27
from ..features.writing_issue_ledger import (
28
ResearchLedgerArtifact,
29
)
30
from ..offload.applied_error_rate_contract import (
31
requires_applied_spelling_counts,
32
)
33
from ..offload.extract_feature_selection import (
34
CandidateFeatureSetId,
35
ExtractAcceptedScorerInputV3,
36
ExtractCandidateFeatureSelectionV1,
37
ExtractFeatureSelectionV2,
38
)
39
40
41
def required_consumer_ref(
42
ref: FeatureConsumerTextRefV1 | None,
43
field_name: str,
44
) -> FeatureConsumerTextRefV1:
45
"""Return a required ledger text reference or fail closed."""
46
47
if ref is None:
48
raise RuntimeError(f"Ledger policy manifest missing {field_name}.")
49
return ref
50
51
52
def lexical_candidate_texts_for_refs(
53
ledger_artifacts: list[ResearchLedgerArtifact],
54
lexical_input_refs: list[FeatureConsumerTextRefV1],
55
) -> list[str]:
56
"""Resolve lexical candidate text streams from row-level ledger artifacts."""
57
58
if len(ledger_artifacts) != len(lexical_input_refs):
59
raise RuntimeError("Ledger artifacts and lexical refs have different row counts.")
60
return [
61
lexical_candidate_text_for_ref(
62
parser_stabilized_text=artifact.parser_stabilized_text,
63
lexical_input_ref=ref,
64
)
65
for artifact, ref in zip(ledger_artifacts, lexical_input_refs, strict=True)
66
]
67
68
69
def feature_selection_for_pipeline(
70
accepted: ExtractAcceptedScorerInputV3,
71
candidate_feature_set_ids: tuple[CandidateFeatureSetId, ...] = (),
72
) -> ExtractFeatureSelectionV2:
73
"""Attach candidate intent to one already validated accepted scorer selection."""
74
75
return ExtractFeatureSelectionV2(
76
schema_version="extract_feature_selection.v2",
77
accepted=accepted,
78
candidates=ExtractCandidateFeatureSelectionV1(
79
candidate_feature_set_ids=candidate_feature_set_ids
80
),
81
)
82
83
84
def applied_counts_for_selected_families(
85
*,
86
prepared_parser_text_artifacts: list[ResearchParserTextArtifact] | None,
87
selected_family_ids: tuple[str, ...],
88
) -> list[AppliedSpellingFeatureCounts] | None:
89
"""Return applied correction counts only for families whose values require them."""
90
91
if prepared_parser_text_artifacts is None or not requires_applied_spelling_counts(
92
selected_family_ids
93
):
94
return None
95
return [artifact.applied_spelling_counts for artifact in prepared_parser_text_artifacts]