.. _utils_reference: Utilities: ``mtldp.utils`` ========================== ``mtldp.utils`` (repository ``mtldp-meta-utils``) contains every function that manipulates the ``mtldp.meta`` data structures without touching raw vendor data: configuration and paths, file input and output, time and geographic helpers, filters, aggregation, plotting, database export and the command line tools. It is a namespace package: import the sub-modules directly, for example ``from mtldp.utils.data_io import load_traffic_network_from_pickle``. .. list-table:: :header-rows: 1 :widths: 24 76 * - Sub-module - Purpose * - ``config`` - ``Region``, ``ProdRegion``, ``AppRegion`` and the path managers (:ref:`regions`). * - ``scripts`` - The command line tools (:ref:`meta_utils_cli`). * - ``data_io`` - Load and save networks, trajectories and SPaT; download OpenStreetMap data. * - ``time`` - Conversions between timestamps, dates, clock times and time of day. * - ``geo_utils`` - Distances, headings, coordinate transforms and the movement-index convention. * - ``filter`` - Cut OSM files to a bounding box; select trajectories. * - ``aggregation`` - Aggregate trajectory indices per movement, junction, link, region or arterial. * - ``algs`` - Shortest paths on the link graph, time-of-day plan segmentation, reference distances. * - ``converter`` - Network and trajectory objects to ``DataFrame`` and back. * - ``interpolation`` - Resample trajectory points on a regular time grid. * - ``performance`` - Corridor travel time. * - ``visualizer`` - Time-space diagrams, metric plots, tables and the region map. * - ``database`` - MySQL table definitions and upload helpers. * - ``common`` - CLI decorator, logging and chunking helpers. ``data_io`` ----------- Traffic network ~~~~~~~~~~~~~~~ ``dump_traffic_network_to_pickle(network, pickle_filename)`` and ``load_traffic_network_from_pickle(pickle_filename)`` persist the whole ``Network`` object graph. The network is a deeply recursive structure, so the dump runs in a thread with a 64 MiB stack and the module raises the recursion limit on import. The loader returns ``None`` (and logs the traceback) when the file cannot be read. ``save_network_to_csv(network, folder)`` writes ``junctions.csv``, ``links.csv``, ``segments.csv``, ``movements.csv`` and ``arterials.csv`` using the converters described below. .. _network_geojson: Network geometry as GeoJSON ^^^^^^^^^^^^^^^^^^^^^^^^^^^ ``save_network_geometry_to_json(network, output_file=None, arterial_only=False, junction_id_list=None)`` writes the geometry of a network for web visualisation (the review tool and the signal portal consume it). The file is a JSON object with five keys, each a GeoJSON ``FeatureCollection``: .. list-table:: :header-rows: 1 :widths: 16 18 66 * - Layer - Geometry - Properties (every feature has ``id`` and ``kind``) * - ``junctions`` - ``Point`` - ``name``, ``node_type``, ``controller_id`` * - ``movements`` - ``MultiLineString`` - ``node_id``, ``movement_index``, ``turn`` * - ``links`` - ``LineString`` - ``from_direction``, ``heading`` * - ``segments`` - ``LineString`` - ``speed_limit`` (m/s), ``lane_num`` * - ``arterials`` - ``LineString`` per direction - ``arterial_id``, ``direction``; ``id`` is ``"|"`` Coordinates follow the GeoJSON order ``[lon, lat]`` with six decimals. ``arterial_only`` keeps the junctions touched by an arterial and their elements; ``junction_id_list`` keeps the listed junctions. The file is written with ``dumps_compact_coords``, which indents the structure but puts every ``coordinates`` array on one line to keep the files small. With ``output_file=None`` the dictionary is returned instead of written. Trajectories ~~~~~~~~~~~~ Raw vendor files are read by two loaders that normalise them to the same columns (``trip_id``, ``timestamp`` in seconds, ``latitude``, ``longitude``, ``speed`` in m/s, ``heading``, ``elevation`` in meters) and set ``traj_id = trip_id``: * ``load_GM_trajectory_from_CSV(filename_list, format_version)`` with ``format_version`` ``GM_2024_pre`` (columns ``ts``, ``vehiclevix``, ``subtaskdatalat``, ``subtaskdatalng``, ``subtaskdataspdinkm``, …, plus vehicle length and width) or ``GM_2025`` (columns ``t_epoch``, ``trip_id``, ``lat``, ``lon``, ``speed_kmph``, ``direction_deg``, ``elev``). Timestamps are converted from milliseconds and speeds from km/h; * ``load_SL_trajectory_from_parquet(filename_list, format_version)`` for the ``SL_2025`` Parquet delivery (``journey_id``, ``capture_time``, ``speed_mph``, ``heading_deg_north``, ``elevation_ft``). Rows whose ``fuzzed_point`` flag is not ``False`` are discarded. ``dump_points_to_CSV_by_date(points_df, filename_fn, append=True)`` groups the points by their ``date`` column and writes one CSV per date with the fixed column order ``trip_id``, ``traj_id``, ``timestamp``, ``longitude``, ``latitude``, ``elevation``, ``speed``, ``heading``, ``date``, ``date_time``, ``tod``; existing files are appended to. Processed trajectories are read by ``load_trajectories_from_csv(filename, junction_id_list, sort, drop_duplicate, load_row_num, load_points)`` (Polars reader, optional junction filter, optional skipping of the per-point list columns) and turned into objects by ``build_trajectory_dict_from_csv(trajectories_filename, junction_id_list, load_points, sort, deduplicate, load_row_num)``, which returns a ``TrajectoryDict``. ``load_realtime_backup_trajectory`` and ``preload_realtime_backup_trajectory`` read the compressed JSON-lines backups of the real-time feed, and ``denoise_trajectories(trajs_df, max_time, max_control_delay, max_distance)`` clips outliers. SPaT ~~~~ ``load_region_spat_from_csv(region, estimate=False)`` reads the seven CSV files of the region's ``raw_spat`` directory into a ``HistoryRegionSPaT``; ``dump_region_spat_to_pickle`` and ``load_region_spat_from_pickle(pickle_path, estimate=False)`` persist it (``estimate=True`` addresses the ``*_estimate.pickle`` twin). ``ProcessedDataLoader`` ~~~~~~~~~~~~~~~~~~~~~~~ ``ProcessedDataLoader(path_manager, timezone)`` is the loader behind ``AppRegion`` and ``ProdRegion``: ``load_network()``, ``load_spat(estimate)``, ``load_trajs_df(...)`` and ``load_trajs_dict(date_str_list, junction_id_list, load_points, overwrite_buffer, load_row_num, signalized)``. The last one parses each day's CSV once and caches the resulting ``TrajectoryDict`` as ``trajectories/trajectories/cache/_trajs.pkl`` (``_signalized_trajs.pkl`` for the signalized files); later calls read the pickle and re-apply the junction filter. Passing ``overwrite_buffer=True`` or a ``load_row_num`` (debug mode, never cached) forces a re-parse. ``OSMDownloader`` ~~~~~~~~~~~~~~~~~ ``OSMDownloader(bbox, query_template=None)`` downloads OpenStreetMap XML through the Overpass API (``overpass-api.de`` first, ``overpass.kumi.systems`` as fallback on HTTP 429 or 504). The default query selects all ``highway`` nodes, ways and relations except footways, paths, service roads, cycleways and similar non-vehicular features. ``get_osm_xml(filename_prefix, region_id, lat_step_size_threshold=0.1, lon_step_size_threshold=0.1, expand_meters=200)`` performs the tiling described in :ref:`sub_regions` and writes ``_bbox_dict.json``. Already downloaded tiles are skipped, so an interrupted download can be resumed. ``time`` -------- Four representations of time are used throughout MTLDP: .. list-table:: :header-rows: 1 :widths: 22 78 * - Name - Meaning * - ``timestamp`` - Unix epoch time in **seconds** (float). * - ``date`` - Local date string ``YYYY-MM-DD`` (``DATE_FORMAT``). * - ``date_time`` - Local clock time string ``HH:MM`` (``HHMM_FORMAT``); parsers also accept ``HH:MM:SS``. * - ``tod`` - Time of day as **hours since local midnight**, e.g. ``7.5`` for 07:30; valid in ``[0, 24)``. The default time zone of the conversion functions is UTC (``DEFAULT_TIMEZONE``); the region's time zone must be passed explicitly (``DEFAULT_LOCAL_TIMEZONE`` is ``America/Detroit``). * Timestamp conversions: ``timestamp_to_date_time_and_tod(timestamp, ..., timezone_name)`` returns ``(date, date_time, tod)``; ``timestamp_to_hhmm``, ``get_seconds_in_day_from_timestamp``, ``get_floor_timestamp(timestamp, floor_minute, timezone_name)`` and ``timestamp_to_time_in_cycle(ts, tod_start, cycle, offset, timezone_name)`` (position of a timestamp inside a signal cycle). The inverse is ``get_timestamp_from_date_tod(date, tod, timezone_name)``. * Clock and ``tod`` conversions: ``date_time_to_tod``, ``tod_to_date_time``, ``date_time_to_seconds_in_day``, ``seconds_in_day_to_tod``, ``seconds_in_day_to_date_time``, ``get_seconds_between_tods``, ``get_seconds_between_date_time``, and ``tod_range_to_str(tod_range, sep='-')`` (``[7, 9]`` becomes ``0700-0900``, used in file names). * Ranges: ``get_date_range_str_list(start_date, end_date=None, date_interval=1, weekdays=True, weekends=True)`` lists the dates between two bounds (inclusive); ``weekdays=False`` or ``weekends=False`` drops the corresponding days, which is what the ``--exclude-weekend`` and ``--exclude-weekday`` options of the pipeline commands do. ``get_date_time_range(start, end, minute_interval)`` lists clock strings; ``get_date_name(date_list)`` yields ``_``. * ``DataFrame`` helpers: ``df_add_date_time_and_tod(df, attr='timestamp', timezone_name=...)`` adds the ``date``, ``date_time`` and ``tod`` columns from a ``timestamp`` column (used by ``split_points``); ``df_add_date``, ``df_add_tod``, ``df_add_date_time`` and ``df_add_seconds_in_day`` add one column each. ``geo_utils`` ------------- * Distances (``distance``): ``haversine_distance((lat, lon), (lat, lon))`` in meters, ``get_traveled_distance(lat_list, lon_list)`` (cumulative distance along a trace, starting at 0), ``get_dis_to_point``, ``destination_point(lat, lon, distance, bearing)`` and ``segment_gps_trace(lat_list, lon_list, segment_length=50)``. * Headings (``heading``): headings are degrees in ``(-180, 180]``. ``get_heading_direction`` bins a heading into the cardinal *from* direction, ``get_gps_trace_heading_info`` returns the forward and backward headings of a trace and their length-weighted versions, ``weighted_circular_mean_degree``, ``normalize_degree`` and ``reverse_degree``. * Transforms (``transform``): ``gps_to_xy_meters(lat, lon, ref_lat, ref_lon)`` and ``xy_meters_to_gps`` convert between GPS and a local metric frame anchored at a reference point (``Region.convert_GPS_to_xy`` uses the south-west corner of the bounding box). * Geometry (``advanced``): ``shift_geometry(geometry, shift_distance=7, shift_direction='left')`` offsets a polyline sideways (display geometry), ``get_shifted_gps_trace`` moves a trace along a bearing. * Movement convention (``constants``): ``signal_movement_dict`` defines the 16 movement indices (1 to 8 lefts and throughs, 9 to 12 rights, 13 to 16 U-turns), ``get_movement_id(geo_dir, turn_dir)``, ``get_turning_direction(upstream_from_dir, downstream_from_dir)``, ``get_turning_direction_from_headings(from_heading, to_heading)`` (straight within ±45°, left, right, U-turn beyond ±135°), ``movement_conflicting_dict``, ``shared_approach_movement_dict``, ``same_approach_through_movement_dict``, ``get_movement_name_by_movement_index`` and ``calc_r2`` (straightness of a polyline). The buffers ``MAP_BBOX_EXTEND_METER`` (200 m, map download and OSM cutting) and ``GPS_POINTS_BBOX_EXTEND_METER`` (100 m) and the plot ordering ``DIRECTION_ORDER`` (``S N E W``) live here too. ``filter`` ---------- ``filter.osm.filter_osm_by_bbox(input_file, output_file, bbox)`` cuts an OSM file to a bounding box with *complete ways* semantics (the behaviour of ``osmium extract -s complete_ways``): a way is kept as soon as one of its nodes is inside the box, and all nodes it references are written even when they lie outside. This keeps the road graph connected across the border, which matters for map matching. The input does not need to be sorted; it is read in three passes with ``pyosmium``. ``create_sub_regions``, ``copy_regions`` and ``extract_osm_map`` use it. ``filter.trajectory`` (re-exported as ``mtldp.utils.filter``) selects subsets of a ``TrajectoryDict`` and always returns a new dictionary: * by attribute: ``get_tod_trajs_dict(trajs_dict, (start, end))``, ``get_movements_trajs_dict``, ``get_junctions_trajs_dict``, ``get_link_trajs_dict``, ``filter_trajs_dict_based_on_traj_ids``, ``direction_filter(region, trajs_dict, direction_list=['s', 'l'])``; * by size and stops: ``length_filter`` (at least one point), ``distance_filter(trajs_dict, distance=50)``, ``max_stop_filter``, ``exact_stop_filter``, ``min_stop_filter``; * by quality: ``error_filter(trajs_dict, region)`` drops trajectories whose average matching error exceeds ten meters per lane of the approach, ``large_frequency_filter`` drops trajectories with implausible jumps (over 50 m/s or gaps above 20 s), and ``filter_trajs_dict_based_on_experience(trajs_dict, tod_interval)`` applies a set of practical thresholds and returns ``(kept, outliers)``. The ``filter_trajs_dict_based_on_*`` names are aliases of the corresponding ``get_*`` functions. ``aggregation`` --------------- ``get_aggregated_df(trajs_df, layer='movement', resolution=2, timezone='America/Detroit', aggregate_date=False)`` aggregates a table of trajectories into time slices of ``resolution`` minutes at the ``movement``, ``node`` (junction), ``link`` or ``all`` (region) level. Each slice reports the trajectory count (``traj_num``), the counts of non-stopping and once-stopping trajectories, and the mean or sum of ``control_delay``, ``stop_delay``, ``stop_nums``, ``free_v``, ``queue_dis``, ``queue_ratio``, ``cycle_failure``, ``spill_over``, travel time and travel distance. ``aggregate_date=True`` folds several days into one time-of-day profile. ``get_aggregated_arterial_df(df, arterial_list, tod_interval, fill_all_time=True)`` produces one row per arterial direction and time slot (``tod_interval`` in hours) with totals of trajectories, travel time and distance, delay, stops, spill-overs and split failures, plus ``through_`` variants when the table has a ``through`` column. ``algs`` -------- * ``build_link_networkx_graph(network, allow_backward=True)`` builds the directed graph whose nodes are links and whose edges are movements (sink movements are skipped, U-turns optionally) and stores it in ``network.networkx_graph``. * ``shortest_path_between_links(network, upstream_link_id, downstream_link_id, weight_attrib='length')`` runs Dijkstra on that graph and returns ``(link_id_list, weight)``. It is what ``NetworkLinkPath`` and the arterial builder use to fill gaps between links. * ``tod_opt(features, n_cut, n_min_steps=1)`` splits equal-length time series into ``n_cut`` homogeneous periods by dynamic programming (time-of-day plan design). * ``calc_ref_dist.reference_distance(...)`` rebases a cumulative distance so that zero is at a reference point such as a stop bar, and ``veh_len_calibration`` corrects for vehicle length. ``converter`` ------------- * ``list_str.list2str(values, con_by='|')`` and ``common.list_to_str(values, connector=' ')`` are the two delimiters used in CSV columns: pipe for trajectory list columns, space for network element lists. * ``trajectory.df_to_trajectory_dict(trajs_df, load_points=True)`` turns a trajectory table into a ``TrajectoryDict`` (list columns are split on ``|``, ``stop_details`` is decoded with ``parse_stop_details``), and ``trajs_dict_to_trips`` groups a ``TrajectoryDict`` into ``Trip`` objects. * ``traffic_network.convert_junction_to_df``, ``convert_link_to_df``, ``convert_segment_to_df``, ``convert_movement_to_df`` and ``convert_corridor_to_df`` export one network layer each as a ``DataFrame`` (the columns of the ``network/csv`` files and of the database tables). ``interpolation`` ----------------- ``interpolate_points(traj, attributes, resolution, degree=1)`` resamples the selected point attributes of a trajectory on a regular time grid, marks generated rows with an ``interpolated`` column and fills the remaining columns from the nearest original row on the same side of the stop bar. ``performance`` --------------- ``arterial_travel_time.get_corridor_travel_times(trip_dict, movement_list, distance, upstream_dis=50)`` measures the end-to-end travel time of the trips that traverse a complete list of corridor movements, trims outliers beyond two standard deviations and returns ``(mean, count, std)``. ``visualizer`` -------------- * ``Plotter(output_dir, fig_size, fig_format, dpi)`` wraps Matplotlib with the primitives used by all plots: ``create_plot``, ``save_figure``, time-space lines (``plot_ts``, ``plot_no_agg_ts``), signal bars (``plot_signal_bar``, ``plot_signal_lines``), histograms, stacked bars and cycle markers. ``TableDrawer(output_dir)`` writes tables to Excel sheets or CSV. * ``convert`` turns trajectories and SPaT into plot coordinates: ``movement_trajs_dict_to_xy_lists`` (optionally folding every trajectory into one signal cycle), ``movement_spat_to_signal_bar`` (green, yellow, clearance, red and early-green intervals with the colours in ``SIGNAL_TO_COLOR``), ``trajs_dict_to_stop_cat_color``, arrival and delay histograms and the corridor helpers ``extract_y_location_list`` and ``corridor_trajs_dict_to_movement_lists``. * ``trajectory`` draws time-space diagrams: ``plot_no_agg_movement_ts`` (real clock), ``plot_movement_ts`` and ``plot_movement_ts_with_category`` (folded into the cycle, optionally coloured by number of stops) and ``plot_bidirectional_corridor_ts`` (both directions of an arterial with the signal bars of every junction). * ``metric`` draws and tabulates indices: ``plot_movement_control_delay_scatter``, ``plot_movement_control_delay_hist``, ``plot_movement_arrival_hist``, ``plot_movement_trajs_num_with_category`` (average daily observations by time of day and stop category), ``write_junction_metrics_to_table`` and ``get_arterial_metrics_to_table``. * ``map.plot_region_bbox(master_config_file_path)`` draws the sub-region bounding boxes on a Mapbox basemap and writes ``region_bounding_box.png`` and ``.html`` into the region's ``figures`` directory (the ``draw_region_bbox`` command). ``database`` ------------ ``MySQLTable(table_name, column_list, unique_keys, unique_key_name)`` wraps an SQLAlchemy table definition (InnoDB, ``utf8mb4``). ``traffic_network`` defines the ``junctions``, ``links``, ``segments``, ``movements``, ``arterials``, ``lanesets`` and ``arterial_movement`` tables and ``trajectory_and_index`` the movement, junction and arterial index tables, the per-trajectory table and the matched points table. ``upload.get_mysql_engine(config_path)`` builds an engine from a JSON credentials file, ``upload_network_to_mysql(network, engine, version)`` uploads the network layers, and ``save_df_to_db`` uploads a large ``DataFrame`` in chunks. SQLAlchemy is an optional dependency: the modules import without it but cannot connect. ``common`` ---------- * ``typer_run`` turns a function with typed parameters into a Typer command line entry point. * ``split_trip_df_to_chunks(df, num_chunks)`` splits a point table into chunks on trip boundaries, so that parallel workers never see half a trip. * ``add_filelog_handler(log_dir, log_name)`` adds a daily rotating file handler to the root logger (``build_traffic_network`` writes ``network/map.log`` with it). * ``show_attributes_and_methods(instance)`` prints the attributes and methods of an object; the example notebooks use it to explore the data structures.