Skip to main content

Writing an adapter

An adapter is a pure translator: external format in, CDM out. It has no configuration, no thresholds, no enrichment and no opinions. This page is the tutorial for writing one, built on the reference adapter — packages/cdm/synapse_cdm/adapters/pntmap.py, which is worth reading first because every rule the CDM cares about shows up in it at least once.

The worked example​

One PNTMAP GNSS interference alert, and the exact CDM output it produces under a frozen clock. Both files are read from the repository when this page is generated, not transcribed:

inputpackages/cdm/synapse_cdm/fixtures/pntmap/jamming_gulf_of_riga.json
{
  "alert_id": "PNTMAP-2026-04-29-0117",
  "alert_time": "2026-04-29T06:12:44Z",
  "valid_until": "2026-04-29T07:00:00Z",
  "severity": "critical",
  "interference": {
    "type": "jamming",
    "band": "L1",
    "signal_strength_dbm": -71.5,
    "confidence": 0.87,
    "affected_constellations": [
      "GPS",
      "GALILEO"
    ]
  },
  "emitter": {
    "emitter_id": "EMT-4471",
    "lat": 57.512,
    "lon": 21.884,
    "geolocation_method": "tdoa",
    "accuracy_m": 2500,
    "attribution": "hostile"
  },
  "affected_area": {
    "type": "Polygon",
    "coordinates": [
      [
        [
          21.6,
          57.3
        ],
        [
          22.15,
          57.3
        ],
        [
          22.15,
          57.72
        ],
        [
          21.6,
          57.72
        ],
        [
          21.6,
          57.3
        ]
      ]
    ]
  },
  "receiver_count": 14,
  "reporting_node": "PNT-SENSOR-LIEPAJA-02"
}
golden outputpackages/cdm/synapse_cdm/fixtures/pntmap/golden/jamming_gulf_of_riga.cdm.json
[
  {
    "affiliation": "HOSTILE",
    "attributes": {
      "entity_id_basis": "emitter.emitter_id",
      "interference_type": "jamming",
      "source_extras": {
        "emitter": {
          "attribution": "hostile",
          "geolocation_method": "tdoa"
        },
        "receiver_count": 14,
        "reporting_node": "PNT-SENSOR-LIEPAJA-02"
      },
      "symbol_basis": "derived from affiliation; the source states no SIDC"
    },
    "confidence": 0.87,
    "entity_id": "2bcbebf2-45ac-511c-a635-f115c0b9b7ac",
    "entity_type": "INTERFERENCE_SOURCE",
    "integrity": null,
    "kinematics": null,
    "object_kind": "entity",
    "ontology_types": [],
    "position": {
      "accuracy_m": 2500,
      "alt_m": null,
      "lat": 57.512,
      "lon": 21.884,
      "position_source": "ESTIMATED",
      "vertical": null
    },
    "quality": null,
    "residual": null,
    "schema_version": "3.0.0",
    "source": {
      "adapter": "pntmap",
      "adapter_version": "1.0.0",
      "format_name": "PNTMAP GNSS interference alert",
      "format_version": null,
      "observed_at": null,
      "original_id": null,
      "record_index": null,
      "source_hash": null,
      "synthetic": true,
      "system": "PNTMAP",
      "transformations": []
    },
    "source_ids": [
      {
        "external_id": "EMT-4471",
        "system": "PNTMAP"
      }
    ],
    "status": null,
    "symbol": "10260000000000000000",
    "valid_from": "2026-04-29T06:12:44.000Z",
    "valid_to": "2026-04-29T07:00:00.000Z"
  },
  {
    "event_id": "a24f7f97-3783-52af-96e3-7f465435341e",
    "event_type": "GNSS_INTERFERENCE",
    "geometry": {
      "coordinates": [
        [
          [
            21.6,
            57.3
          ],
          [
            22.15,
            57.3
          ],
          [
            22.15,
            57.72
          ],
          [
            21.6,
            57.72
          ],
          [
            21.6,
            57.3
          ]
        ]
      ],
      "type": "Polygon"
    },
    "integrity": null,
    "object_kind": "event",
    "observed_at": "2026-04-29T06:12:44.000Z",
    "oes": {
      "confidence": null,
      "effective_from": null,
      "effective_to": null,
      "entity_relations": [],
      "event_class": "OBSERVATION",
      "event_relations": [],
      "evidence": [],
      "extensions": {},
      "security": null,
      "spec_version": "0.1.0",
      "status": null,
      "type_id": "sc.pnt.gnss_interference.v1",
      "verification": null
    },
    "payload": {
      "frequency_band": "L1",
      "interference_type": "JAMMING",
      "signal_strength_dbm": -71.5,
      "source_extras": {
        "affected_constellations": [
          "GPS",
          "GALILEO"
        ]
      }
    },
    "quality": null,
    "received_at": "2026-04-29T06:15:00.000Z",
    "related_entities": [
      "2bcbebf2-45ac-511c-a635-f115c0b9b7ac"
    ],
    "residual": null,
    "schema_version": "3.0.0",
    "severity": "CRITICAL",
    "source": {
      "adapter": "pntmap",
      "adapter_version": "1.0.0",
      "format_name": "PNTMAP GNSS interference alert",
      "format_version": null,
      "observed_at": null,
      "original_id": null,
      "record_index": null,
      "source_hash": null,
      "synthetic": true,
      "system": "PNTMAP",
      "transformations": []
    },
    "source_ids": [
      {
        "external_id": "PNTMAP-2026-04-29-0117",
        "system": "PNTMAP"
      }
    ],
    "status": null
  }
]

