matching: normalize route_short_name by parsing FahrtBezeichner [todo]

Author: derhuerst

lib/extract-data-srv-from-vdv-fahrtbezeichner.js

@@ -0,0 +1,27 @@
+import last from 'lodash/last.js'
+
+// Normalizes a string identifying a source of realtime data, extracted from a {Ist,Soll}Fahrt.FahrtID.FahrtBezeichner, so that it can be used in metric labels, logs, etc.
+// Note: The results are used for two purposes:
+// - In the matching process, it is used to determine the route_short_name of a VDV Fahrt.
+// - As a "data source" label in some metrics.
+// Prevents too many metric label combinations, which would cause many Prometheus time series to be created. (https://prometheus.io/docs/practices/naming/#labels)
+const extractDataSourceFromFahrtBezeichner = (fahrtBezeichner) => {
+	// Note: There might be FahrtBezeichner values like `75861#DLr-D#ODEG`.
+	const _parts = (fahrtBezeichner ?? '').split('#')
+	const src = _parts.length > 1 && last(_parts).trim() || null
+	if (src === null || src === '!ADD!') {
+		return 'unknown'
+	}
+	if (src.toLowerCase() === 'nahverkehrsgesellschaft jerichower land') {
+		return 'NJL'
+	}
+	if (src.length > 4) {
+		// We assume this case to be rather rare. In case a realtime data source is added or changed to such a long identifier, we'll detect that and adapt the mapping here.
+		return '-too-long-'
+	}
+	return src
+}
+
+export {
+	extractDataSourceFromFahrtBezeichner,
+}

lib/gtfs-matching.js

@@ -11,6 +11,9 @@ import {
 	connectToNats,
 	JSONCodec,
 } from './nats.js'
+import {
+	extractDataSourceFromFahrtBezeichner as defaultExtractDataSourceFromFahrtBezeichner,
+} from './extract-data-srv-from-vdv-fahrtbezeichner.js'
 import {
 	createMatchWithGtfs,
 } from './raw-match.js'
@@ -42,6 +45,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		natsConsumerName,
 		natsAckWait, // in milliseconds
 		matchConcurrency,
