Data preparation guide

LINESTRING (longitude1 latitude1, longitude2 latitude2, ...)

LINESTRING (145.18156 -37.87340, 145.18092 -37.87356)
LINESTRING (4.8952 52.3702, 4.8960 52.3710, 4.8975 52.3725)

1import pandas as pd
2import pyarrow as pa
3import pyarrow.parquet as pq
4
5# Create sample data
6data = {
7    'id': [1, 2, 3, 4, 5],
8    'is_navigable': [True, True, False, True, True],
9    'geometry': [
10        'LINESTRING (4.8952 52.3702, 4.8960 52.3710)',
11        'LINESTRING (4.8960 52.3710, 4.8975 52.3725)',
12        'LINESTRING (4.8975 52.3725, 4.8990 52.3740)',
13        'LINESTRING (4.8990 52.3740, 4.9005 52.3755)',
14        'LINESTRING (4.9005 52.3755, 4.9020 52.3770)'
15    ]
16}
17
18# Create DataFrame
19df = pd.DataFrame(data)
20
21# Define schema with correct types
22schema = pa.schema([
23    ('id', pa.int64()),
24    ('is_navigable', pa.bool_()),
25    ('geometry', pa.string())
26])
27
28# Convert to PyArrow Table with schema
29table = pa.Table.from_pandas(df, schema=schema)
30
31# Write to Parquet
32pq.write_table(table, 'my_road_data.parquet')
33
34print(f"Created Parquet file with {len(df)} records")

1import geopandas as gpd
2import pandas as pd
3
4# Read source data
5gdf = gpd.read_file('roads.shp')
6
7# Prepare for GEM
8gem_data = pd.DataFrame({
9    'id': range(1, len(gdf) + 1),  # Generate unique IDs
10    'is_navigable': gdf['navigable'].fillna(True),  # Default to True
11    'geometry': gdf.geometry.apply(lambda g: g.wkt)  # Convert to WKT
12})
13
14# Filter to LineStrings only
15gem_data = gem_data[gem_data['geometry'].str.startswith('LINESTRING')]
16
17# Save as Parquet
18gem_data.to_parquet('gem_input.parquet', index=False)
19
20print(f"Exported {len(gem_data)} road segments")

1from pyspark.sql import SparkSession
2from pyspark.sql.types import StructType, StructField, LongType, BooleanType, StringType
3
4# Initialize Spark
5spark = SparkSession.builder.appName("GEM Data Prep").getOrCreate()
6
7# Define schema
8schema = StructType([
9    StructField("id", LongType(), False),
10    StructField("is_navigable", BooleanType(), False),
11    StructField("geometry", StringType(), False)
12])
13
14# Read your source data
15source_df = spark.read.format("your_format").load("your_data")
16
17# Transform to GEM schema
18gem_df = source_df.select(
19    source_df["road_id"].alias("id"),
20    source_df["navigable"].alias("is_navigable"),
21    source_df["wkt_geometry"].alias("geometry")
22)
23
24# Write as Parquet
25gem_df.write.parquet("gem_input.parquet")

