.. _pipeline: 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: .. code-block:: text 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 (:ref:`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. .. figure:: pipeline.svg :width: 100% :alt: Data flow of the pre-processing pipeline Data flow of the pipeline. Solid arrows are the data produced by each stage, dashed arrows the network files a stage reads. .. _pipeline_build_network: 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 :ref:`preproc_reference`. 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: .. list-table:: :header-rows: 1 :widths: 38 62 * - 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 (:ref:`network_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``. .. code-block:: bash (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 .. _pipeline_split_points: 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/_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. .. code-block:: bash (mtldp) $ split_points -c configs/my_region.json -v SL_2025 -d /data/deliveries/2025-03 .. _pipeline_map_match: Stage 3: ``match_points_to_map`` -------------------------------- **Input**: ``trajectories/raw_pts/_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: #. the raw points are sorted by ``trip_id`` and ``timestamp`` and de-duplicated; #. points outside the sub-region bounding box are dropped; #. 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; #. the segments are resolved against the network, adding ``link_id``, ``upstream_intersection``, ``downstream_intersection`` and ``junction_id`` (the downstream intersection); #. numeric columns are rounded and the table is saved. **Output**: ``trajectories/points/_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``. .. code-block:: bash (mtldp) $ match_points_to_map -c configs/my_region.json -s 2025-03-03 -e 2025-03-09 .. _pipeline_process_trajs: Stage 4: ``process_trajs`` -------------------------- **Input**: ``trajectories/points/_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``: #. points without a matched link and points whose matching error exceeds ``lane number × max-away-per-lane + 15`` meters are dropped; #. 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; #. 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; #. 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/_trajs.csv`` (``all`` mode) or ``_signalized_trajs.csv`` (``signalized`` mode), one row per trajectory. The columns are the attributes exported by ``mtldp.meta.Trajectory``: .. list-table:: :header-rows: 1 :widths: 26 74 * - Column - Definition * - ``trip_id``, ``traj_id`` - The trip and the trajectory (``_``, ``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 ``_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. .. code-block:: bash (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 (:ref:`spat_prepare`). **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. .. code-block:: bash (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): .. code-block:: bash (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.