Lane-level matching data

1
root
2
 |-- id: string
3
 |-- points: array
4
 |    |-- element: struct
5
 |    |    |-- coord: struct
6
 |    |    |    |-- x: double
7
 |    |    |    |-- y: double
8
 |    |    |-- heading: double
9
 |    |    |-- lane: long
10
 |    |    |-- timestamp: long
11
 |    |    |-- velocity: double

1
{
2
  "id": "trace-001",
3
  "points": [
4
    {"coord": {"x": 4.8952, "y": 52.3702}, "heading": 34.2, "lane": null, "timestamp": 1760000000000, "velocity": 13.9},
5
    {"coord": {"x": 4.8960, "y": 52.3710}, "heading": 34.5, "lane": null, "timestamp": 1760000001000, "velocity": 14.1}
6
  ]
7
}

1
uuid,shape
2
trace-001,"LINESTRING (4.8952 52.3702, 4.8960 52.3710, 4.8975 52.3725)"
3
trace-002,"LINESTRING (4.8990 52.3740, 4.9005 52.3755, 4.9020 52.3770)"

1
uuid,heading,coordinate
2
event-001,92.5,POINT (4.8952 52.3702)
3
event-002,180.0,POINT (4.8990 52.3740)

1
import pyarrow as pa
2
import pyarrow.parquet as pq
3

4
coord = pa.struct([pa.field("x", pa.float64()), pa.field("y", pa.float64())])
5
point = pa.struct([
6
    pa.field("coord", coord),
7
    pa.field("heading", pa.float64()),
8
    pa.field("lane", pa.int64()),
9
    pa.field("timestamp", pa.int64()),
10
    pa.field("velocity", pa.float64()),
11
])
12
schema = pa.schema([pa.field("id", pa.string()), pa.field("points", pa.list_(point))])
13

14
traces = [
15
    {
16
        "id": "trace-001",
17
        "points": [
18
            {"coord": {"x": 4.8952, "y": 52.3702}, "heading": 34.2, "lane": None, "timestamp": 1760000000000, "velocity": 13.9},
19
            {"coord": {"x": 4.8960, "y": 52.3710}, "heading": 34.5, "lane": None, "timestamp": 1760000001000, "velocity": 14.1},
20
        ],
21
    },
22
]
23
pq.write_table(pa.Table.from_pylist(traces, schema=schema), "traces.parquet")

1
from pyspark.sql import SparkSession
2
from pyspark.sql.types import ArrayType, DoubleType, LongType, StringType, StructField, StructType
3

4
spark = SparkSession.builder.appName("gem-trace-input").getOrCreate()
5

6
coord = StructType([StructField("x", DoubleType(), False), StructField("y", DoubleType(), False)])
7
point = StructType([
8
    StructField("coord", coord, True),
9
    StructField("heading", DoubleType(), False),
10
    StructField("lane", LongType(), True),
11
    StructField("timestamp", LongType(), False),
12
    StructField("velocity", DoubleType(), False),
13
])
14
schema = StructType([StructField("id", StringType(), True), StructField("points", ArrayType(point, True), True)])
15

16
rows = [
17
    ("trace-001", [
18
        {"coord": {"x": 4.8952, "y": 52.3702}, "heading": 34.2, "lane": None, "timestamp": 1760000000000, "velocity": 13.9},
19
        {"coord": {"x": 4.8960, "y": 52.3710}, "heading": 34.5, "lane": None, "timestamp": 1760000001000, "velocity": 14.1},
20
    ]),
21
]
22
spark.createDataFrame(rows, schema).coalesce(1).write.mode("overwrite").parquet("traces_spark")

1
az storage blob upload --account-name "<STORAGE_ACCOUNT_NAME>" \
2
                       --container-name "default" \
3
                       --name "traces.parquet" \
4
                       --file "$(ls traces_spark/part-*.parquet)" \
5
                       --auth-mode login

1
import csv
2

3
profiles = [
4
    ("trace-001", "LINESTRING (4.8952 52.3702, 4.8960 52.3710, 4.8975 52.3725)"),
5
    ("trace-002", "LINESTRING (4.8990 52.3740, 4.9005 52.3755, 4.9020 52.3770)"),
6
]
7
with open("road_profile.csv", "w", newline="") as file:
8
    writer = csv.writer(file)
9
    writer.writerow(["uuid", "shape"])