+		extractDataSourceFromFahrtBezeichner,
 		publishUnmatchedTripUpdates,
 	} = {
 		natsConsumerName: process.env.MATCHING_CONSUMER_NAME
@@ -59,6 +63,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			//     SELECT num_cores FROM cpu_cores LIMIT 1
 			// same as with hafas-gtfs-rt-feed: https://github.com/derhuerst/hafas-gtfs-rt-feed/blob/8.2.6/lib/match.js#L54-L61
 			: Math.ceil(1 + osCpus().length * 1.2),
+		extractDataSourceFromFahrtBezeichner: defaultExtractDataSourceFromFahrtBezeichner,
 		publishUnmatchedTripUpdates: process.env.MATCHING_PUBLISH_UNMATCHED_TRIPUPDATES
 			? process.env.MATCHING_PUBLISH_UNMATCHED_TRIPUPDATES === 'true'
 			: false,
@@ -123,6 +128,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		stop: stopMatching,
 	} = await createMatchWithGtfs({
 		logger,
+		extractDataSourceFromFahrtBezeichner,
 	})
 
 	const publishGtfsRtTripUpdateToNats = (gtfsRtTripUpdate, logCtx) => {

lib/raw-match.js

@@ -14,12 +14,14 @@ const formattingLogger = createLogger('formatting', {
 const createMatchWithGtfs = async (cfg) => {
 	const {
 		logger,
+		extractDataSourceFromFahrtBezeichner,
 	} = cfg
 
 	const {
 		formatVdvAusIstFahrtAsGtfsRtTripUpdate,
 	} = await createFormatVdvAusIstFahrtAsGtfsRtTripUpdate({
 		logger: formattingLogger,
+		extractDataSourceFromFahrtBezeichner,
 	})
 	const {
 		matchGtfsRtTripUpdateWithScheduleStopTimes,

lib/vdv-aus-istfahrt-as-gtfs-rt-tripupdate.js

@@ -1,5 +1,6 @@
 import {ok, deepStrictEqual} from 'node:assert/strict'
 import { strictEqual } from 'node:assert'
+import {OPERATORS} from './operators.js'
 import {
 	unixTimestampFromIso8601,
 } from './util.js'
@@ -23,6 +24,35 @@ const stripDataProviderPrefixFromAusHaltID = (ausHaltId) => {
 		: ausHaltId
 }
 
+// VBBr trips have a dual naming scheme, with both a letter and a line number, e.g.
+// - "5/522" from Fontanestr. to Wilhelmsdorf/Malge
+// AUS just uses the letter, while the GTFS uses both.
+const vbbrRouteShortNames = new Map([
+	['B', 'B/522'],
+])
+const getRouteShortName = (cfg) => {
+	const {
+		logger,
+		istFahrt,
+		extractDataSourceFromFahrtBezeichner,
+	} = cfg
+
+	if (!istFahrt.LinienText) {
+		logger.warn({
+			istFahrt,
+		}, 'cannot compute route_short_name without IstFahrt.LinienText')
+		return null
+	}
+	const linienText = istFahrt.LinienText.trim()
+
+	const src = extractDataSourceFromFahrtBezeichner(istFahrt.FahrtID.FahrtBezeichner)
+	if (src === 'VBBr' && vbbrRouteShortNames.has(linienText)) {
+		return vbbrRouteShortNames.get(linienText)
+	}
+
+	return linienText
+}
+
 const formatIstHaltAsStopTimeEvent = (istHalt, abfahrtAnkunft) => {
 	const plannedIso = istHalt[`${abfahrtAnkunft}szeit`] || null
 	if (plannedIso === null) {
@@ -108,6 +138,7 @@ const formatVdvAusIstHaltAsGtfsRtStopTimeUpdate = (istHalt) => {
 const createFormatVdvAusIstFahrtAsGtfsRtTripUpdate = async (cfg) => {
 	const {
 		logger,
+		extractDataSourceFromFahrtBezeichner,
 	} = cfg
 
 	const formatVdvAusIstFahrtAsGtfsRtTripUpdate = (istFahrt) => {
@@ -202,7 +233,10 @@ const createFormatVdvAusIstFahrtAsGtfsRtTripUpdate = async (cfg) => {
 		}
 
 		// not part of the GTFS Realtime spec, we just use it for matching and/or debug-logging
-		const route_short_name = istFahrt.LinienText || null
+		const route_short_name = getRouteShortName({
+			istFahrt,
+			extractDataSourceFromFahrtBezeichner,
+		})
 		Object.defineProperty(tripUpdate.trip, kRouteShortName, {value: route_short_name})
 		Object.defineProperty(tripUpdate, kFahrtID, {value: istFahrt.FahrtID ?? null})
 		Object.defineProperty(tripUpdate, kUmlaufID, {value: istFahrt.UmlaufID ?? null})

lib/vdv-reconciliation.js

@@ -3,7 +3,6 @@ import {
 	Counter,
 	Summary,
 } from 'prom-client'
-import last from 'lodash/last.js'
 import Redlock from 'redlock'
 import {createLogger} from './logger.js'
 import {register} from './metrics.js'
@@ -13,6 +12,9 @@ import {
 	connectToNats,
 	JSONCodec,
 } from './nats.js'
+import {
+	extractDataSourceFromFahrtBezeichner as defaultExtractDataSourceFromFahrtBezeichner,
+} from './extract-data-srv-from-vdv-fahrtbezeichner.js'
 import {connectToRedis} from './redis.js'
 import {
 	computeVdvFahrtId,
@@ -58,24 +60,7 @@ const runVdvReconciliation = async (cfg, opt = {}) => {
 		reconciliationConcurrency: process.env.RECONCILIATION_CONCURRENCY
 			? parseInt(process.env.RECONCILIATION_CONCURRENCY)
 			: 30,
-		// Normalizes a string identifying a source of realtime data, extracted from a {Ist,Soll}Fahrt.FahrtID.FahrtBezeichner, so that it can be used in metric labels, logs, etc.
-		// Prevents too many metric label combinations, which would cause many Prometheus time series to be created. (https://prometheus.io/docs/practices/naming/#labels)
-		extractDataSourceFromFahrtBezeichner: (fahrtBezeichner) => {
-			// note: There might be FahrtBezeichner values like `75861#DLr-D#ODEG`.
-			const _parts = (fahrtBezeichner ?? '').split('#')
-			const src = _parts.length > 1 && last(_parts).trim() || null
-			if (src === null || src === '!ADD!') {
-				return 'unknown'
-			}
-			if (src.toLowerCase() === 'nahverkehrsgesellschaft jerichower land') {
-				return 'NJL'
-			}
-			if (src.length > 4) {
-				// We assume this case to be rather rare. In case a realtime data source is added or changed to such a long identifier, we'll detect that and adapt the mapping here.
-				return '-too-long-'
-			}
-			return src
-		},
+		extractDataSourceFromFahrtBezeichner: defaultExtractDataSourceFromFahrtBezeichner,
 		...opt,
 	}
 	ok(Number.isInteger(natsAckWait), 'opt.natsAckWait must be an integer')