feat(roads): extract NDW DATEX fields that were only in sourceRaw

Audited the live NDW situations feed (972 records, 97 distinct leaf
paths) against the parser. Confirmed nothing is lost at ingest: sourceRaw
keeps every leaf (36,719 values, no namespace-collision drops) and
persists to JSONB. But several fields with a typed home stayed buried
there, and the emitter read path dropped four typed columns.

Promote to typed fields (DATEX parser):
- subtype now falls back to the record's specific discriminator
  (accidentType, obstructionType, generalNetworkManagementType for
  movable-bridge openings, abnormalTrafficType, speedManagementType)
  instead of the generic record-class name; causeType still wins.
- detourGeometry: the alternativeRoute diversion polyline (was dropped;
  38 of 814 live records carry one).
- schedule: validPeriod date windows (53 live records).
- externalRefs.external: the provider location code (NDW RIS-index, the
  Dutch road-register reference; 28 live records). NDW carries no human
  roadName/roadNumber, so this is the closest it gives to a road identity.
- causeDescription as a headline/description fallback.

Fix a read-path loss in readObservations: schedule, confidence,
is_forecast and related_ids are written to typed columns but were never
selected back, so the public emitters (GeoJSON/TraFF/JSON-LD) silently
dropped them. Now read and reconstructed.

Adds MultiLineStringGeometry to @openconditions/core. Verified against
the live feed; a testcontainer round-trip covers the new fields.

Author: Medformatik

packages/core/src/model.ts

@@ -1,7 +1,8 @@
-import type { Geometry, LineString, Point } from "geojson";
+import type { Geometry, LineString, MultiLineString, Point } from "geojson";
 
 export type GeoJsonGeometry = Geometry;
 export type LineStringGeometry = LineString;