10
    writer.writerows(profiles)
11

12
events = [
13
    ("event-001", 92.5, "POINT (4.8952 52.3702)"),
14
    ("event-002", 180.0, "POINT (4.8990 52.3740)"),
15
]
16
with open("road_events.csv", "w", newline="") as file:
17
    writer = csv.writer(file)
18
    writer.writerow(["uuid", "heading", "coordinate"])
19
    writer.writerows(events)

1
import geopandas as gpd
2
import pandas as pd
3

4
drives = gpd.read_file("drives.gpkg").to_crs("EPSG:4326").explode(index_parts=False)
5
drives = drives[drives.geom_type == "LineString"].reset_index(drop=True)
6

7
part = drives.groupby("drive_id").cumcount()
8
drive_id = drives["drive_id"].astype(str)
9

10
pd.DataFrame({
11
    "uuid": drive_id.where(part == 0, drive_id + "-" + part.astype(str)),
12
    "shape": drives.geometry.to_wkt(),
13
}).to_csv("road_profile.csv", index=False)

1
pip install pyarrow shapely
2
python validate_traces_parquet.py traces.parquet
3
python validate_traces_csv.py road_profile.csv

1
"""Check a lane-level matching trace file (Parquet) before uploading it to GEM.
2

3
Usage: python validate_traces_parquet.py <file.parquet>
4
"""
5
import sys
6
from pathlib import Path
7

8
import pyarrow as pa
9
import pyarrow.compute as pc
10
import pyarrow.parquet as pq
11

12
POINT_TYPES = {"heading": pa.float64(), "lane": pa.int64(), "timestamp": pa.int64(), "velocity": pa.float64()}
13
REQUIRED_VALUES = ["heading", "timestamp", "velocity"]
14

15

16
def by_name(fields):
17
    return {field.name.lower(): field for field in fields}
18

19

20
def child(structs, field):
21
    return pc.struct_field(structs, [structs.type.get_field_index(field.name)])
22

23

24
def count(mask):
25
    return pc.sum(mask).as_py() or 0
26

27

28
def schema_errors(schema):
29
    columns = by_name(schema)
30
    errors = []
31
    if "id" not in columns:
32
        errors.append("Column 'id' is missing.")
33
    elif columns["id"].type not in (pa.string(), pa.large_string()):
34
        errors.append(f"Column 'id' must be a string, not '{columns['id'].type}'.")
35
    points = columns.get("points")
36
    if points is None:
37
        return errors + ["Column 'points' is missing. Store one trace per row, with its points in a list."]
38
    is_list = pa.types.is_list(points.type) or pa.types.is_large_list(points.type)
39
    if not is_list or not pa.types.is_struct(points.type.value_type):
40
        return errors + [f"Column 'points' must be a list of structs, not '{points.type}'."]
41
    fields = by_name(points.type.value_type)
42
    coord = fields.get("coord")
43
    if coord is None or not pa.types.is_struct(coord.type):
44
        errors.append("Field 'points.coord' must be a struct with the fields 'x' and 'y'.")
45
    else:
46
        axes = by_name(coord.type)
47
        for axis in ("x", "y"):
48
            if axis not in axes or axes[axis].type != pa.float64():
49
                errors.append(f"Field 'points.coord.{axis}' must be a double.")
50
    for name, expected in POINT_TYPES.items():
51
        if name not in fields:
52
            errors.append(f"Field 'points.{name}' is missing.")
53
        elif fields[name].type != expected:
54
            errors.append(f"Field 'points.{name}' must be '{expected}', not '{fields[name].type}'.")
55
    return errors
56

57

58
def data_findings(path, schema):
59
    columns = by_name(schema)
60
    fields = by_name(columns["points"].type.value_type)
61
    axes = by_name(fields["coord"].type)
62
    no_id = no_points = empty_points = no_coordinates = duplicates = outside = bad_heading = 0
63
    missing = dict.fromkeys(REQUIRED_VALUES, 0)
64
    seen = set()
65
    for batch in pq.ParquetFile(path).iter_batches(columns=[columns["id"].name, columns["points"].name]):
66
        ids, points = batch.column(columns["id"].name), batch.column(columns["points"].name)
67
        no_id += count(pc.fill_null(pc.equal(ids, ""), True))
68
        no_points += count(pc.fill_null(pc.equal(pc.list_value_length(points), 0), True))
