| """Redundancy removal for validated GCMD classification records.""" |
|
|
| from __future__ import annotations |
|
|
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from gcmd_classifier.models import ClassificationRecord, OutputWarning |
| from gcmd_classifier.vocabulary.index import VocabularyIndex |
|
|
| _STAGE = "redundancy_removal" |
|
|
|
|
| class RedundancyResult(BaseModel): |
| """Non-redundant classifications plus structured redundancy diagnostics.""" |
|
|
| model_config = ConfigDict(extra="forbid", frozen=True) |
|
|
| classifications: tuple[ClassificationRecord, ...] = Field(default_factory=tuple) |
| removed: tuple[ClassificationRecord, ...] = Field(default_factory=tuple) |
| warnings: tuple[OutputWarning, ...] = Field(default_factory=tuple) |
|
|
| @property |
| def retained_count(self) -> int: |
| """Number of retained classifications.""" |
| return len(self.classifications) |
|
|
| @property |
| def removed_count(self) -> int: |
| """Number of removed redundant classifications.""" |
| return len(self.removed) |
|
|
|
|
| def remove_redundant_classifications( |
| records: tuple[ClassificationRecord, ...] | list[ClassificationRecord], |
| vocabulary: VocabularyIndex, |
| ) -> RedundancyResult: |
| """Remove exact duplicates and branch-provenance ancestor redundancy.""" |
| retained: list[ClassificationRecord] = [] |
| removed: list[ClassificationRecord] = [] |
| warnings: list[OutputWarning] = [] |
| seen_uuids: dict[str, ClassificationRecord] = {} |
| seen_paths: dict[str, ClassificationRecord] = {} |
|
|
| for record in records: |
| if record.UUID in seen_uuids: |
| removed.append(record) |
| warnings.append( |
| _warning( |
| "DUPLICATE_UUID_REMOVED", |
| "Duplicate UUID classification was removed.", |
| record=record, |
| details={"duplicate_of_branch_id": seen_uuids[record.UUID].branch_id}, |
| ) |
| ) |
| continue |
| if record.canonical_path in seen_paths: |
| removed.append(record) |
| warnings.append( |
| _warning( |
| "DUPLICATE_CANONICAL_PATH_REMOVED", |
| "Duplicate canonical-path classification was removed.", |
| record=record, |
| details={"duplicate_of_branch_id": seen_paths[record.canonical_path].branch_id}, |
| ) |
| ) |
| continue |
| seen_uuids[record.UUID] = record |
| seen_paths[record.canonical_path] = record |
| retained.append(record) |
|
|
| retained, ancestry_removed, ancestry_warnings = _remove_branch_ancestors(retained, vocabulary) |
| removed.extend(ancestry_removed) |
| warnings.extend(ancestry_warnings) |
| return RedundancyResult( |
| classifications=tuple(retained), |
| removed=tuple(removed), |
| warnings=tuple(warnings), |
| ) |
|
|
|
|
| def _remove_branch_ancestors( |
| records: list[ClassificationRecord], |
| vocabulary: VocabularyIndex, |
| ) -> tuple[list[ClassificationRecord], list[ClassificationRecord], list[OutputWarning]]: |
| to_remove: set[int] = set() |
| removed: list[ClassificationRecord] = [] |
| warnings: list[OutputWarning] = [] |
|
|
| for ancestor_index, ancestor in enumerate(records): |
| for descendant_index, descendant in enumerate(records): |
| if ancestor_index == descendant_index: |
| continue |
| if not vocabulary.is_ancestor(ancestor.UUID, descendant.UUID): |
| continue |
| relation = _branch_relation(ancestor.branch_id, descendant.branch_id) |
| if relation == "same_lineage": |
| if ancestor_index not in to_remove: |
| to_remove.add(ancestor_index) |
| removed.append(ancestor) |
| warnings.append( |
| _warning( |
| "ANCESTOR_REMOVED_SAME_BRANCH", |
| ( |
| "Ancestor classification was removed because a deeper descendant " |
| "exists in the same branch lineage." |
| ), |
| record=ancestor, |
| details={ |
| "descendant_uuid": descendant.UUID, |
| "descendant_branch_id": descendant.branch_id, |
| }, |
| ) |
| ) |
| elif relation == "independent": |
| warnings.append( |
| _warning( |
| "ANCESTOR_PRESERVED_INDEPENDENT_BRANCH", |
| ( |
| "Ancestor classification was preserved because branch provenance " |
| "indicates an independent classification decision." |
| ), |
| record=ancestor, |
| details={ |
| "descendant_uuid": descendant.UUID, |
| "descendant_branch_id": descendant.branch_id, |
| }, |
| ) |
| ) |
| else: |
| warnings.append( |
| _warning( |
| "ANCESTOR_PRESERVED_INSUFFICIENT_PROVENANCE", |
| ( |
| "Ancestor classification was preserved because branch provenance " |
| "was insufficient to prove redundancy." |
| ), |
| record=ancestor, |
| details={ |
| "descendant_uuid": descendant.UUID, |
| "descendant_branch_id": descendant.branch_id, |
| }, |
| ) |
| ) |
| retained = [record for index, record in enumerate(records) if index not in to_remove] |
| return retained, removed, warnings |
|
|
|
|
| def _branch_relation(ancestor_branch_id: str | None, descendant_branch_id: str | None) -> str: |
| if not ancestor_branch_id or not descendant_branch_id: |
| return "insufficient" |
| if ancestor_branch_id == descendant_branch_id: |
| return "same_lineage" |
| if descendant_branch_id.startswith(f"{ancestor_branch_id}/"): |
| return "same_lineage" |
| if ancestor_branch_id.startswith(f"{descendant_branch_id}/"): |
| return "same_lineage" |
| return "independent" |
|
|
|
|
| def _warning( |
| code: str, |
| message: str, |
| *, |
| record: ClassificationRecord, |
| details: dict | None = None, |
| ) -> OutputWarning: |
| warning_details = { |
| "UUID": record.UUID, |
| "canonical_path": record.canonical_path, |
| "branch_id": record.branch_id, |
| } |
| if details: |
| warning_details.update(details) |
| return OutputWarning( |
| code=code, |
| message=message, |
| stage=_STAGE, |
| details=warning_details, |
| ) |
|
|