1import pandas as pd
2import re
3
4def validate_gem_data(filepath):
5    """Validate a Parquet file for GEM compatibility."""
6    
7    print(f"Validating: {filepath}")
8    errors = []
9    warnings = []
10    
11    # Read the file
12    try:
13        df = pd.read_parquet(filepath)
14    except Exception as e:
15        return [f"Cannot read file: {e}"], []
16    
17    print(f"Total records: {len(df)}")
18    
19    # Check required columns
20    required_cols = ['id', 'is_navigable', 'geometry']
21    missing_cols = [c for c in required_cols if c not in df.columns]
22    if missing_cols:
23        errors.append(f"Missing required columns: {missing_cols}")
24        return errors, warnings
25    
26    # Check for null values
27    for col in required_cols:
28        null_count = df[col].isnull().sum()
29        if null_count > 0:
30            errors.append(f"Column '{col}' has {null_count} null values")
31    
32    # Check ID uniqueness
33    duplicate_ids = df['id'].duplicated().sum()
34    if duplicate_ids > 0:
35        errors.append(f"Found {duplicate_ids} duplicate IDs")
36    
37    # Check data types
38    if not pd.api.types.is_integer_dtype(df['id']):
39        errors.append(f"Column 'id' should be integer, got {df['id'].dtype}")
40    
41    if not pd.api.types.is_bool_dtype(df['is_navigable']):
42        errors.append(f"Column 'is_navigable' should be boolean, got {df['is_navigable'].dtype}")
43    
44    # Validate geometries
45    linestring_pattern = r'^LINESTRING\s*\([^)]+\)$'
46    invalid_geom = 0
47    for idx, geom in df['geometry'].items():
48        if not isinstance(geom, str):
49            invalid_geom += 1
50        elif not re.match(linestring_pattern, geom.strip(), re.IGNORECASE):
51            invalid_geom += 1
52    
53    if invalid_geom > 0:
54        errors.append(f"Found {invalid_geom} invalid geometries (must be WKT LINESTRING)")
55    
56    # Check for empty geometries
57    empty_geom = df['geometry'].str.contains(r'LINESTRING\s*\(\s*\)', case=False, regex=True).sum()
58    if empty_geom > 0:
59        warnings.append(f"Found {empty_geom} empty geometries")
60    
61    # Summary
62    print(f"\nValidation Results:")
63    print(f"  Errors: {len(errors)}")
64    print(f"  Warnings: {len(warnings)}")
65    
66    if errors:
67        print("\nErrors:")
68        for e in errors:
69            print(f"  ❌ {e}")
70    
71    if warnings:
72        print("\nWarnings:")
73        for w in warnings:
74            print(f"  ⚠️ {w}")
75    
76    if not errors:
77        print("\n✅ File is valid for GEM!")
78    
79    return errors, warnings
80
81# Usage
82errors, warnings = validate_gem_data('my_road_data.parquet')

1from shapely import wkt
2from shapely.geometry import LineString
3
4def fix_geometry(geom):
5    """Convert various geometry formats to WKT LineString."""
6    try:
7        # If it's already a valid WKT string
8        parsed = wkt.loads(geom)
9        if isinstance(parsed, LineString):
10            return geom
11        else:
12            return None  # Not a LineString
13    except:
14        return None
15
16df['geometry'] = df['geometry'].apply(fix_geometry)
17df = df.dropna(subset=['geometry'])

1# Option 1: Keep first occurrence
2df = df.drop_duplicates(subset=['id'], keep='first')
3
4# Option 2: Regenerate IDs
5df['id'] = range(1, len(df) + 1)

# Filter to LineStrings only
df = df[df['geometry'].str.upper().str.startswith('LINESTRING')]

1import geopandas as gpd
2from shapely import wkt
3
4# Read and reproject
5gdf = gpd.read_file('roads.shp')
6gdf = gdf.to_crs('EPSG:4326')  # Convert to WGS84
7
8# Extract WKT
9df['geometry'] = gdf.geometry.apply(lambda g: g.wkt)

{region}_{data_type}_{date}_{version}.parquet

1import pandas as pd
2
3# Sample Amsterdam road segments
4sample_data = {
5    'id': [1, 2, 3, 4, 5],
6    'is_navigable': [True, True, True, True, False],
7    'geometry': [
8        'LINESTRING (4.8952 52.3702, 4.8960 52.3710)',
9        'LINESTRING (4.8960 52.3710, 4.8975 52.3725)',
10        'LINESTRING (4.8975 52.3725, 4.8990 52.3740)',
11        'LINESTRING (4.8990 52.3740, 4.9005 52.3755)',
12        'LINESTRING (4.9005 52.3755, 4.9020 52.3770)'
13    ]
14}
15
16df = pd.DataFrame(sample_data)
17df.to_parquet('sample_gem_input.parquet', index=False)
18print("Sample file created: sample_gem_input.parquet")