69
        for trace_id in ids.drop_null().to_pylist():
70
            duplicates += trace_id in seen
71
            seen.add(trace_id)
72

73
        flat = pc.list_flatten(points)
74
        empty_points += flat.null_count
75
        flat = flat.filter(pc.is_valid(flat))
76
        for name in REQUIRED_VALUES:
77
            missing[name] += child(flat, fields[name]).null_count
78
        coord = child(flat, fields["coord"])
79
        x, y = child(coord, axes["x"]), child(coord, axes["y"])
80
        no_coordinates += count(pc.or_(pc.is_null(x), pc.is_null(y)))
81
        outside += count(pc.or_(pc.greater(pc.abs(x), 180), pc.greater(pc.abs(y), 90)))
82
        heading = child(flat, fields["heading"])
83
        bad_heading += count(pc.or_(pc.less(heading, 0), pc.greater(heading, 360)))
84

85
    errors, warnings = [], []
86
    if no_id:
87
        errors.append(f"{no_id} trace(s) have no 'id'.")
88
    if no_points:
89
        errors.append(f"{no_points} trace(s) have no points.")
90
    if empty_points:
91
        errors.append(f"{empty_points} point(s) are empty.")
92
    if no_coordinates:
93
        errors.append(f"{no_coordinates} point(s) have no coordinates.")
94
    errors += [f"{missing_count} point(s) have no '{name}' value." for name, missing_count in missing.items() if missing_count]
95
    if duplicates:
96
        warnings.append(f"{duplicates} trace(s) reuse the 'id' of an earlier trace. "
97
                        "Use unique IDs to join the results back to your data.")
98
    if outside:
99
        warnings.append(f"{outside} point(s) have coordinates outside the longitude/latitude range. "
100
                        "Use WGS84 with x = longitude and y = latitude.")
101
    if bad_heading:
102
        warnings.append(f"{bad_heading} point(s) have a heading outside 0-360 degrees.")
103
    return errors, warnings
104

105

106
def main(file_name):
107
    path = Path(file_name)
108
    if not path.is_file():
109
        print(f"ERROR: {path} does not exist.")
110
        return 1
111
    try:
112
        schema = pq.read_schema(path)
113
    except pa.ArrowInvalid:
114
        print(f"ERROR: {path.name} is not a valid Parquet file.")
115
        return 1
116

117
    errors, warnings = schema_errors(schema), []
118
    if not errors:
119
        errors, warnings = data_findings(path, schema)
120

121
    for message in errors:
122
        print(f"ERROR: {message}")
123
    for message in warnings:
124
        print(f"WARNING: {message}")
125
    if not errors:
126
        print(f"OK: {path.name} is ready for lane-level matching.")
127
    return 1 if errors else 0
128

129

130
if __name__ == "__main__":
131
    if len(sys.argv) != 2:
132
        sys.exit(__doc__)
133
    sys.exit(main(sys.argv[1]))

1
"""Check a lane-level matching trace file (CSV) before uploading it to GEM.
2

3
Usage: python validate_traces_csv.py <file.csv>
4
"""
5
import csv
6
import sys
7
from pathlib import Path
8

9
import shapely
10

11
LAYOUTS = {
12
    "road profile": ("uuid", "shape"),
13
    "road events": ("uuid", "heading", "coordinate"),
14
}
15
GEOMETRIES = {"road profile": ("shape", "LineString"), "road events": ("coordinate", "Point")}
16
MAX_LISTED = 20
17

18

19
def detect_layout(header):
20
    for layout, columns in LAYOUTS.items():
21
        if set(columns) <= set(header):
22
            return layout
23
    return None
24

25

26
def header_error(header):
27
    expected = {column for columns in LAYOUTS.values() for column in columns}
28
    misspelled = [name for name in header if name not in expected and name.strip().lower() in expected]
29
    hint = f" Fix {misspelled}: header names are case-sensitive and can't contain spaces." if misspelled else ""
30
    return (f"Unrecognized header {header}. Use 'uuid,shape' for a road profile "
31
            f"or 'uuid,heading,coordinate' for road events.{hint}")
32

33

34
def row_findings(reader, layout):
35
    errors, seen = [], set()
36
    duplicates = outside = bad_heading = 0
37
    column, geometry_type = GEOMETRIES[layout]
38
    for line, row in enumerate(reader, start=2):
