CAL & Adapters

LOE 3 — Critical Abstraction Layer, protocol adapters, canonical data model, normalization, and sync

The Critical Abstraction Layer (CAL) normalizes all operational data sources into a stable canonical model. Protocol adapters translate source-specific formats (CoT XML, GeoJSON, KML, NOTAM, UDL) into typed canonical entities.

Architecture

External Sources (CoT, KML, NOTAM, UDL …)
         │
         ▼
   [Protocol Adapters]  ←── adapters/
         │ Raw → Canonical translation
         ▼
   [Normalization Pipeline]  ←── cal/normalization/
         │ Provenance tagging, confidence scoring
         ▼
   [Validation]  ←── cal/validation/
         │ Schema enforcement, timeliness checks
         ▼
   [Sync/Cache]  ←── cal/sync/
         │ Local-first cache, delta sync, TTL eviction
         ▼
   [Event Bus]  ←── engine/core/event_bus.py

Adapter Framework

Base Adapter — adapters/base/base_adapter.py

All adapters extend BaseAdapter, which provides:

Required Implementation:

class MyAdapter(BaseAdapter):
    adapter_id = "my-adapter"
    supported_formats = ["my-format"]

    def ingest(self, raw: Any) -> list[Any]:
        # Parse raw data → list of canonical entities
        ...

    def validate(self, raw: Any) -> ValidationResult:
        # Validate raw data structure → ValidationResult(valid, errors)
        ...

    def schema(self) -> dict:
        # Return JSON Schema describing expected input format
        ...

Protocol Adapters

CoT Adapter — adapters/cot/cot_adapter.py

Cursor on Target (CoT) is the primary tactical data format for military situational awareness.

Ingestion Logic:

1. Parse XML input; validate root element is <event>
2. Verify required attributes: version, uid, type, time, start, stale
3. Extract <point> child element with lat, lon, hae (altitude)
4. Map CoT type prefix to track type:

1. Extract optional callsign from <detail>/<contact> element
2. Build canonical Track entity with full metadata and provenance

Validation: Reports specific errors — wrong root element, missing point, missing required attributes, malformed XML.

GeoJSON Adapter — examples/sample-adapter/adapter.py

GeoJSON FeatureCollection adapter serves as both the reference implementation and active adapter.

Ingestion Logic:

1. Parse JSON input (accepts dict, bytes, or str)
2. Validate FeatureCollection type and Feature array structure
3. For each Feature with Point geometry:
   • Extract coordinates [lon, lat]
   • Map track_type property to TrackType enum
   • Extract optional: callsign, altitude_m, heading_deg, speed_mps, timestamp, confidence
   • Calculate completeness score based on optional field presence
4. Build canonical Track entity with TrackPosition, EntityMetadata
5. Skip invalid features defensively — increment error count, continue processing

Expected GeoJSON Properties:

{
  "type": "FeatureCollection",
  "features": [{
    "type": "Feature",
    "geometry": { "type": "Point", "coordinates": [-77.0, 38.9] },
    "properties": {
      "callsign": "VIPER-01",
      "track_type": "friendly",
      "altitude_m": 3048,
      "heading_deg": 90,
      "speed_mps": 200,
      "confidence": 0.95
    }
  }]
}

KML Adapter — adapters/kml/kml_adapter.py

Planned Features:

• KML 2.3 Placemark → Track entity
• NetworkLink streaming support
• Polygon features → Airspace/Threat zone entities

NOTAM Adapter — adapters/notam/notam_adapter.py

Planned Features:

• ICAO NOTAM plain text parsing
• AIXM 5.1 XML parsing
• Affected area geometry extraction

UDL Adapter — adapters/udl/udl_adapter.py

Planned Features:

• Unified Data Library REST/JSON client with authentication
• Entity mapping to canonical types
• Streaming subscription support

CAL Components

Canonical Data Model — cal/models/

The canonical model is defined in sdk/contracts/canonical_models.py and consists of Pydantic v2 models with strict validation. See the API Reference for complete model documentation.

Normalization Pipeline — cal/normalization/

Multi-stage pipeline that transforms adapter output:

1. Schema validation — Pydantic model enforcement
2. Provenance tagging — Source ID, format, ingestion timestamp
3. Confidence scoring — Based on source reliability and data completeness
4. Timeliness tagging — Stale/fresh determination based on TTL

Validation — cal/validation/

Cross-source validation and conflict resolution:

• Schema enforcement against canonical model
• Confidence scoring based on source history
• Conflict resolution when multiple sources report the same entity
• Anomaly detection for out-of-bounds values

Sync/Cache — cal/sync/

Local-first data store with DDIL-optimized sync:

• In-memory cache with configurable TTL
• Delta sync protocol for link recovery
• Eviction policies based on staleness and priority
• Signed manifest verification for offline mission packages