{"id":"abc","gers":"550e8400-e29b-41d4-a716-446655440000","confidence":99,"lr_id":"52.0197-76.36744#550e8400-e29b-41d4-a716-446655440000","lr_gers":"0.0-100.0#abc"}

1Parquet traces ──────────────────────────────┐
2                                              ├─→ prepare → match → pack → results.parquet
3CSV traces → (convert to Parquet traces) ─────┘

1root
2 |-- id: string (nullable = true)
3 |-- points: array (nullable = true)
4 |    |-- element: struct (containsNull = true)
5 |    |    |-- coord: struct (nullable = true)
6 |    |    |    |-- x: double (nullable = false)   # longitude (WGS84)
7 |    |    |    |-- y: double (nullable = false)   # latitude (WGS84)
8 |    |    |-- heading: double (nullable = false)  # degrees, 0–360
9 |    |    |-- lane: long (nullable = true)        # optional pre-assigned lane id
10 |    |    |-- timestamp: long (nullable = false)  # ms or sequential index
11 |    |    |-- velocity: double (nullable = false) # m/s, 0.0 if unknown

1from pyspark.sql import SparkSession
2from pyspark.sql.types import (
3    StructType, StructField, StringType, DoubleType, LongType, ArrayType
4)
5
6spark = SparkSession.builder.appName("GEM Trace Prep").getOrCreate()
7
8coord = StructType([
9    StructField("x", DoubleType(), False),  # longitude
10    StructField("y", DoubleType(), False),  # latitude
11])
12
13trace_point = StructType([
14    StructField("coord", coord, True),
15    StructField("heading", DoubleType(), False),
16    StructField("lane", LongType(), True),
17    StructField("timestamp", LongType(), False),
18    StructField("velocity", DoubleType(), False),
19])
20
21trace_schema = StructType([
22    StructField("id", StringType(), True),
23    StructField("points", ArrayType(trace_point, True), True),
24])
25
26rows = [
27    ("trace-001", [
28        {"coord": {"x": 4.8952, "y": 52.3702}, "heading": 34.2, "lane": None, "timestamp": 0, "velocity": 0.0},
29        {"coord": {"x": 4.8960, "y": 52.3710}, "heading": 34.2, "lane": None, "timestamp": 1, "velocity": 0.0},
30    ]),
31]
32
33spark.createDataFrame(rows, trace_schema).write.mode("overwrite").parquet("traces.parquet")

LINESTRING (longitude1 latitude1, longitude2 latitude2, ...)
POINT (longitude latitude)

1LINESTRING (145.18156 -37.87340, 145.18092 -37.87356)
2LINESTRING (4.8952 52.3702, 4.8960 52.3710, 4.8975 52.3725)
3POINT (4.8952 52.3702)

1import csv
2
3# Trace polylines as WKT LINESTRINGs
4traces = [
5    ("trace-001", "LINESTRING (4.8952 52.3702, 4.8960 52.3710, 4.8975 52.3725)"),
6    ("trace-002", "LINESTRING (4.8990 52.3740, 4.9005 52.3755, 4.9020 52.3770)"),
7]
8
9with open("road_profile.csv", "w", newline="") as f:
10    writer = csv.writer(f)
11    writer.writerow(["uuid", "shape"])
12    writer.writerows(traces)
13
14print(f"Wrote {len(traces)} traces")

1import csv
2
3# Single-point events: (uuid, heading, WKT POINT)
4events = [
5    ("event-001", 92.5, "POINT (4.8952 52.3702)"),
6    ("event-002", 180.0, "POINT (4.8990 52.3740)"),
7]
8
9with open("road_events.csv", "w", newline="") as f:
10    writer = csv.writer(f)
11    writer.writerow(["uuid", "heading", "coordinate"])
12    writer.writerows(events)
13
14print(f"Wrote {len(events)} events")