39
        trace_id = (row.get("uuid") or "").strip()
40
        if not trace_id:
41
            errors.append(f"Line {line}: 'uuid' is empty.")
42
        elif trace_id in seen:
43
            duplicates += 1
44
        seen.add(trace_id)
45

46
        geometry = shapely.from_wkt(row.get(column) or "", on_invalid="ignore")
47
        if geometry is None or geometry.geom_type != geometry_type or geometry.is_empty:
48
            errors.append(f"Line {line}: '{column}' is not a valid WKT {geometry_type.upper()}.")
49
        else:
50
            x_min, y_min, x_max, y_max = geometry.bounds
51
            outside += max(abs(x_min), abs(x_max)) > 180 or max(abs(y_min), abs(y_max)) > 90
52

53
        if layout == "road events":
54
            try:
55
                bad_heading += not 0 <= float(row.get("heading") or "") <= 360
56
            except ValueError:
57
                errors.append(f"Line {line}: 'heading' is not a number.")
58

59
    warnings = []
60
    if duplicates:
61
        warnings.append(f"{duplicates} row(s) reuse the 'uuid' of an earlier row. "
62
                        "Use unique IDs to join the results back to your data.")
63
    if outside:
64
        warnings.append(f"{outside} row(s) have coordinates outside the longitude/latitude range. "
65
                        "Use WGS84 with longitude first.")
66
    if bad_heading:
67
        warnings.append(f"{bad_heading} row(s) have a heading outside 0-360 degrees.")
68
    return errors, warnings
69

70

71
def main(file_name):
72
    path = Path(file_name)
73
    if not path.is_file():
74
        print(f"ERROR: {path} does not exist.")
75
        return 1
76
    with path.open(newline="", encoding="utf-8-sig") as file:
77
        if "," not in file.readline():
78
            print("ERROR: The header has no commas. Separate the values with commas.")
79
            return 1
80
        file.seek(0)
81
        reader = csv.DictReader(file)
82
        header = reader.fieldnames or []
83
        layout = detect_layout(header)
84
        if layout is None:
85
            print(f"ERROR: {header_error(header)}")
86
            return 1
87
        errors, warnings = row_findings(reader, layout)
88

89
    if layout == "road profile" and set(LAYOUTS["road events"]) <= set(header):
90
        warnings.append("The header has the columns of both layouts, so GEM reads the file as a road profile.")
91

92
    for message in errors[:MAX_LISTED]:
93
        print(f"ERROR: {message}")
94
    if len(errors) > MAX_LISTED:
95
        print(f"ERROR: ... and {len(errors) - MAX_LISTED} more.")
96
    for message in warnings:
97
        print(f"WARNING: {message}")
98
    if not errors:
99
        print(f"OK: {path.name} is ready for lane-level matching ({layout}).")
100
    return 1 if errors else 0
101

102

103
if __name__ == "__main__":
104
    if len(sys.argv) != 2:
105
        sys.exit(__doc__)
106
    sys.exit(main(sys.argv[1]))

1
root
2
 |-- roads: array (nullable = true)
3
 |    |-- element: struct (containsNull = true)
4
 |    |    |-- endConnectorId: string (nullable = true)
5
 |    |    |-- lane: struct (nullable = true)
6
 |    |    |    |-- confidence: double (nullable = false)
7
 |    |    |    |-- endOffset: double (nullable = true)
8
 |    |    |    |-- laneId: string (nullable = true)
9
 |    |    |    |-- startOffset: double (nullable = true)
10
 |    |    |    |-- wkt: string (nullable = true)
11
 |    |    |-- roadAreaId: string (nullable = true)
12
 |    |    |-- startConnectorId: string (nullable = true)
13
 |-- traceId: string (nullable = true)
14
 |-- mapRevisionId: long (nullable = false)

1
{
2
  "traceId": "trace-001",
3
  "mapRevisionId": 2760154,
4
  "roads": [
5
    {
6
      "roadAreaId": "12345:678:90",
7
      "startConnectorId": "12345:678:11",
8
      "endConnectorId": "12345:678:12",
9
      "lane": {
10
        "laneId": "12345:678:42",
11
        "wkt": "LINESTRING (4.8952 52.3702, 4.8960 52.3710)",
12
        "confidence": 100.0,
13
        "startOffset": 0.0,
14
        "endOffset": 37.5
15
      }
16
    }
17
  ]
18
}