Aviation Safety - Technical Reference¶

# use_cases/aviation_safety/parser.py
class AviationSafetyParser(BaseParser):
    """Parser for aviation safety markdown incident reports."""

    def parse_all(self) -> Iterator[IncidentRecord]:
        """Parse all markdown files in data directory."""
        for file in self.data_dir.glob("*.md"):
            yield self.parse_file(file)

    def parse_file(self, path: Path) -> IncidentRecord:
        """Parse a single incident report."""
        content = path.read_text()
        sections = self._split_sections(content)

        return IncidentRecord(
            id=self._extract_id(sections),
            date=self._extract_field(sections, "Metadata", "Date"),
            location=self._extract_field(sections, "Metadata", "Location"),
            aircraft=self._parse_aircraft(sections),
            description=sections.get("Incident Description", ""),
            findings=self._parse_findings(sections),
            # ...
        )

@dataclass
class IncidentRecord:
    """Aviation safety incident record."""
    id: str
    date: str = ""
    time: str = ""
    location: str = ""
    country: str = ""
    flight_phase: str = ""
    aircraft: AircraftInfo = field(default_factory=AircraftInfo)
    description: str = ""
    outcome: Outcome = field(default_factory=Outcome)
    weather: WeatherConditions = field(default_factory=WeatherConditions)
    findings: Findings = field(default_factory=Findings)
    safety_recommendations: list[SafetyRecommendation] = field(default_factory=list)

@dataclass
class AircraftInfo:
    """Aircraft information from incident report."""
    aircraft_type: str = ""
    registration: str = ""
    operator: str = ""

@dataclass
class Findings:
    """Incident findings and analysis."""
    primary_cause: str = ""
    contributing_factors: list[str] = field(default_factory=list)
    human_factors: str = ""
    wildlife: dict[str, str] = field(default_factory=dict)

# use_cases/aviation_safety/episode_builder.py
def build_episode_content(record: IncidentRecord) -> str:
    """Build markdown episode from incident record."""
    lines = [
        f"# Aviation Safety Occurrence: {record.id}",
        "",
        "## Occurrence Metadata",
        f"- **Occurrence ID**: {record.id}",
        f"- **Date**: {record.date}",
        f"- **Location**: {record.location}",
        f"- **Country**: {record.country}",
        f"- **Flight Phase**: {record.flight_phase}",
    ]

    if record.aircraft.aircraft_type:
        lines.extend([
            "",
            "## Aircraft",
            f"- **Type**: {record.aircraft.aircraft_type}",
            f"- **Registration**: {record.aircraft.registration}",
            f"- **Operator**: {record.aircraft.operator}",
        ])

    lines.extend([
        "",
        "## Occurrence Description",
        record.description,
    ])

    if record.findings.primary_cause:
        lines.extend([
            "",
            "## Findings",
            f"### Primary Cause",
            record.findings.primary_cause,
        ])

    return "\n".join(lines)

# use_cases/aviation_safety/__init__.py
from .parser import AviationSafetyParser, IncidentRecord
from aletheia.core.ontology import GenericOntologyLoader
from aletheia.core.episodes import register_episode_builder
from .episode_builder import build_episode_content

Parser = AviationSafetyParser
Ontology = GenericOntologyLoader

DATA_DIR = "use_cases/aviation_safety/data"
ONTOLOGY_DIR = "use_cases/aviation_safety/ontology"

register_episode_builder(
    "aviation_safety",
    build_episode_content,
    source_description="Aviation safety data",
)

class ECCAIRSTaxonomyParser:
    """Convert ECCAIRS XML taxonomy to OWL ontology."""

    def parse(self) -> list[ECCAIRSEntity]:
        """Parse ECCAIRS XML file."""
        tree = ET.parse(self.xml_path)
        root = tree.getroot()

        for entity in root.findall(".//ENTITY"):
            yield ECCAIRSEntity(
                id=entity.get("Id"),
                name=entity.get("Description"),
                attributes=self._parse_attributes(entity),
            )

    def to_ttl(self, output_path: Path) -> None:
        """Generate OWL ontology."""
        # Creates owl:Class for each ENTITY
        # Creates owl:DatatypeProperty for attributes
        # Creates owl:ObjectProperty for references

pattern = r'\*\*([^*]+)\*\*:\s*(.+)'
# Captures: ("Date", "2024-02-15")

def build_episode_content(record: IncidentRecord) -> str:
    # Add custom entity extraction
    if record.findings.wildlife:
        lines.append(f"- **Bird Species**: {record.findings.wildlife.get('species')}")