Hoe onze enrichment pipeline 2.0 P2000-berichten begrijpt

Een rauw P2000-bericht is een korte, cryptische regel tekst. Geen JSON, geen gestructureerde velden — gewoon een string die bedoeld is voor piepers:

P1 BRW 05-3421 HV1 TS 06-4531 Woningbrand Keizersgracht 412 1016GD Amsterdam

Voor een brandweerman is dit genoeg: spoed, woningbrand, Keizersgracht, Amsterdam. Voor een computer is het een onontwarbare reeks afkortingen. De uitdaging: haal hier in milliseconden gestructureerde data uit — prioriteit, locatie, voertuigen, incidenttype, coördinaten — zonder fouten.

In dit artikel beschrijf ik hoe de enrichment pipeline 2.0 van Zwaailicht.nu dat doet, welke zeven enrichers we hebben, en welke we nog willen bouwen.

De architectuur: drop-in enrichers

De pipeline is ontworpen rond één principe: een nieuwe enricher toevoegen = één bestand aanmaken. Geen configuratie, geen registratie, geen imports bijwerken.

Elke enricher is een Python-klasse die overerft van BaseEnricher:

class BaseEnricher(ABC):
    name: str = "base"
    priority: int = 50          # volgorde (lager = eerder)
    output_field: str = ""      # veld op de context
    output_type: type = None    # dataclass voor het resultaat
    db_columns: dict = {}       # data_key → database_kolom

Bij het opstarten scant de pipeline automatisch alle bestanden in de enrichers/-map en ontdekt nieuwe enrichers zonder dat je ze ergens hoeft te registreren. Ze worden gesorteerd op priority en sequentieel uitgevoerd.

Typed context in plaats van losse dicts

Elke enricher ontvangt een getypeerde EnrichmentContext — een dataclass met expliciete velden voor alle mogelijke verrijkingen. Hierdoor kan de highway-enricher veilig context.priority.code opvragen in plaats van te hopen dat context["priority"]["code"] bestaat.

@dataclass
class EnrichmentContext:
    priority: PriorityData | None = None
    location: LocationData | None = None
    highway: HighwayData | None = None
    units: UnitData | None = None
    medical: MedicalData | None = None
    incident: IncidentData | None = None
    coordinates: tuple[float, float] | None = None

Enrichers die eerder draaien vullen de context in; enrichers die later draaien lezen die data. Zo kan de geocoding-enricher (die als laatste draait) het adres gebruiken dat de locatie-enricher heeft geëxtraheerd.

Sync en async

De meeste enrichers zijn pure regex-operaties — geen I/O, geen netwerkcalls. Alleen de geocoding-enricher maakt HTTP-verzoeken. Daarom ondersteunt de pipeline twee modi:

• Sync enrichers implementeren enrich_sync() — geen coroutine-overhead
• Async enrichers implementeren async def enrich() — voor I/O-operaties

De pipeline detecteert automatisch welk type elke enricher is en roept de juiste methode aan.

Automatische database-mapping

Elke enricher declareert via db_columns hoe zijn output naar de database wordt geschreven:

class MedicalEnricher(BaseEnricher):
    db_columns = {
        "codes": "medical_codes",
        "requires_specialist": "requires_specialist",
    }

Wanneer een melding wordt opgeslagen, itereert de storage-laag over alle enricher-resultaten en mapt ze dynamisch naar de juiste database-kolommen. Een nieuwe enricher die db_columns declareert, wordt automatisch opgeslagen zonder dat je de INSERT-query hoeft aan te passen.

De zeven enrichers

1. Priority (prioriteit = 10)

Extraheert de urgentiecode uit het bericht. Het Nederlandse alarmeringssysteem gebruikt een hiërarchie van codes:

De enricher zoekt met regex naar deze patronen, normaliseert ze (P 1 → P1, PRIO 1 → PRIO1), en slaat op of het een spoedmelding betreft (zwaailicht en sirene toegestaan).

2. Location (prioriteit = 20)

