The raw data pipeline

The pipeline turns the three raw inputs of a region (an OpenStreetMap file, the vendor GPS files and the SPaT work orders) into the processed data that AppRegion loads. It is a chain of four commands, each reading the output of the previous one from the region’s processed directory:

build_traffic_network  ->  split_points  ->  match_points_to_map  ->  process_trajs

parse_spat runs independently of the chain and eva_trajs_qc checks its result. All four commands accept the master configuration and fan out over the sub-regions (Sub-regions). Each stage is idempotent for a given date, so a date range can be re-run after fixing the map or the parameters.

Data flow of the pipeline. Solid arrows are the data produced by each stage, dashed arrows the network files a stage reads.

Stage 1: build_traffic_network

Input (raw_map/): map.osm, optionally arterial.json and the override files overwrite/node.csv (node_id,controller_id) and overwrite/movement.csv (movement_id,movement_index).

What happens. For every sub-region, in parallel, build_network of mtldp.preproc parses the OSM XML and builds the Network (nodes, segments, links, movements, arterials, NetworkX graph and the signalized network) as described in Pre-processing: mtldp.preproc. The two thresholds of the signalized network come from the configuration file (search_link_length_threshold, min_link_entry_threshold). Then the network is exported:

Output in network/	Content
traffic_network.pickle	The complete Network object; the input of every later stage and of AppRegion.
filtered.osm	Directed OSM XML in which every segment is a one-way way; the reference map of the map matcher.
shapefile/edges.shp	The same segments as a shapefile with the fields osmid, u, v read by fmm.
traffic_network.json	GeoJSON layers of the whole network (Network geometry as GeoJSON).
signalized_traffic_network.json, signalized_arterial.json, signalized_selected_junctions.json	GeoJSON layers of the signalized network, of its arterials and of the junctions in junction_id_list.
overwrite.json	The consolidated overrides that were applied.
map.log	Build log with the warnings (missing lane tags, unresolved turns, …).

Finally the cached ubodt.txt of the map matcher is deleted, because it is only valid for the network it was computed from.

Typical map fixes. When the log or the review of the network reveals a problem, the fix goes into the raw inputs and the stage is re-run: two signal nodes at one intersection are merged with merge_osm (or in JOSM), a wrong controller mapping or movement index goes into the override CSV files, a missing corridor into arterial.json.

(mtldp) $ merge_osm -i raw_data/my_region/raw_map/map_raw.osm -o raw_data/my_region/raw_map/map.osm
(mtldp) $ build_traffic_network -c configs/my_region.json

Stage 2: split_points