Six things to notice, because each one is a rule rather than a style choice.

One payload became two objects. The alert describes a thing that exists — an interference source, at a place, over an interval — and a thing that happened — interference, observed, over an area. Different canonical kinds, so to_cdm() returns [entity, event] and the event names the entity in related_entities.

affiliation is HOSTILE only because the payload said "attribution": "hostile". Remove that one key and the entity is UNKNOWN, despite being a jamming emitter in the Gulf of Riga. Inferring hostility would be an intelligence judgement made inside a translator — invisible to the audit trail, unattributable, and wrong the first time the "jammer" turns out to be a friendly EW exercise.

receiver_count: 14 and reporting_node survive in attributes.source_extras. Nothing mapped them; nothing dropped them either. affected_constellations survives as a list, which is the structure-preserving behaviour of lossless.residual().

The two identifiers went to different objects. The entity's source_ids holds EMT-4471 — the emitter. The event's holds PNTMAP-2026-04-29-0117 — the alert. That is what PNTMAP deduplicates on and what an auditor holding this event will search for.

entity_id_basis is stated. It says the id was keyed on emitter.emitter_id, which is stable across alerts. Had the emitter been anonymous, the basis would read alert_id — stable for that alert and not for the object — and a consumer accumulating a track needs to know which of the two it is holding.

received_at is 2026-04-29T06:15:00.000Z on every run, because the harness injects a frozen clock. That is what makes the golden file above a fixed expectation rather than a diff that changes every second.

FileSHA-256
packages/cdm/synapse_cdm/fixtures/pntmap/jamming_gulf_of_riga.json998a72caaab19c92aa864770f6a9690f819f615f50f1040aab0b685cde07cf2b
packages/cdm/synapse_cdm/fixtures/pntmap/golden/jamming_gulf_of_riga.cdm.json2eb13fdc250a46675e3296b9818ccc7e8a5e8b1e922bf11a174a7b4ebd976d99

1. Declare the class​

The sketch below is now a finished adapter

adapters/tak.py ships as adapter #2 and adapters/ais.py as #3, so the class outlined here exists for real, twice. Read pntmap.py first — every rule appears in it at least once — then tak.py, which is where the awkward cases live: XML rather than JSON, a bidirectional from_cdm(), a source sentinel that must become null, an enum collapse that has to stay recoverable, and the two fixture forms a non-JSON adapter needs in order to be checked at all.