+export type MultiLineStringGeometry = MultiLineString;
 export type PointGeometry = Point;
 
 export interface Attribution {

packages/core/src/readObservations.ts

@@ -32,6 +32,10 @@ interface Row {
   data_updated_at: string | Date;
   fetched_at: string | Date;
   expires_at: string | Date | null;
+  schedule: Observation["schedule"] | null;
+  confidence: string | null;
+  is_forecast: boolean | null;
+  related_ids: string[] | null;
   attributes: Record<string, unknown> | null;
   subject: Observation["subject"] | null;
   origin: Provenance;
@@ -63,6 +67,10 @@ function rowToObservation(row: Row): Observation {
     origin: row.origin,
     ...(row.subject ? { subject: row.subject } : {}),
     ...(row.label != null ? { label: row.label } : {}),
+    ...(row.schedule ? { schedule: row.schedule } : {}),
+    ...(row.confidence != null ? { confidence: row.confidence as Observation["confidence"] } : {}),
+    ...(row.is_forecast != null ? { isForecast: row.is_forecast } : {}),
+    ...(row.related_ids ? { relatedIds: row.related_ids } : {}),
   };
   const specific =
     row.kind === "measurement"
@@ -123,6 +131,7 @@ export async function readObservations(
       severity, severity_source, headline, description, label,
       metric, value, level, unit, aggregation,
       status, valid_from, valid_to, data_updated_at, fetched_at, expires_at,
+      schedule, confidence, is_forecast, related_ids,
       attributes, subject, origin,
       ST_AsGeoJSON(geom) AS geojson,
       (stale_after IS NOT NULL AND stale_after < now()) AS is_stale

packages/roads/src/__tests__/datex.test.ts

@@ -369,3 +369,83 @@ describe("parseDatexSituations — OpenLR unresolved markers", () => {
     expect(events[0]!.externalRefs?.openlr).toBeUndefined();
   });
 });
+
+describe("parseDatexSituations — full v3 field extraction", () => {
+  it("keeps causeType as the subtype when a cause is present", () => {
+    const xml = v3Record(
+      `<cause><causeType>roadMaintenance</causeType></cause><accidentType>multiVehicleAccident</accidentType>${POINT_LOC}`
+    );
+    expect(parseDatexSituations(xml, NDW_SOURCE)[0]!.subtype).toBe("roadMaintenance");
+  });
+
+  it("falls back to the record's specific sub-type when there is no cause", () => {
+    const cases: [string, string][] = [
+      ["generalNetworkManagementType", "bridgeSwingInOperation"],
+      ["abnormalTrafficType", "slowTraffic"],
+      ["accidentType", "multiVehicleAccident"],
+      ["obstructionType", "objectOnTheRoad"],
+      ["speedManagementType", "speedRestrictionInOperation"],
+    ];
+    for (const [el, val] of cases) {
+      const xml = v3Record(`<${el}>${val}</${el}>${POINT_LOC}`);
+      expect(parseDatexSituations(xml, NDW_SOURCE)[0]!.subtype).toBe(val);
+    }
+  });
+
+  it("extracts the alternativeRoute diversion polyline as detourGeometry", () => {
+    const xml = v3Record(
+      `<reroutingItineraryDescription>Follow signs</reroutingItineraryDescription>` +
+        `<alternativeRoute xsi:type="ItineraryByIndexedLocations"><locationContainedInItinerary index="0">` +
+        `<location xsi:type="LinearLocation"><gmlLineString srsName="WGS 84"><posList>52.0 13.0 52.1 13.1</posList></gmlLineString></location>` +
+        `</locationContainedInItinerary></alternativeRoute>${POINT_LOC}`
+    );
+    const [ev] = parseDatexSituations(xml, NDW_SOURCE);
+    expect(ev!.detour).toBe("Follow signs");
+    expect(ev!.detourGeometry).toEqual({
+      type: "LineString",
+      coordinates: [
+        [13.0, 52.0],
+        [13.1, 52.1],
+      ],
+    });
+  });
+
+  it("extracts validPeriod windows into schedule, keeping overall start/end", () => {
+    const xml = `<?xml version="1.0" encoding="UTF-8"?>
+<messageContainer xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" modelBaseVersion="3"><payload xsi:type="SituationPublication"><situation id="S"><situationRecord xsi:type="RoadOrCarriagewayOrLaneManagement" id="R" version="1">
+<situationRecordVersionTime>2024-01-01T00:00:00Z</situationRecordVersionTime>
+<validity><validityStatus>definedByValidityTimeSpec</validityStatus><validityTimeSpecification>
+<overallStartTime>2026-06-01T00:00:00Z</overallStartTime><overallEndTime>2026-07-01T00:00:00Z</overallEndTime>
+<validPeriod><startOfPeriod>2026-06-10T06:00:00Z</startOfPeriod><endOfPeriod>2026-06-10T18:00:00Z</endOfPeriod></validPeriod>
+</validityTimeSpecification></validity>
+${POINT_LOC}</situationRecord></situation></payload></messageContainer>`;
+    const [ev] = parseDatexSituations(xml, NDW_SOURCE);
+    expect(ev!.validFrom).toBe("2026-06-01T00:00:00Z");
+    expect(ev!.validTo).toBe("2026-07-01T00:00:00Z");
+    expect(ev!.schedule).toEqual([
+      { dateStart: "2026-06-10T06:00:00Z", dateEnd: "2026-06-10T18:00:00Z" },
+    ]);
+  });
+
+  it("uses causeDescription as a headline fallback when there is no public comment", () => {
+    const xml = v3Record(
+      `<cause><causeDescription><values><value lang="en">Roadworks ahead</value></values></causeDescription></cause>${POINT_LOC}`
+    );
+    expect(parseDatexSituations(xml, NDW_SOURCE)[0]!.headline).toBe("Roadworks ahead");
+  });
+
+  it("extracts the external location reference (NDW RIS-index)", () => {
+    const xml = v3Record(
+      `<locationReference xsi:type="PointLocation">` +
+        `<externalReferencing><externalReferencingSystem>RIS-index</externalReferencingSystem>` +
+        `<externalLocationCode>NLUTC0226A0578000005</externalLocationCode></externalReferencing>` +
+        `<pointByCoordinates><pointCoordinates><latitude>52</latitude><longitude>13</longitude></pointCoordinates></pointByCoordinates>` +
+        `</locationReference>`
+    );
+    const [ev] = parseDatexSituations(xml, NDW_SOURCE);
+    expect(ev!.externalRefs?.external).toEqual({
+      system: "RIS-index",
+      code: "NLUTC0226A0578000005",
+    });
+  });
+});

packages/roads/src/datex.ts

@@ -1,5 +1,5 @@
 import { normaliseSeverity } from "@openconditions/core";
-import type { Confidence } from "@openconditions/core";
+import type { Confidence, RecurringWindow } from "@openconditions/core";
 import type { Geometry } from "geojson";
 import type { Restriction, RoadEvent, UnresolvedRoadEvent } from "./model.js";
 import { dedupeRoadEvents } from "./dedupe.js";
@@ -115,7 +115,14 @@ function parseLatLonList(raw: string | undefined): [number, number][] {
  * no coordinate geometry (Alert-C/TMC only) return null (decoded in Phase 2).
  */
 function resolveGeometry(rec: XmlObject): Geometry | null {
-  const locRef = getXmlChild(rec, "locationReference") ?? getXmlChild(rec, "groupOfLocations");
+  return resolveGeometryFrom(
+    getXmlChild(rec, "locationReference") ?? getXmlChild(rec, "groupOfLocations")
+  );
+}
+
+/** Walk a location subtree (locationReference, groupOfLocations, alternativeRoute,
+ * …) for any coordinate-bearing element and assemble a GeoJSON geometry. */
+function resolveGeometryFrom(locRef: XmlObject | undefined): Geometry | null {
   if (!locRef) return null;
 
   const lines: [number, number][][] = [];
@@ -164,6 +171,38 @@ function resolveGeometry(rec: XmlObject): Geometry | null {
   return null;
 }
 
+/** The diversion route geometry from an `alternativeRoute`, when it is linear. */
+function detourGeometryOf(rec: XmlObject): RoadEvent["detourGeometry"] {
+  const g = resolveGeometryFrom(getXmlChild(rec, "alternativeRoute"));
+  return g && (g.type === "LineString" || g.type === "MultiLineString") ? g : undefined;
+}
+
+/**
+ * validPeriod windows → schedule: each bounded date range (startOfPeriod /
+ * endOfPeriod) plus the daily time window (recurringTimePeriodOfDay) when the
+ * source supplies one. The overall start/end stay on validFrom/validTo.
+ */
+function scheduleOf(timeSpec: XmlObject | undefined): RecurringWindow[] | undefined {
+  if (!timeSpec) return undefined;
+  const out: RecurringWindow[] = [];
+  for (const vp of getXmlChildren(timeSpec, "validPeriod")) {
+    const win: RecurringWindow = {};
+    const ds = getXmlChildText(vp, "startOfPeriod");
+    const de = getXmlChildText(vp, "endOfPeriod");
+    if (ds) win.dateStart = ds;
+    if (de) win.dateEnd = de;
+    const tod = getXmlChild(vp, "recurringTimePeriodOfDay");
+    if (tod) {
+      const ts = getXmlChildText(tod, "startTimeOfPeriod");
+      const te = getXmlChildText(tod, "endTimeOfPeriod");
+      if (ts) win.timeStart = ts;
+      if (te) win.timeEnd = te;
+    }
+    if (Object.keys(win).length > 0) out.push(win);
+  }
+  return out.length > 0 ? out : undefined;
+}
+
 function directionOf(rec: XmlObject): string | undefined {
   const locRef = getXmlChild(rec, "locationReference");
   if (!locRef) return undefined;
@@ -234,14 +273,33 @@ function lanesOf(rec: XmlObject): RoadEvent["lanesAffected"] | undefined {
   return lanes.closed != null || lanes.total != null ? lanes : undefined;
 }
 
-/** Source cause/obstruction subtype (e.g. "roadMaintenance", "brokenDownVehicle"). */
-function causeOf(rec: XmlObject): string | undefined {
+/**
+ * The most specific source sub-classification for the record. The cause's
+ * causeType wins when present (e.g. "roadMaintenance"); otherwise the record's
+ * own typed discriminator is used — each DATEX situationRecord subclass carries
+ * its own (accidentType, obstructionType, generalNetworkManagementType for
+ * movable-bridge openings, abnormalTrafficType for congestion detail,
+ * speedManagementType, …). Without this, those records fell back to the generic
+ * record class name and lost their specific kind.
+ */
+function subtypeOf(rec: XmlObject): string | undefined {
   return (
     getXmlChildText(getXmlChild(rec, "cause"), "causeType") ??
-    getXmlChildText(rec, "vehicleObstructionType")
+    getXmlChildText(rec, "accidentType") ??
+    getXmlChildText(rec, "obstructionType") ??
+    getXmlChildText(rec, "environmentalObstructionType") ??
+    getXmlChildText(rec, "vehicleObstructionType") ??
+    getXmlChildText(rec, "generalNetworkManagementType") ??
+    getXmlChildText(rec, "abnormalTrafficType") ??
+    getXmlChildText(rec, "speedManagementType")
   );
 }
 
+/** Human cause text (DATEX `cause/causeDescription`), a multilingual block. */
+function causeDescriptionOf(rec: XmlObject): string | undefined {
+  return multilingual(getXmlChild(getXmlChild(rec, "cause"), "causeDescription"), "en");
+}
+
 function speedLimitOf(rec: XmlObject): number | undefined {
   const raw = getXmlChildText(rec, "temporarySpeedLimit");
   if (raw == null) return undefined;
@@ -359,7 +417,8 @@ function collectOpenLr(rec: XmlObject): string | undefined {
   return getXmlChildText(rec, "openlrBinary") ?? undefined;
 }
 
-function collectRefs(rec: XmlObject): RoadEvent["externalRefs"] {
+/** Alert-C/TMC reference (country + table + primary specific-location code). */
+function tmcOf(rec: XmlObject): NonNullable<RoadEvent["externalRefs"]>["tmc"] | undefined {
   const locRef = getXmlChild(rec, "locationReference");
   if (!locRef) return undefined;
 
@@ -374,11 +433,26 @@ function collectRefs(rec: XmlObject): RoadEvent["externalRefs"] {
   const table = getXmlChildText(alertC, "alertCLocationTableNumber");
   const code = getXmlChildText(getXmlChild(primary, "alertCLocation"), "specificLocation");
   if (country && table && code) {
-    return { tmc: { country, table: parseFloat(table), code: parseInt(code, 10) } };
+    return { country, table: parseFloat(table), code: parseInt(code, 10) };
   }
   return undefined;
 }
 
+/** Provider external location code (NDW's `externalReferencing`, e.g. RIS-index). */
+function externalLocationOf(rec: XmlObject): { system: string; code: string } | undefined {
+  const er = getXmlChild(getXmlChild(rec, "locationReference"), "externalReferencing");
+  const system = getXmlChildText(er, "externalReferencingSystem");
+  const code = getXmlChildText(er, "externalLocationCode");
+  return system && code ? { system, code } : undefined;
+}
+
+function externalRefsOf(rec: XmlObject): RoadEvent["externalRefs"] {
+  const tmc = tmcOf(rec);
+  const external = externalLocationOf(rec);
+  if (!tmc && !external) return undefined;
+  return { ...(tmc ? { tmc } : {}), ...(external ? { external } : {}) };
+}
+
 interface SituationRecord {
   rec: XmlObject;
   situationSeverity: string;
@@ -506,6 +580,7 @@ export function parseDatexSituations(
 
     const publicComment = getXmlChild(rec, "generalPublicComment");
     const fallbackComment = getXmlChild(rec, "comment");
+    const causeDesc = causeDescriptionOf(rec);
 
     const shared = {
       id: `${src.id}:${recId(rec)}`,
@@ -514,7 +589,7 @@ export function parseDatexSituations(
       domain: "roads" as const,
       kind: "event" as const,
       type,
-      subtype: causeOf(rec) ?? recType ?? undefined,
+      subtype: subtypeOf(rec) ?? recType ?? undefined,
       category,
       isPlanned,
       ...severityFields,
@@ -528,16 +603,22 @@ export function parseDatexSituations(
       restrictions: dimensionRestrictionsOf(rec),
       vehiclesAffected: vehiclesAffectedOf(rec),
       detour: detourOf(rec),
+      detourGeometry: detourGeometryOf(rec),
       delaySeconds: leafNumber(rec, "delayTimeValue"),
       queueLengthMeters: leafNumber(rec, "queueLength"),
       relatedIds: relatedRefsOf(rec),
       sourceRaw: rec,
       headline:
         multilingual(publicComment, "en") ??
         multilingual(fallbackComment, "en") ??
+        causeDesc ??
         defaultHeadline(type),
       description:
-        multilingual(publicComment, "en") ?? multilingual(fallbackComment, "en") ?? undefined,
+        multilingual(publicComment, "en") ??
+        multilingual(fallbackComment, "en") ??
+        causeDesc ??
+        undefined,
+      schedule: scheduleOf(timeSpec),
       validFrom: text(timeSpec?.["overallStartTime"]) ?? null,
       validTo: text(timeSpec?.["overallEndTime"]) ?? null,
       origin: {
@@ -557,14 +638,14 @@ export function parseDatexSituations(
       withGeom.push({
         ...shared,
         geometry,
-        externalRefs: collectRefs(rec),
+        externalRefs: externalRefsOf(rec),
       });
     } else {
       // openlr is defined here because we checked !geometry && !openlr above.
       unresolved.push({
         ...shared,
         geometry: undefined,
-        externalRefs: { ...collectRefs(rec), openlr: openlr! },
+        externalRefs: { ...externalRefsOf(rec), openlr: openlr! },
       });
     }
   }

packages/roads/src/model.ts

@@ -2,6 +2,7 @@ import type {
   ConditionEvent,
   LineStringGeometry,
   Measurement,
+  MultiLineStringGeometry,
   PointGeometry,
 } from "@openconditions/core";
 
@@ -87,6 +88,9 @@ export interface RoadEvent extends ConditionEvent {
   restrictions?: Restriction[];
   vehiclesAffected?: string[];
   detour?: string;
+  /** Diversion/alternative-route geometry, when the source provides one
+   * (DATEX `alternativeRoute`); the `detour` string is its prose counterpart. */
+  detourGeometry?: LineStringGeometry | MultiLineStringGeometry;
   /** Quantified impact, when the source gives it. */
   delaySeconds?: number;
   queueLengthMeters?: number;
@@ -105,6 +109,10 @@ export interface RoadEvent extends ConditionEvent {
       direction?: number;
       extent?: number;
     };
+    /** A provider-specific external location code (e.g. NDW's RIS-index, the
+     * Dutch road-register reference) — the closest thing some feeds give to a
+     * road identity, decodable only against that provider's network dataset. */
+    external?: { system: string; code: string };
     linear?: unknown;
   };
   /** The original provider record, verbatim — a lossless passthrough so no
@@ -161,6 +169,7 @@ export function roadAttributes(ev: RoadEvent): Record<string, unknown> {
     attrs["vehiclesAffected"] = ev.vehiclesAffected;
   }
   if (ev.detour != null) attrs["detour"] = ev.detour;
+  if (ev.detourGeometry != null) attrs["detourGeometry"] = ev.detourGeometry;
   if (ev.delaySeconds != null) attrs["delaySeconds"] = ev.delaySeconds;
   if (ev.queueLengthMeters != null) attrs["queueLengthMeters"] = ev.queueLengthMeters;
   if (ev.workersPresent != null) attrs["workersPresent"] = ev.workersPresent;

services/ingest/src/__tests__/pipeline.test.ts

@@ -212,6 +212,18 @@ describe("store round-trip — typed columns + attributes JSONB", () => {
       workZoneType: "moving",
       speedLimitKph: 50,
       regions: ["Berlin"],
+      detourGeometry: {
+        type: "LineString",
+        coordinates: [
+          [13.4, 52.5],
+          [13.42, 52.51],
+        ],
+      },
+      schedule: [{ dateStart: "2026-06-10T06:00:00Z", dateEnd: "2026-06-10T18:00:00Z" }],
+      externalRefs: { external: { system: "RIS-index", code: "NL123" } },
+      confidence: "likely",
+      isForecast: true,
+      relatedIds: ["parent-1", "parent-2"],
       sourceRaw: { provider_field: "verbatim" },
       origin: { kind: "feed", attribution: { provider: "X", license: "CC0-1.0" } },
       dataUpdatedAt: "2026-06-23T10:00:00Z",
@@ -234,6 +246,20 @@ describe("store round-trip — typed columns + attributes JSONB", () => {
     expect(got!.workZoneType).toBe("moving");
     expect(got!.speedLimitKph).toBe(50);
     expect(got!.regions).toEqual(["Berlin"]);
+    expect(got!.detourGeometry).toEqual({
+      type: "LineString",
+      coordinates: [
+        [13.4, 52.5],
+        [13.42, 52.51],
+      ],
+    });
+    expect(got!.schedule).toEqual([
+      { dateStart: "2026-06-10T06:00:00Z", dateEnd: "2026-06-10T18:00:00Z" },
+    ]);
+    expect(got!.externalRefs?.external).toEqual({ system: "RIS-index", code: "NL123" });
+    expect(got!.confidence).toBe("likely"); // typed column, was dropped on read
+    expect(got!.isForecast).toBe(true);
+    expect(got!.relatedIds).toEqual(["parent-1", "parent-2"]);
     expect(got!.source).toBe("rt"); // feed id NOT clobbered by sourceRaw
     expect(got!.sourceRaw).toEqual({ provider_field: "verbatim" }); // verbatim passthrough survives
   }, 30_000);