Input: the raw vendor files, either in raw_trajs/ of the master region or in the directory given with --raw-traj-dir. --format-version selects the loader: GM_2024_pre and GM_2025 are CSV deliveries, SL_2025 is a Parquet delivery organised as a Hive tree (year=…/month=…/day=…/hour=…/*.parquet).

What happens. The raw files are read one at a time (one partition directory at a time for Parquet) and normalised to the common point schema: trip_id, traj_id (equal to trip_id at this stage), timestamp in seconds, latitude, longitude, speed in m/s, heading and elevation in meters. Privacy-fuzzed points of the SL delivery are discarded. For each sub-region the points inside its bounding box are selected, the local date, date_time and tod columns are computed in the region’s time zone, and the rows are appended to the file of their date.

Output: trajectories/raw_pts/<date>_raw.csv per sub-region and local date with the columns trip_id, traj_id, timestamp, longitude, latitude, elevation, speed, heading, date, date_time, tod.

Because the output is appended to, the stage must run once per delivery; running it twice on the same files duplicates the points (the next stage de-duplicates on trip_id and timestamp, but at a cost). delete_processed_trajs -d raw_pts clears the directory before a re-run.

(mtldp) $ split_points -c configs/my_region.json -v SL_2025 -d /data/deliveries/2025-03

Stage 3: match_points_to_map

Input: trajectories/raw_pts/<date>_raw.csv for every date of the range, and the network files shapefile/edges.shp, traffic_network.pickle and, if present, ubodt.txt.

What happens. For each sub-region a FmmModule is created once (candidate search radius 3000 m, 8 candidates per point, GPS error 100 m); the UBODT is loaded from ubodt.txt or computed and saved. Then for every date:

1. the raw points are sorted by trip_id and timestamp and de-duplicated;

2. points outside the sub-region bounding box are dropped;

3. the points are written in the fmm input format (id;x;y;timestamp), matched, and the result is read back: every point gets its segment_id, the matching error in meters, the offset along the segment and the shortest-path distance spdist since the previous point. Unmatched points are dropped, and a trip that leaves the network and comes back is restarted rather than discarded;

4. the segments are resolved against the network, adding link_id, upstream_intersection, downstream_intersection and junction_id (the downstream intersection);

5. numeric columns are rounded and the table is saved.

Output: trajectories/points/<date>_matched.csv, a Feather file despite its extension (read it with pandas.read_feather), with the raw point columns plus segment_id, error, offset, spdist, link_id, upstream_intersection, downstream_intersection and junction_id. The matching statistics of every run are appended to trajectories/points/mm_log.txt.

(mtldp) $ match_points_to_map -c configs/my_region.json -s 2025-03-03 -e 2025-03-09

Stage 4: process_trajs

Input: trajectories/points/<date>_matched.csv and traffic_network.pickle.

What happens. For every date, the matched points are read (only the columns needed), split into chunks of complete trips and processed in a pool of worker processes. Each worker runs process_points_to_trajs:

1. points without a matched link and points whose matching error exceeds lane number × max-away-per-lane + 15 meters are dropped;

2. the points of a trip are cut into one trajectory per link. The trajectory is attached to the movement from its link to the next link of the trip (or to the sink movement of the link at the end of the trip), and junction_id is the downstream intersection of the link;

3. with --fill-dummy-trips (the default), gaps between links that are not joined by a movement are filled along the shortest path with dummy trajectories without points;

4. the performance indices are computed for each trajectory (below).

In --process-mode signalized the matched points are first restricted to the segments of the signalized network and their link_id and junction_id are re-assigned on that network, so that every trajectory describes the approach to a signalized intersection along the extended signalized link. --filter-junctions keeps only the junctions listed in the sub-region configuration.

Output: trajectories/trajectories/<date>_trajs.csv (all mode) or <date>_signalized_trajs.csv (signalized mode), one row per trajectory. The columns are the attributes exported by mtldp.meta.Trajectory:

Column	Definition
trip_id, traj_id	The trip and the trajectory (<trip_id>_<n>, n counting the links of the trip; dummy trajectories add a second suffix).
date, tod, timestamp	Local date, time of day (hours) and epoch time of the first point.
points_num, avg_error	Number of points and mean map-matching error in meters (0 and -1 for dummies).
junction_id, link_id, segment_list	The downstream intersection, the link and the segments traversed (|-separated).
movement_id, movement_index	The movement to the next link (or <link>_dest) and its index (-1 for sink movements).
timestamp_list, latitude_list, longitude_list, speed_list, error_list	The point series, |-separated (None for dummies).
travel_time, travel_distance, avg_speed, dis_diff	Duration in seconds, path length of the GPS trace in meters, their ratio, and travel distance minus link length. These four are the only indices of --simple-process.
distance_list	Signed distance of every point to the junction along the path: negative upstream, 0 at the junction, positive downstream.
min_dis_to_junction, comments	Closest approach to the junction, and how the trace relates to it (Cross, NotReached, OverCross, Only1Point, NoPoints, with |AwayJunction appended when the closest approach exceeds ten meters per lane).
free_v	Free-flow speed in m/s: the fastest free-flow interval observed on the trajectory, never below 75 % of the link speed limit; the speed limit itself when no free-flow interval exists.
arrival_time, free_arrival_time	When the vehicle crossed the junction (interpolated at distance 0, extrapolated at free_v when it never did), and when it would have crossed had it driven at free_v from its first point.
control_delay, service_level	max(0, arrival_time - free_arrival_time) in seconds, and the Highway Capacity Manual level of service A to F derived from it (10, 20, 35, 55 and 80 s thresholds).
stop_nums, stop_delay, stop_details	Number of stops (speed below 1.5 m/s for at least 3.5 s), their total duration, and the distance and time span of each stop relative to arrival_time.
queue_dis	Distance in meters from the stop bar back to the farthest upstream stop of the vehicle (its position at the back of the queue); 0 when it never stopped.
split_failure	True when the vehicle stopped at least twice and was delayed 60 s or more, i.e. it was not served within one cycle.

(mtldp) $ process_trajs -c configs/my_region.json -s 2025-03-03 -e 2025-03-09 --process-mode signalized

The processed files are what AppRegion.load_trajs_dict and load_trajs_df read; the first load of a day is cached as a pickle next to the CSV files.

SPaT: parse_spat

Input: the seven CSV files of raw_spat/ prepared from the agency work orders (Preparation of Signal Phase and Timing (SPaT) data). Output: spat/spat.pickle, a HistoryRegionSPaT that AppRegion loads on construction. The command works on exactly the region it is given, so it is run once per sub-region that has SPaT data.

(mtldp) $ parse_spat -c configs/my_region_subs/0_1.json

Quality control: eva_trajs_qc

After process_trajs, eva_trajs_qc compares the daily volume of the processed trajectories between periods (for example the same week of two different months, or before and after a change of the map):

(mtldp) $ eva_trajs_qc -c configs/my_region_subs/0_1.json \
             -d 2025-03-03,2025-03-09@2025-06-02,2025-06-08

It plots, day by day, the number of trajectories and trips, the vehicle miles and hours travelled, the number of points and the average sampling interval of the signalized trajectories, and saves the figure in the region’s figures directory. A sudden drop in one panel points to a data delivery problem, a map problem (matching failures) or a processing regression.