Read ais.py when your format is harder than that: a binary payload rather than text, ten in-band sentinels rather than one (including a 0.0 that is also a plausible reading), several message types behind one entry point, a time field that states a second and no date — and no extension point at all on the way out, so the egress direction has to name what it cannot carry instead of parking it.

The contract is checked at class-definition time, so a mistake here fails at import rather than at 03:00 on the first outbound push.

from synapse_cdm.adapter import Adapter
from synapse_cdm.models import CDMBase


class TakAdapter(Adapter):
    name = "tak"                     # unique; how the harness and every SourceRef name you
    version = "0.1.0"                # semver; goes into source.adapter_version
    direction = "bidirectional"      # then you MUST override from_cdm()
    system = "TAK"

    metadata = AdapterMetadata(...)  # required; see "2. Declare what you are" below

    TRANSFORMS = {"event.@time": "re-rendered into the CDM's fixed-millisecond form"}

    def to_cdm(self, raw: bytes | dict) -> list[CDMBase]:
        ...

Six gates run when Python defines the class:

GateFails when
identityname, version, direction or system is missing — provenance that cannot name its translator is not provenance
directionnot one of ingest, egress, bidirectional
capabilityegress/bidirectional without overriding from_cdm() — an adapter that cannot emit must not claim it can
capabilityingest with from_cdm() overridden — declare bidirectional so the capability is discoverable
uniquenessname is already registered by another class
metadatametadata is absent, is not an AdapterMetadata, or its id, adapter_version or direction disagrees with the class's own

to_cdm() raises on a payload it cannot translate. It must not return a partial object and it must not return an empty list to signal failure: an empty list means "this payload legitimately carries nothing", and the two cases have to stay distinguishable.

2. Declare what you are​

Adapter API v2 requires a metadata block on every adapter class, and requires it at the same moment name and version are required — when Python defines the class. It is not optional and nothing is derived for you: a maturity level, a licence class or a format edition that the framework invented would be this repository making a claim about somebody else's standard that nobody checked.

from synapse_cdm.manifest import (AdapterMetadata, Capabilities, ClaimStatus, Direction,
                                  Evidence, FormatRef, LicenseClass, Limits, Maturity,
                                  MaturityLevel, Residual)


class TakAdapter(Adapter):
    ...
    metadata = AdapterMetadata(
        id="tak",                                  # == the class's `name`
        name="Cursor-on-Target",                   # for a human
        adapter_version="0.1.0",                   # == the class's `version`
        format=FormatRef(name="Cursor-on-Target (TAK)", version=None),
        direction=Direction.BIDIRECTIONAL,         # == the class's `direction`
        license_class=LicenseClass.PUBLIC_GOVERNMENT,
        maturity=Maturity(level=MaturityLevel.L4, basis="what verified it, and where",
                          external_exercise=None),
        claim_status=ClaimStatus.VERIFIED,
        claim_external_system=None,
        profiles=[],                               # SC-OES profiles you produce for
        capabilities=Capabilities(
            wire=True,
            directions_exercised=["ingest", "egress"],
            message_types=["CoT atoms (`a-.-...`)"],
            limits=Limits(max_input_bytes=None, ..., absent_because={...}),
        ),
        limitations=["what this adapter does NOT do — at least one line"],
        limitations_empty_reason=None,
        residual=Residual.LEGACY,
        payload_adapter=None,
        constituents=[],
        evidence=Evidence(available=False),
    )

Running python -m synapse_cdm.manifests --out manifests writes manifests/<id>.json from that declaration, and --check fails on a file that is missing, stale or left behind by an adapter that no longer exists. The schema those files validate against is generated too, into schemas/manifests/adapter-manifest.schema.json. Never edit either by hand — the declaration is the authority and the files are its publication, exactly as /schemas is for the models.

The rules that are easy to get wrong​

  • limitations may not be empty by default. Every adapter has them — an edition it does not implement, a message type it declines, a field with no canonical home — and one declaring none is one nobody has audited. If you genuinely have none, say so in limitations_empty_reason; silence is not the same statement.
  • A format.version you cannot read from a document is None, never a guess, and a limitation has to say that no document states it. An invented edition number is a claim about a publisher's document, and a consumer checking compatibility against it is checking a guess.
  • Every one of the five limits is either a number or absent WITH a reason. "This format does not nest" is a fact about the format and belongs in the declaration; silence reads as an oversight.

