DIU OMEN Solicitation License

Examples & Use Cases

Reference implementations: sample GeoJSON adapter, Blue-Force Track monitor plugin, and mission package usage


OMEN ships with plug-and-play reference implementations in examples/ that demonstrate the adapter contract, plugin lifecycle, and mission data handling.

Sample Adapter — GeoJSON to Canonical Tracks

Location: examples/sample-adapter/

The SampleGeoJSONAdapter is a fully functional adapter that translates GeoJSON FeatureCollections into canonical Track entities. It serves as both a reference implementation and the active GeoJSON adapter.

Files

File Purpose Lines
adapter.py Full adapter implementation with Pydantic models 264
run_adapter.py Entry point — loads sample data, validates, ingests, prints results 41
test_adapter.py Pytest suite — validates correct and invalid input handling 40+
sample_input.geojson Sample GeoJSON FeatureCollection with Point features
requirements.txt Dependencies (pydantic)

How It Works

1. Input Validationvalidate(raw) checks:

  • Top-level type is "FeatureCollection"
  • Features array exists and is non-empty
  • Each feature has "type": "Feature" and a geometry object
  • Geometry type is "Point" with valid coordinates array

2. Ingestioningest(raw) translates:

  • GeoJSON coordinates [lon, lat]TrackPosition(latitude, longitude)
  • Feature properties → Track with EntityMetadata (source_id, confidence, completeness)
  • Completeness score is calculated based on optional field presence (callsign, altitude, heading, speed)

3. Error Handling:

  • Invalid features are skipped defensively — errors are counted, processing continues
  • Health metrics track total ingestions and error counts
  • Pydantic validation catches out-of-bounds coordinates (lat ±90, lon ±180)

Usage

cd examples/sample-adapter
pip install -r requirements.txt
python run_adapter.py

Expected Output:

=== OMEN Sample Adapter Demo ===
Adapter: sample-geojson
Formats: ['geojson']

--- Validation ---
Valid: True

--- Ingestion ---
Ingested 2 entities:
  Track: VIPER-01 (friendly) @ 38.9, -77.0
  Track: UNKNOWN-42 (unknown) @ 39.1, -76.8

--- Health ---
Healthy: True | Ingestions: 2 | Errors: 0

Sample Input

{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "geometry": {
        "type": "Point",
        "coordinates": [-77.0364, 38.8977]
      },
      "properties": {
        "callsign": "VIPER-01",
        "track_type": "friendly",
        "altitude_m": 3048.0,
        "heading_deg": 90.0,
        "speed_mps": 200.0,
        "confidence": 0.95
      }
    }
  ]
}

Running Tests

cd examples/sample-adapter
pip install -r requirements.txt
pytest test_adapter.py -v

Tests cover:

  • Valid GeoJSON ingestion produces correct Track entities
  • Invalid GeoJSON (wrong type) is rejected by validation
  • Adapter health metrics update correctly
  • Error counting works for malformed features

Sample Plugin — Blue-Force Track Monitor

Location: examples/sample-plugin/

The BFTMonitorPlugin demonstrates the plugin lifecycle by monitoring blue-force tracks and emitting proximity alerts.

Files

File Purpose Lines
plugin.py BFT Monitor plugin with proximity alerting 128
run_plugin.py Entry point — registers with engine, sends test events 62
test_plugin.py Pytest suite — lifecycle and event handling tests 40+
plugin.json Plugin manifest (capabilities, dependencies, schema)
requirements.txt Dependencies

How It Works

Plugin ID: sample-bft-monitor

1. Configuration:

  • alert_radius_nm — proximity alert threshold in nautical miles (default: 5.0)

2. Event Handling: The plugin subscribes to track_update events and:

  • Maintains an internal registry of known blue-force tracks
  • Computes great-circle distance between tracks
  • Emits proximity_alert events when tracks are closer than alert_radius_nm

3. Proximity Calculation: Uses the Haversine formula for great-circle distance between geographic positions.

Usage

cd examples/sample-plugin
pip install -r requirements.txt
python run_plugin.py

Expected Output:

=== OMEN Sample Plugin Demo ===
Plugin: sample-bft-monitor

--- Registering with engine ---
Registered with manifest: BFT Monitor v0.1.0

--- Starting plugin ---
[BFT Monitor] Started with alert_radius_nm=5.0

--- Sending track updates ---
Dispatched track_update for EAGLE-01
Dispatched track_update for HAWK-02

--- Results ---
Output events: 1
  proximity_alert: EAGLE-01 ↔ HAWK-02 (2.3 nm)

Plugin Manifest — plugin.json

{
  "plugin_id": "sample-bft-monitor",
  "name": "BFT Monitor",
  "version": "0.1.0",
  "capabilities": [
    "subscribe:track_update",
    "publish:proximity_alert",
    "read:track"
  ],
  "dependencies": [],
  "classification": "UNCLASSIFIED",
  "ai_assisted": false
}

Running Tests

cd examples/sample-plugin
pip install -r requirements.txt
pytest test_plugin.py -v

Sample Mission Package

Location: examples/sample-mission-package/

Mission packages contain all data needed for offline operation in DDIL environments.

Package Structure

sample-mission-package/ ├── manifest.json # Package metadata, checksums, signature, expiry ├── route.json # Mission route waypoints ├── threats.json # Known threat entities ├── airspace.json # Airspace restrictions and boundaries ├── notams.json # Active NOTAMs for the area of operations └── bluforce.json # Initial blue-force track positions

Manifest Format

{
  "package_id": "mission-alpha-2026",
  "classification": "UNCLASSIFIED",
  "created_at": "2026-01-15T08:00:00Z",
  "expires_at": "2026-01-16T08:00:00Z",
  "area_of_operations": {
    "center": [38.9, -77.0],
    "radius_nm": 50
  },
  "checksums": {
    "route.json": "sha256:abc123...",
    "threats.json": "sha256:def456...",
    "airspace.json": "sha256:ghi789..."
  },
  "signature": "cosign:..."
}

Key Properties:

  • Expiry — packages become invalid after expires_at (staleness protection)
  • Checksums — SHA-256 integrity verification for every data file
  • Signature — Cosign signature for tamper detection
  • Classification — labeled with highest-classification content level

Writing Your Own Adapter

See the SDK Adapter Authoring Guide for step-by-step instructions:

  1. Extend BaseAdapter from adapters/base/base_adapter.py
  2. Set adapter_id and supported_formats class attributes
  3. Implement ingest(), validate(), and schema() methods
  4. Run against AdapterHarness to verify contract compliance
  5. Register in the adapter registry

Writing Your Own Plugin

See the SDK Plugin Authoring Guide for step-by-step instructions:

  1. Extend IPlugin from sdk/interfaces/plugin_interface.py
  2. Create a plugin.json manifest with capability declarations
  3. Implement on_start(), on_stop(), and on_event() methods
  4. Register with PluginRuntime
  5. Test with TestEngineHarness