De locatie-enricher is het complexste onderdeel. P2000-berichten gebruiken honderden afkortingen voor plaatsnamen (SCHPHL voor Schiphol, RDMN voor Rotterdam), schrijven postcodes soms aan huisnummers vast, en mengen voertuigcodes door het adres.

De enricher combineert meerdere databronnen: - 200+ P2000-plaatsnaamafkortingen — handmatig samengesteld - CBS plaatsnamenlijst — alle officiële Nederlandse woonplaatsen - Postcodeherkenning — 4 cijfers + 2 letters (1234AB) - Straatnaampatronen — herkent Nederlandse suffixen (-straat, -weg, -laan, -gracht)

Het resultaat: straat, huisnummer, postcode, stad, en een betrouwbaarheidsscore.

3. Highway (prioriteit = 30)

Meldingen op snelwegen volgen een eigen formaat. A1 RE 25,4 KP Hoevelaken bevat: - Snelwegcode: A1 (autosnelweg), N201 (provinciale weg) - Richting: RE (rechts), LI (links), of een plaatsnaam - Hectometerpaal: HMP 25,4 — de exacte locatie - Knooppunt: KP Hoevelaken

De enricher herkent dit formaat en filtert vals-positieven: “A1” als snelweg versus “A1” als prioriteitscode. Hiervoor gebruikt hij de context — als de priority-enricher al “A1” als prioriteit heeft geëxtraheerd, is “A1” in de rest van het bericht waarschijnlijk geen snelweg.

4. Units (prioriteit = 40)

P2000-berichten vermelden welke eenheden zijn gealarmeerd. Meer dan 50 voertuigcodes worden herkend:

De enricher herkent ook kazernecodes (BRT-01, BDH-06), brandweereenheidnummers (06-1234), ambulancenummers (17104), en capcodes — de 6-7 cijferige pager-adressen waarmee individuele eenheden worden gealarmeerd. Het aantal capcodes geeft een indicatie van de omvang van de inzet.

5. Medical (prioriteit = 50)

Ambulancemeldingen bevatten vaak medische diagnosecodes. De enricher herkent:

• Standaardcodes: CVA (beroerte), ACS (acuut coronair syndroom), AED (defibrillator beschikbaar), TIA (mini-beroerte), OHCA (hartstilstand buiten ziekenhuis)
• Situationele patronen: reanimatie, trauma, ademnood, psychiatrisch
• Specialist-indicatoren: MMT, Lifeliner — duiden op een ernstig incident waarbij een traumachirurg per helikopter komt

De enricher combineert de diagnosecodes met de units-context: als een Lifeliner in de eenhedenlijst staat, markeert hij het als “specialist vereist” — ook als dat niet expliciet in de tekst staat.

6. Incident (prioriteit = 60)

De incident-enricher classificeert het type melding op basis van een register van 60+ patronen, geordend van specifiek naar algemeen:

1. GRIP-niveaus (1-4) — grootschalige incidenten met gecoördineerde inzet
2. Gevaarlijke stoffen — BRZO, ammoniak, chloor, asbest, gaslekkage
3. Brand — woningbrand, voertuigbrand, buitenbrand, industrie
4. Verkeer — aanrijding, beknelling, te water
5. Medisch — reanimatie, diverse medische termen
6. Water — waterongeval, persoon te water, duikteam
7. Hulpverlening — liftopsluiting, stormschade, dier in nood

Het register is geordend op specificiteit: “woningbrand” matcht voor het generiekere “brand”. Als geen patroon matcht, gebruikt de enricher de service-context: een onherkenbaar brandweerbericht wordt nog steeds als “fire” geclassificeerd.

Elk patroon in het register heeft een incident_class (de opslag-sleutel), een type en optioneel een subtype, plus vlaggen voor GRIP-niveau en gevaarlijke stoffen.

7. Geocoding (prioriteit = 100)

De laatste enricher in de keten — en de enige die async is. Hij neemt het adres dat de locatie- en snelweg-enrichers hebben geëxtraheerd en vertaalt het naar GPS-coördinaten via de Photon geocoder (gebaseerd op OpenStreetMap).