Maturity — seven rungs, and each is a claim about evidence​

Declare the last rung your evidence positively supports. A rung passed vacuously is not a rung declared.

LevelNameWhat it asserts
L0DOCUMENTEDthe relationship to the source standard is documented
L1DECODEDthe source representation can be parsed
L2CANONICALthe source can be translated into the CDM
L3PROVENANCE VERIFIEDthe required provenance survives translation
L4ROUNDTRIP VERIFIEDapplicable information survives source → CDM → source
L5PUBLIC CONFORMANCE VERIFIEDevery applicable public conformance gate passes
L6EXTERNALLY EXERCISEDverified against an independent real implementation

L6 cannot be awarded from this repository's own evidence. Every fixture here is synthetic, so the top rung is not the gate-writer's to award; the model refuses an L6 that names no external system, date and record.

An ingest-only adapter stops at L3: there is no egress direction for information to be lost in, so the roundtrip check is inapplicable rather than absent — and inapplicable is never printed or recorded as PASS.

Claim status — a separate axis, never derived from maturity​

StatusWhat it asserts
DOCUMENTEDthe mapping is written down
IMPLEMENTEDcode exists that performs the mapping
PROVISIONALthe mapping passes this repository's public gates against a provisional internal profile (binding: provisional-internal-profile); the gates say nothing about the standard's own encoding. Added 2026-09-20; the only status a provisional binding may claim above IMPLEMENTED
VERIFIEDthe mapping passes this repository's public gates against the standard's own encoding (binding: standard-encoding)
EXERCISEDit has been run against an independent implementation
INTEGRATEDintegration with a NAMED external system has actually occurred
DEPLOYEDit is running in a named operational or exercise deployment

The two axes are separate because one is about how thoroughly this repository has checked the translation and the other is about what has happened in the world — and the second cannot be earned by running a test. INTEGRATED and DEPLOYED are refused unless they name the external system, because the name is part of the claim.

3. Map the fields​

packages/cdm/synapse_cdm/FORMAT_COVERAGE.md already holds the Cursor-on-Target, STANAG 4676 and GeoJSON mappings row by row, with every known gap named. That table is your specification — and a test resolves every path in its CDM field column against the actual Pydantic models, so a renamed field breaks the build rather than just the prose.

4. Park everything else​

List the dotted paths you consumed, hand them to lossless.residual(), and put the result in attributes["source_extras"] or payload["source_extras"].

CONSUMED_TOP = ("alert_id", "alert_time", "valid_until", "severity")

extras = lossless.residual(alert, (*CONSUMED_TOP, *CONSUMED_EMITTER, "interference"))

Do not enumerate leftovers by hand. The block a source adds in its next firmware release is exactly the one nobody remembers, and residual() collects it without having been told it exists.

Keeping the consumed paths as data rather than burying them in the translation code is deliberate: "what does this adapter understand?" is then answerable by reading one list.

5. Refuse what you cannot read​

A missing required field, or an unmappable severity, raises. Do not default it — an alert that arrives labelled INFO because its severity was unreadable is worse than one that fails loudly.

An enum that has an UNKNOWN member is a different case: use it, and keep the source's own word in attributes. Stating "not known" is honest; inventing JAMMING is not.

Source sentinels are translated, never forwarded. AIS says "speed unknown" with 102.3; CoT says it with 9999999. An adapter that forwards either puts a ship at 102 knots on a commander's map. Because the sentinel's value then appears nowhere in the output, the translation is declared in TRANSFORMS — which is what that mechanism is for.

6. Ship fixtures​

At least three synthetic payloads under packages/cdm/synapse_cdm/fixtures/<name>/, including one that exercises the awkward path: a missing position, an unknown type, a vendor block you have never seen. No real data, ever.