1import geopandas as gpd
2import pandas as pd
3
4# Read and reproject to WGS84
5gdf = gpd.read_file("roads.shp").to_crs("EPSG:4326")
6
7# Keep LineStrings only and export the road profile columns
8profile = pd.DataFrame({
9    "uuid": [f"trace-{i}" for i in range(len(gdf))],
10    "shape": gdf.geometry.apply(lambda g: g.wkt),
11})
12profile = profile[profile["shape"].str.upper().str.startswith("LINESTRING")]
13
14profile.to_csv("road_profile.csv", index=False)
15print(f"Exported {len(profile)} traces")

1from pyspark.sql import functions as F
2
3df = spark.read.parquet("traces.parquet")
4df.printSchema()  # Compare against the Parquet schema above
5
6assert "id" in df.columns and "points" in df.columns, "Missing id/points columns"
7
8problems = df.filter(
9    F.col("id").isNull() | (F.size("points") == 0)
10).count()
11print("OK" if problems == 0 else f"{problems} traces with null id or no points")

1import csv
2import re
3
4LINESTRING_RE = re.compile(r"^LINESTRING\s*\([^)]+\)$", re.IGNORECASE)
5POINT_RE = re.compile(r"^POINT\s*\([^)]+\)$", re.IGNORECASE)
6
7def validate_traces(filepath):
8    """Validate a road profile or road events CSV for GEM lane matching."""
9    errors, warnings = [], []
10
11    with open(filepath, newline="") as f:
12        reader = csv.DictReader(f)
13        columns = set(reader.fieldnames or [])
14
15        # Detect format from headers
16        if {"uuid", "shape"}.issubset(columns):
17            fmt, geom_col, geom_re = "road_profile", "shape", LINESTRING_RE
18        elif {"uuid", "heading", "coordinate"}.issubset(columns):
19            fmt, geom_col, geom_re = "road_events", "coordinate", POINT_RE
20        else:
21            return [f"Unrecognized columns: {sorted(columns)}"], []
22
23        print(f"Detected format: {fmt}")
24
25        seen_ids = set()
26        for line_no, row in enumerate(reader, start=2):
27            uuid = (row.get("uuid") or "").strip()
28            if not uuid:
29                errors.append(f"Line {line_no}: missing uuid")
30            elif uuid in seen_ids:
31                errors.append(f"Line {line_no}: duplicate uuid '{uuid}'")
32            else:
33                seen_ids.add(uuid)
34
35            geom = (row.get(geom_col) or "").strip()
36            if not geom_re.match(geom):
37                errors.append(f"Line {line_no}: invalid {geom_col} geometry")
38
39            if fmt == "road_events":
40                heading = (row.get("heading") or "").strip()
41                try:
42                    value = float(heading)
43                    if not 0.0 <= value < 360.0:
44                        warnings.append(f"Line {line_no}: heading {value} outside [0, 360)")
45                except ValueError:
46                    errors.append(f"Line {line_no}: heading '{heading}' is not a number")
47
48    print(f"\nErrors: {len(errors)}  Warnings: {len(warnings)}")
49    for error in errors:
50        print(f"  ❌ {error}")
51    for warning in warnings:
52        print(f"  ⚠️ {warning}")
53    if not errors:
54        print("\n✅ File is valid for GEM lane matching!")
55
56    return errors, warnings
57
58# Usage
59errors, warnings = validate_traces("road_profile.csv")

1from shapely import wkt
2from shapely.geometry import LineString, Point
3
4def is_valid(geom, expected):
5    try:
6        return isinstance(wkt.loads(geom), expected)
7    except Exception:
8        return False
9
10# Road profile
11df = df[df["shape"].apply(lambda g: is_valid(g, LineString))]
12# Road events
13df = df[df["coordinate"].apply(lambda g: is_valid(g, Point))]

# CSV: keep the first occurrence
df = df.drop_duplicates(subset=["uuid"], keep="first")

1import geopandas as gpd
2
3gdf = gpd.read_file("roads.shp").to_crs("EPSG:4326")  # Reproject to WGS84
4df["shape"] = gdf.geometry.apply(lambda g: g.wkt)      # lon lat order

1root
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)

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