De query wordt opgebouwd uit de context: straat + huisnummer + stad, met “Netherlands” als landsafbakening. Resultaten worden gecached om herhaalde verzoeken voor hetzelfde adres te voorkomen. De geocoder valideert dat de coördinaten binnen de grenzen van Nederland vallen.

De flow, samengevat

"P1 BRW 05-3421 HV1 TS Woningbrand Keizersgracht 412 Amsterdam"
    │
    ├─ PriorityEnricher  → code=P1, level=1, is_emergency=true
    ├─ LocationEnricher  → street=Keizersgracht, house_number=412, city=Amsterdam
    ├─ HighwayEnricher   → (geen match)
    ├─ UnitsEnricher     → units=[HV1, TS], station=05-3421
    ├─ MedicalEnricher   → (geen match — brandweermelding)
    ├─ IncidentEnricher  → type=fire, class=woningbrand
    └─ GeocodingEnricher → lat=52.3654, lon=4.8872

Resultaat: een volledig verrijkte melding met 25+ gestructureerde velden, klaar voor opslag, kaartweergave en statistieken. Gemiddelde verwerkingstijd: onder de 100 milliseconden.

Toekomst: welke enrichers willen we nog bouwen?

De modulaire architectuur maakt het eenvoudig om nieuwe enrichers toe te voegen. Een paar ideeën:

Meerdere-meldingen-correlatie

Grote incidenten genereren meerdere P2000-berichten — eerst brandweer, dan ambulance, soms politie. Een correlatie-enricher zou meldingen die binnen een kort tijdvenster en dezelfde locatie vallen, aan elkaar koppelen. Zo kun je zien dat de 8 losse meldingen eigenlijk één grote woningbrand zijn.

Weer-context

Een weer-enricher zou actuele weergegevens toevoegen: windkracht, temperatuur, neerslag. Bij een buitenbrand is windkracht relevant (verspreidingsrisico). Bij gladheid-meldingen in de winter geeft de temperatuur context. Data is beschikbaar via de KNMI Open Data API.

Historische context

Een enricher die kijkt naar historische patronen op dezelfde locatie. Is dit de derde keer deze maand dat er een ambulance naar dit adres gaat? Zijn er vaker brandmeldingen in dit winkelcentrum? Dit geeft een extra laag context die helpt bij het beoordelen van de ernst.

Vervoerstijd-schatting

Met de coördinaten van de melding en de dichtstbijzijnde kazerne of ambulancepost, kan een enricher de geschatte aanrijtijd berekenen. We hebben al een OSRM-routeringservice draaien — dit zou een directe uitbreiding zijn.

Populatiedichtheid

Een enricher die op basis van de coördinaten de bevolkingsdichtheid van het gebied opzoekt. Een brand in een dichtbevolkt stadscentrum heeft een ander risicoprofiel dan een buitenbrand op een industrieterrein. CBS levert gedetailleerde buurtstatistieken die hiervoor bruikbaar zijn.

Objectherkenning

Sommige meldingen noemen specifieke objecten: “flat”, “ziekenhuis”, “basisschool”, “verzorgingshuis”, “station”. Een enricher die deze kwetsbare objecten herkent, kan de impact-inschatting verbeteren. Een brand in een verzorgingshuis is een ander verhaal dan een containerbrand.

Zelf bouwen?

De volledige broncode is beschikbaar. Een minimale enricher ziet er zo uit:

# enrichers/my_enricher.py

class MyEnricher(BaseEnricher):
    name = "my_enricher"
    priority = 70
    output_field = "my_data"
    output_type = MyDataClass
    db_columns = {"field": "db_column"}

    def enrich_sync(self, message, context):
        # Analyseer het bericht
        match = MY_PATTERN.search(message)
        if not match:
            return None
        return EnrichmentResult(
            enricher_name=self.name,
            data={"field": match.group(1)},
            confidence=0.9,
        )

Drop dit bestand in enrichers/, definieer een MyDataClass, voeg het veld toe aan EnrichmentContext, en de pipeline pikt het automatisch op.

Vragen of ideeën voor nieuwe enrichers? We horen het graag.