Spaces:
Paused
Paused
| """ | |
| EAF Exporter | |
| Exports tiered annotations to ELAN Annotation Format (EAF) XML files. | |
| EAF is the native format for ELAN (https://archive.mpi.nl/tla/elan), | |
| a tool widely used for linguistic annotation of audio/video data. | |
| The EAF format supports: | |
| - Time-aligned annotations with millisecond precision | |
| - Hierarchical tier structures with parent-child relationships | |
| - Multiple linguistic types with different constraints | |
| - Media references (audio/video files) | |
| """ | |
| import logging | |
| import os | |
| import xml.etree.ElementTree as ET | |
| from datetime import datetime | |
| from typing import Dict, List, Any, Optional, Tuple, Set | |
| from .base import BaseExporter, ExportContext, ExportResult | |
| logger = logging.getLogger(__name__) | |
| class EAFExporter(BaseExporter): | |
| """ | |
| Exports tiered annotations to ELAN Annotation Format (EAF). | |
| This exporter creates valid EAF 3.0 XML files that can be opened | |
| directly in ELAN for review or further annotation. | |
| """ | |
| format_name = "eaf" | |
| description = "ELAN Annotation Format (EAF) for linguistic annotation" | |
| file_extensions = [".eaf"] | |
| def can_export(self, context: ExportContext) -> Tuple[bool, str]: | |
| """ | |
| Check if the context contains tiered_annotation schema. | |
| Args: | |
| context: ExportContext to validate | |
| Returns: | |
| Tuple of (can_export, reason) | |
| """ | |
| for schema in context.schemas: | |
| if schema.get("annotation_type") == "tiered_annotation": | |
| return True, "" | |
| return False, "No tiered_annotation schema found in configuration" | |
| def export( | |
| self, | |
| context: ExportContext, | |
| output_path: str, | |
| options: Optional[dict] = None | |
| ) -> ExportResult: | |
| """ | |
| Export annotations to EAF format. | |
| Args: | |
| context: ExportContext with annotation data | |
| output_path: Directory path for output files | |
| options: Optional settings: | |
| - author: Author name for EAF header | |
| - include_empty_tiers: Whether to include tiers with no annotations | |
| Returns: | |
| ExportResult with status and file paths | |
| """ | |
| options = options or {} | |
| files_written = [] | |
| warnings = [] | |
| stats = {"instances": 0, "annotations": 0, "tiers": 0} | |
| # Create output directory | |
| os.makedirs(output_path, exist_ok=True) | |
| # Find tiered_annotation schemas | |
| tiered_schemas = [ | |
| s for s in context.schemas | |
| if s.get("annotation_type") == "tiered_annotation" | |
| ] | |
| for instance_id, item in context.items.items(): | |
| # Get annotations for this instance | |
| instance_annotations = [ | |
| a for a in context.annotations | |
| if a.get("instance_id") == instance_id | |
| ] | |
| for schema in tiered_schemas: | |
| schema_name = schema.get("name", "tiered") | |
| # Get tiered annotation data for this schema | |
| tiered_data = None | |
| for ann in instance_annotations: | |
| if schema_name in ann.get("labels", {}): | |
| try: | |
| import json | |
| raw_value = ann["labels"][schema_name] | |
| if isinstance(raw_value, str): | |
| tiered_data = json.loads(raw_value) | |
| elif isinstance(raw_value, dict): | |
| tiered_data = raw_value | |
| except (json.JSONDecodeError, TypeError): | |
| pass | |
| break | |
| if not tiered_data: | |
| continue | |
| # Get media URL | |
| source_field = schema.get("source_field", "audio_url") | |
| media_url = item.get(source_field, "") | |
| # Generate EAF XML | |
| root = self._create_eaf_document( | |
| schema, | |
| tiered_data, | |
| media_url, | |
| options | |
| ) | |
| # Write to file | |
| safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in instance_id) | |
| filename = f"{safe_id}_{schema_name}.eaf" | |
| filepath = os.path.join(output_path, filename) | |
| tree = ET.ElementTree(root) | |
| tree.write(filepath, encoding="utf-8", xml_declaration=True) | |
| files_written.append(filepath) | |
| stats["instances"] += 1 | |
| # Count annotations | |
| annotations = tiered_data.get("annotations", {}) | |
| for tier_anns in annotations.values(): | |
| stats["annotations"] += len(tier_anns) | |
| stats["tiers"] = len(schema.get("tiers", [])) | |
| return ExportResult( | |
| success=True, | |
| format_name=self.format_name, | |
| files_written=files_written, | |
| warnings=warnings, | |
| stats=stats | |
| ) | |
| def _create_eaf_document( | |
| self, | |
| schema: dict, | |
| tiered_data: dict, | |
| media_url: str, | |
| options: dict | |
| ) -> ET.Element: | |
| """ | |
| Create the EAF XML document structure. | |
| Args: | |
| schema: The tiered_annotation schema configuration | |
| tiered_data: The annotation data | |
| media_url: URL/path to the media file | |
| options: Export options | |
| Returns: | |
| ET.Element root of the EAF document | |
| """ | |
| # Root element | |
| root = ET.Element("ANNOTATION_DOCUMENT") | |
| root.set("xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance") | |
| root.set("xsi:noNamespaceSchemaLocation", | |
| "http://www.mpi.nl/tools/elan/EAFv3.0.xsd") | |
| root.set("DATE", datetime.now().isoformat()) | |
| root.set("FORMAT", "3.0") | |
| root.set("VERSION", "3.0") | |
| root.set("AUTHOR", options.get("author", "Potato Annotation Tool")) | |
| # Header | |
| header = ET.SubElement(root, "HEADER") | |
| header.set("MEDIA_FILE", "") | |
| header.set("TIME_UNITS", "milliseconds") | |
| if media_url: | |
| media_descriptor = ET.SubElement(header, "MEDIA_DESCRIPTOR") | |
| media_descriptor.set("MEDIA_URL", media_url) | |
| media_descriptor.set("MIME_TYPE", self._get_mime_type(media_url)) | |
| media_descriptor.set("RELATIVE_MEDIA_URL", "") | |
| # Property elements | |
| props = [ | |
| ("lastUsedAnnotationId", "0"), | |
| ] | |
| for name, value in props: | |
| prop = ET.SubElement(header, "PROPERTY") | |
| prop.set("NAME", name) | |
| prop.text = value | |
| # TIME_ORDER - collect all unique time slots | |
| time_order = ET.SubElement(root, "TIME_ORDER") | |
| time_slots = tiered_data.get("time_slots", {}) | |
| if not time_slots: | |
| # Generate from annotations | |
| time_slots = self._generate_time_slots(tiered_data.get("annotations", {})) | |
| # Sort and add time slots | |
| slot_items = sorted(time_slots.items(), key=lambda x: x[1]) | |
| for slot_id, time_ms in slot_items: | |
| ts = ET.SubElement(time_order, "TIME_SLOT") | |
| ts.set("TIME_SLOT_ID", slot_id) | |
| ts.set("TIME_VALUE", str(int(time_ms))) | |
| # Create reverse mapping for looking up slot IDs | |
| time_to_slot = {v: k for k, v in time_slots.items()} | |
| initial_slot_count = len(time_to_slot) | |
| # TIER elements | |
| tiers = schema.get("tiers", []) | |
| annotations = tiered_data.get("annotations", {}) | |
| annotation_id_counter = [0] # Use list to allow mutation in nested function | |
| for tier_def in tiers: | |
| tier_el = self._create_tier_element( | |
| root, | |
| tier_def, | |
| annotations.get(tier_def["name"], []), | |
| time_to_slot, | |
| annotation_id_counter | |
| ) | |
| # Add any dynamically created time slots to TIME_ORDER | |
| if len(time_to_slot) > initial_slot_count: | |
| slot_to_time = {v: k for k, v in time_to_slot.items()} | |
| new_slots = sorted( | |
| [(sid, slot_to_time[sid]) for sid in slot_to_time if sid not in time_slots], | |
| key=lambda x: x[1] | |
| ) | |
| for slot_id, time_ms in new_slots: | |
| ts = ET.SubElement(time_order, "TIME_SLOT") | |
| ts.set("TIME_SLOT_ID", slot_id) | |
| ts.set("TIME_VALUE", str(int(time_ms))) | |
| # LINGUISTIC_TYPE elements | |
| self._create_linguistic_types(root, tiers) | |
| # CONSTRAINT elements (for dependent tiers) | |
| self._create_constraints(root) | |
| return root | |
| def _create_tier_element( | |
| self, | |
| root: ET.Element, | |
| tier_def: dict, | |
| tier_annotations: List[dict], | |
| time_to_slot: dict, | |
| annotation_id_counter: List[int] | |
| ) -> ET.Element: | |
| """ | |
| Create a TIER element with its annotations. | |
| Args: | |
| root: The root EAF element | |
| tier_def: Tier definition from schema | |
| tier_annotations: List of annotations for this tier | |
| time_to_slot: Mapping of time values to slot IDs | |
| annotation_id_counter: Counter for generating annotation IDs | |
| Returns: | |
| The created TIER element | |
| """ | |
| tier_el = ET.SubElement(root, "TIER") | |
| tier_el.set("TIER_ID", tier_def["name"]) | |
| # Determine linguistic type | |
| if tier_def.get("tier_type") == "dependent": | |
| constraint_type = tier_def.get("constraint_type", "included_in") | |
| ling_type = f"default-lt-{constraint_type}" | |
| tier_el.set("PARENT_REF", tier_def.get("parent_tier", "")) | |
| else: | |
| ling_type = "default-lt" | |
| tier_el.set("LINGUISTIC_TYPE_REF", tier_def.get("linguistic_type", ling_type)) | |
| tier_el.set("DEFAULT_LOCALE", "en") | |
| # Add annotations | |
| for ann in sorted(tier_annotations, key=lambda a: a.get("start_time", 0)): | |
| annotation_id_counter[0] += 1 | |
| ann_id = f"a{annotation_id_counter[0]}" | |
| annotation_el = ET.SubElement(tier_el, "ANNOTATION") | |
| if tier_def.get("tier_type") == "independent": | |
| # ALIGNABLE_ANNOTATION for independent tiers | |
| alignable = ET.SubElement(annotation_el, "ALIGNABLE_ANNOTATION") | |
| alignable.set("ANNOTATION_ID", ann_id) | |
| start_time = int(ann.get("start_time", 0)) | |
| end_time = int(ann.get("end_time", 0)) | |
| start_slot = self._get_or_create_slot(time_to_slot, start_time) | |
| end_slot = self._get_or_create_slot(time_to_slot, end_time) | |
| alignable.set("TIME_SLOT_REF1", start_slot) | |
| alignable.set("TIME_SLOT_REF2", end_slot) | |
| value_el = ET.SubElement(alignable, "ANNOTATION_VALUE") | |
| value_el.text = ann.get("value") or ann.get("label", "") | |
| else: | |
| # REF_ANNOTATION for dependent tiers | |
| ref_ann = ET.SubElement(annotation_el, "REF_ANNOTATION") | |
| ref_ann.set("ANNOTATION_ID", ann_id) | |
| # Reference parent annotation | |
| parent_id = ann.get("parent_id", "") | |
| if parent_id: | |
| # Convert internal ID to EAF annotation ID | |
| # For simplicity, use annotation reference | |
| ref_ann.set("ANNOTATION_REF", parent_id) | |
| value_el = ET.SubElement(ref_ann, "ANNOTATION_VALUE") | |
| value_el.text = ann.get("value") or ann.get("label", "") | |
| return tier_el | |
| def _get_or_create_slot( | |
| self, | |
| time_to_slot: dict, | |
| time_ms: int | |
| ) -> str: | |
| """Get existing slot ID or create a new mapping.""" | |
| if time_ms in time_to_slot: | |
| return time_to_slot[time_ms] | |
| # Create new slot | |
| slot_id = f"ts{len(time_to_slot) + 1}" | |
| time_to_slot[time_ms] = slot_id | |
| return slot_id | |
| def _generate_time_slots( | |
| self, | |
| annotations: Dict[str, List[dict]] | |
| ) -> Dict[str, int]: | |
| """Generate time slots from annotations.""" | |
| times: Set[int] = set() | |
| for tier_anns in annotations.values(): | |
| for ann in tier_anns: | |
| if ann.get("start_time") is not None: | |
| times.add(int(ann["start_time"])) | |
| if ann.get("end_time") is not None: | |
| times.add(int(ann["end_time"])) | |
| return { | |
| f"ts{i+1}": time | |
| for i, time in enumerate(sorted(times)) | |
| } | |
| def _create_linguistic_types( | |
| self, | |
| root: ET.Element, | |
| tiers: List[dict] | |
| ) -> None: | |
| """Create LINGUISTIC_TYPE elements for all tier types.""" | |
| # Default linguistic type for independent tiers | |
| lt = ET.SubElement(root, "LINGUISTIC_TYPE") | |
| lt.set("LINGUISTIC_TYPE_ID", "default-lt") | |
| lt.set("TIME_ALIGNABLE", "true") | |
| lt.set("GRAPHIC_REFERENCES", "false") | |
| # Linguistic types for each constraint type | |
| constraint_types = set() | |
| for tier in tiers: | |
| if tier.get("tier_type") == "dependent": | |
| constraint_types.add(tier.get("constraint_type", "included_in")) | |
| for constraint in constraint_types: | |
| lt = ET.SubElement(root, "LINGUISTIC_TYPE") | |
| lt.set("LINGUISTIC_TYPE_ID", f"default-lt-{constraint}") | |
| lt.set("CONSTRAINTS", self._constraint_to_elan(constraint)) | |
| lt.set("TIME_ALIGNABLE", "true" if constraint in ("time_subdivision", "included_in") else "false") | |
| lt.set("GRAPHIC_REFERENCES", "false") | |
| def _constraint_to_elan(self, constraint_type: str) -> str: | |
| """Map constraint type to ELAN constraint stereotype.""" | |
| mapping = { | |
| "time_subdivision": "Time_Subdivision", | |
| "included_in": "Included_In", | |
| "symbolic_association": "Symbolic_Association", | |
| "symbolic_subdivision": "Symbolic_Subdivision", | |
| } | |
| return mapping.get(constraint_type, "Included_In") | |
| def _create_constraints(self, root: ET.Element) -> None: | |
| """Create CONSTRAINT elements for the standard ELAN constraint types.""" | |
| constraints = [ | |
| ("Time_Subdivision", "Time subdivision of parent annotation's time interval, no time gaps allowed within this interval"), | |
| ("Symbolic_Subdivision", "Symbolic subdivision of a parent annotation. Annotations refer to the same time interval as the parent"), | |
| ("Symbolic_Association", "1-1 association with a parent annotation"), | |
| ("Included_In", "Time alignable annotations within the parent annotation's time interval, gaps are allowed"), | |
| ] | |
| for constraint_id, description in constraints: | |
| constraint = ET.SubElement(root, "CONSTRAINT") | |
| constraint.set("DESCRIPTION", description) | |
| constraint.set("STEREOTYPE", constraint_id) | |
| def _get_mime_type(self, url: str) -> str: | |
| """Determine MIME type from file extension.""" | |
| url_lower = url.lower() | |
| if url_lower.endswith((".mp4", ".m4v")): | |
| return "video/mp4" | |
| elif url_lower.endswith((".webm",)): | |
| return "video/webm" | |
| elif url_lower.endswith((".avi",)): | |
| return "video/avi" | |
| elif url_lower.endswith((".mov",)): | |
| return "video/quicktime" | |
| elif url_lower.endswith((".wav",)): | |
| return "audio/wav" | |
| elif url_lower.endswith((".mp3",)): | |
| return "audio/mpeg" | |
| elif url_lower.endswith((".ogg", ".oga")): | |
| return "audio/ogg" | |
| elif url_lower.endswith((".flac",)): | |
| return "audio/flac" | |
| else: | |
| return "audio/x-wav" # Default | |