The reference adapter's four fixtures are chosen on that principle — a full alert, an alert with no geolocated emitter, an alert with an unknown interference type and unmapped vendor fields, and an emitter on the equator at longitude zero, which exists purely to prove that 0.0 is treated as a coordinate and not as absence.

7. Record the golden output and read it​

python -m synapse_cdm.harness --adapter tak --update-golden
--update-golden is how a defect becomes the expectation

Review the diff before committing it. Both defects found while building the reference adapter were caught by reading a golden file, not by a failing test.

8. Add the tests​

Copy the shape of tests/test_cdm_pntmap_adapter.py: one test per claim in your adapter's docstring.

If you are bidirectional, the harness already round-trips you — declare direction = "bidirectional", override from_cdm(), and the roundtrip column judges what comes back. A JSON emitter is compared by values, not bytes: key order changes and omitted optional fields come back explicit, so the check asks whether any source value went missing on the way out.

An emitter of anything else is compared under the tolerance its class declares in ROUNDTRIP_TOLERANCE, which the report prints. bytes, the default, means from_cdm(to_cdm(raw)) must equal the byte fixture octet for octet — the claim every binary and line-oriented codec here makes — and the parsed twin beside it reports SKIP. values is for XML, whose attribute order and whitespace no emitter can promise: the harness re-ingests what you emitted and no source value may be missing from the parsed twin, and the byte fixture reports SKIP. A value egress legitimately re-stamps goes in ROUNDTRIP_TRANSFORMS with its reason, beside TRANSFORMS, and is printed with every report. Override roundtrip_reference() only to strip an envelope the source format itself says is not part of the message — the harness names every fixture it does that for. Your own round-trip test in tests/ is still worth writing; it is your statement of the claim, and the harness's column is the suite's.

The harness​

python -m synapse_cdm.harness --adapter <name|module:Class> [--fixtures <dir>] [--json]
                              [--schemas schemas] [--now <RFC3339>] [--update-golden]
                              [--synthetic true|false]
python -m synapse_cdm.harness --list-adapters [--json]
--list-adapters landed in 1.1.0

It prints the registry — name, version, direction, fixture directory, system — and exits 0, so the roster stops being something you can only see by getting a name wrong. It has shipped in every release since 1.1.0 (MIGRATIONS.md's 1.1.0 entry records it), so the release PyPI serves has it.

--fixtures is optional for an adapter the package ships and required for one given as module:ClassName. Omitted, the harness asks the import system where the package's own fixtures are and replays those — so --adapter pntmap is the whole command whether you have a clone or only a pip install. For your adapter, which lives outside the package, the harness will not guess at a directory it cannot know: it refuses with exit code 2 rather than reaching for fixtures/<your name>, which would either miss or, worse, hit ours.

Six checks per fixture, and an unrun check reports SKIP, never PASS — a capability nobody tested must not acquire a green tick.

CheckFails when
translateto_cdm() raised. One bad fixture never stops the run; the rest are still judged
schemaan object violates the published JSON Schema in /schemas — not the model, which would be testing the model against itself
provenancesource.* incomplete, synthetic unstated, source_ids empty, an event missing a timestamp
losslesswith MAPPINGS declared, a declared source leaf is LOST — missing, mismatched in value or type, or on the wrong object — under the path-bound preservation ledger; without them, a source value appears nowhere in the output and is not a declared transform (the value-presence heuristic, whose empty result is not proof; the report's preservation.basis says which ran)
roundtripfor an egress/bidirectional adapter, a source value is absent from what from_cdm() emitted. SKIP for ingest-only
goldenthe output differs from the recorded expectation, reported path by path

Nothing in the harness knows anything about any particular adapter. It resolves module:ClassName as readily as a registered name, which is what makes it usable as the gate for adapters the AI adapter factory generates and this repository has never seen — and that property is the whole design constraint.

Determinism​

The clock is frozen (times.FROZEN_NOW) unless --now says otherwise, and ids are derived rather than drawn. A fixture therefore produces identical bytes on every machine, which is the only reason the golden diff means anything. An adapter that reaches for datetime.now() or uuid4() fails the golden check on its second run.