From 3c7e6ab8a9bdcc9193a80e6d8179b61d63094d20 Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 17 Jun 2026 00:38:40 -0700 Subject: [PATCH 01/26] parse / & propagate speed limits --- src/scenic/formats/opendrive/xodr_parser.py | 90 +++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 8175d2c83..1ae17e41d 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -33,6 +33,73 @@ class OpenDriveWarning(UserWarning): def warn(message): warnings.warn(message, OpenDriveWarning, stacklevel=2) +# helper function to convert element to m/s +# might not be necessary either +def parse_speed_to_mps(speed_elem): + """Convert an OpenDRIVE ```` element to m/s.""" + raw = speed_elem.get("max") + if raw in ("no limit", "undefined"): + return None + value = float(raw) + unit = speed_elem.get("unit", "m/s") + if unit == "m/s": + return value + elif unit == "km/h": + return value / 3.6 + elif unit == "mph": + return value * 0.44704 + else: + raise ValueError(f"unsupported speed unit: {unit!r}") + + +# OpenDRIVE road ``type`` attribute values -> Scenic ``tags`` on ``NetworkElement``. +# Everything else (rural, bicycle) is mapped to the same value. +# This part may not even be necessary +ROAD_TYPE_MAP = {"motorway": "highway",} + +# helper function to get the primary type record +# key=lambda record: record[0] means the type record with the smallest s +def _primary_type_record(type_records): + """Return the type record starting at the smallest s (usually s=0).""" + if not type_records: + return None + return min(type_records, key=lambda record: record[0]) + + +def speed_limit_from_type_records(type_records): + """Road speed limit in m/s from parsed ````/```` records.""" + record = _primary_type_record(type_records) + return None if record is None else record[2] + + +def scenic_tags_from_type_records(type_records): + """Scenic semantic tags from parsed road ```` records.""" + record = _primary_type_record(type_records) + if record is None: + return frozenset() + open_drive_type = record[1] + #normalize the open drive type to a scenic tag + tag = ROAD_TYPE_MAP.get(open_drive_type, open_drive_type) + return frozenset({tag}) if tag else frozenset() + + +_SPEED_LIMIT_PROPAGATION_TYPES = ( + roadDomain.LaneGroup, + roadDomain.RoadSection, + roadDomain.LaneSection, + roadDomain.Lane, +) + + +def propagate_speed_limit(speed_limit, elements): + """Copy a road speed limit down the lane hierarchy.""" + if speed_limit is None: + return + for element in elements: + if isinstance(element, _SPEED_LIMIT_PROPAGATION_TYPES): + if element.speedLimit is None: + element.speedLimit = speed_limit + def buffer_union(polys, tolerance=0.01): return polygonUnion(polys, buf=tolerance, tolerance=tolerance) @@ -419,6 +486,8 @@ def __init__(self, name, id_, length, junction, drive_on_right=True): self.remappedStartLanes = None # hack for handling spurious initial lane sections + self.type_records = [] # [(s, openDriveRoadType, speedLimitMps), ...] + def get_ref_line_offset(self, s): if not self.offset: return 0 @@ -767,6 +836,13 @@ def calculate_geometry( def toScenicRoad(self, tolerance): assert self.sec_points + road_speed_limit = speed_limit_from_type_records(self.type_records) + road_tags = scenic_tags_from_type_records(self.type_records) + if len(self.type_records) > 1: + warn( + f"road {self.id_} has {len(self.type_records)} type segments;" + " using the segment with smallest s for speedLimit and tags" + ) allElements = [] # Create lane and road sections roadSections = [] @@ -1192,8 +1268,11 @@ def getEdges(forward): sections=roadSections, signals=tuple(roadSignals), crossings=(), # TODO add these! + speedLimit=road_speed_limit, + tags=road_tags, ) allElements.append(road) + propagate_speed_limit(road_speed_limit, allElements) # Set up parent references if forwardGroup: @@ -1530,6 +1609,17 @@ def parse(self, path): else: pred_link = succ_link = None + for type_elem in r.findall("type"): + s = float(type_elem.get("s")) + road_type = type_elem.get("type") + speed_elem = type_elem.find("speed") + speed_mps = ( + parse_speed_to_mps(speed_elem) + if speed_elem is not None + else None + ) + road.type_records.append((s, road_type, speed_mps)) + if road.length < self.tolerance: warn( f"road {road.id_} has length shorter than tolerance;" From bad180ca38350649a4e5e9abc9bc16b0205e2d5e Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 17 Jun 2026 00:56:45 -0700 Subject: [PATCH 02/26] added unit tests for additions to xodr_parser.py --- tests/formats/opendrive/test_opendrive.py | 150 +++++++++++++++++++++- 1 file changed, 149 insertions(+), 1 deletion(-) diff --git a/tests/formats/opendrive/test_opendrive.py b/tests/formats/opendrive/test_opendrive.py index 62b011c59..1fa2266ba 100644 --- a/tests/formats/opendrive/test_opendrive.py +++ b/tests/formats/opendrive/test_opendrive.py @@ -14,6 +14,9 @@ ParamCubic, RoadMap, makeCurve, + parse_speed_to_mps, + scenic_tags_from_type_records, + speed_limit_from_type_records, ) oldDir = os.getcwd() @@ -40,12 +43,20 @@ def test_map(path, runLocally, pytestconfig): plt.close() -def write_xodr(tmp_path, plan_view): +DEFAULT_PLAN_VIEW = """ + + + + """ + + +def write_xodr(tmp_path, plan_view, road_extras=""): path = tmp_path / "test.xodr" path.write_text( f""" + {road_extras} {plan_view} @@ -67,6 +78,14 @@ def write_xodr(tmp_path, plan_view): return path +def parse_scenic_network(tmp_path, road_extras="", plan_view=DEFAULT_PLAN_VIEW): + path = write_xodr(tmp_path, plan_view, road_extras=road_extras) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + return road_map.toScenicNetwork() + + def test_inconsistent_planview_length_warns(tmp_path): path = write_xodr( tmp_path, @@ -178,3 +197,132 @@ def test_make_curve_rejects_unknown_geometry_type(): with pytest.raises(NotImplementedError, match="unhandled OpenDRIVE geometry type"): makeCurve(0.0, 0.0, 0.0, 10.0, curve_elem) + + +@pytest.mark.parametrize( + "max_attr, unit, expected_mps", + [ + ("25", "m/s", 25.0), + ("90", "km/h", 90 / 3.6), + ("55", "mph", 55 * 0.44704), + ], +) +def test_parse_speed_to_mps_converts_units(max_attr, unit, expected_mps): + speed_elem = ET.fromstring(f'') + assert parse_speed_to_mps(speed_elem) == pytest.approx(expected_mps) + + +@pytest.mark.parametrize("max_attr", ["no limit", "undefined"]) +def test_parse_speed_to_mps_unlimited(max_attr): + speed_elem = ET.fromstring(f'') + assert parse_speed_to_mps(speed_elem) is None + + +def test_parse_speed_to_mps_rejects_unknown_unit(): + speed_elem = ET.fromstring('') + with pytest.raises(ValueError, match="unsupported speed unit"): + parse_speed_to_mps(speed_elem) + + +def test_speed_limit_from_type_records(): + records = [(0.0, "town", 13.89), (10.0, "motorway", 27.78)] + assert speed_limit_from_type_records(records) == pytest.approx(13.89) + + +def test_speed_limit_from_type_records_empty(): + assert speed_limit_from_type_records([]) is None + + +def test_scenic_tags_from_type_records_maps_motorway_to_highway(): + records = [(0.0, "motorway", None)] + assert scenic_tags_from_type_records(records) == frozenset({"highway"}) + + +def test_scenic_tags_from_type_records_passes_through_unknown_types(): + records = [(0.0, "town", None)] + assert scenic_tags_from_type_records(records) == frozenset({"town"}) + + +def test_scenic_tags_from_type_records_empty(): + assert scenic_tags_from_type_records([]) == frozenset() + + +def test_road_type_records_parsed(tmp_path): + path = write_xodr( + tmp_path, + DEFAULT_PLAN_VIEW, + road_extras='', + ) + road_map = RoadMap() + road_map.parse(path) + s, road_type, speed_mps = road_map.roads[7].type_records[0] + assert s == pytest.approx(0.0) + assert road_type == "town" + assert speed_mps == pytest.approx(50 / 3.6) + + +def test_scenic_road_speed_limit_and_tags(tmp_path): + network = parse_scenic_network( + tmp_path, + road_extras='', + ) + road = network.roads[0] + expected = 50 / 3.6 + assert road.speedLimit == pytest.approx(expected) + assert road.tags == frozenset({"town"}) + + +def test_scenic_speed_limit_propagates_to_lane_hierarchy(tmp_path): + network = parse_scenic_network( + tmp_path, + road_extras='', + ) + road = network.roads[0] + expected = 36 / 3.6 + assert road.forwardLanes.speedLimit == pytest.approx(expected) + assert road.sections[0].speedLimit == pytest.approx(expected) + assert road.lanes[0].speedLimit == pytest.approx(expected) + lane_section = road.lanes[0].sections[0] + assert lane_section.speedLimit == pytest.approx(expected) + + +def test_scenic_motorway_tag_mapped_to_highway(tmp_path): + network = parse_scenic_network( + tmp_path, + road_extras='', + ) + road = network.roads[0] + assert road.tags == frozenset({"highway"}) + assert road.speedLimit == pytest.approx(120 / 3.6) + + +def test_scenic_road_without_type_has_no_limit_or_tags(tmp_path): + network = parse_scenic_network(tmp_path) + road = network.roads[0] + assert road.speedLimit is None + assert road.tags == frozenset() + + +def test_scenic_road_no_limit_speed(tmp_path): + network = parse_scenic_network( + tmp_path, + road_extras='', + ) + road = network.roads[0] + assert road.speedLimit is None + assert road.tags == frozenset({"town"}) + + +def test_multiple_type_segments_use_smallest_s(tmp_path): + road_extras = """ + + + """ + with pytest.warns( + OpenDriveWarning, + match="road 7 has 2 type segments", + ): + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = network.roads[0] + assert road.speedLimit == pytest.approx(30 / 3.6) + assert road.tags == frozenset({"town"}) From f5e0b1420270348cc50980aecc70f3021f849430 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sat, 27 Jun 2026 11:11:18 -0700 Subject: [PATCH 03/26] dropped inferencing and heuristics --- src/scenic/formats/opendrive/xodr_parser.py | 115 ++++++++++++++------ 1 file changed, 83 insertions(+), 32 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 1ae17e41d..9155a2627 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -33,9 +33,9 @@ class OpenDriveWarning(UserWarning): def warn(message): warnings.warn(message, OpenDriveWarning, stacklevel=2) -# helper function to convert element to m/s -# might not be necessary either -def parse_speed_to_mps(speed_elem): +# Semantic tags and speed limits from OpenDRIVE / and junction metadata. + +def speed_to_mps(speed_elem): """Convert an OpenDRIVE ```` element to m/s.""" raw = speed_elem.get("max") if raw in ("no limit", "undefined"): @@ -52,14 +52,7 @@ def parse_speed_to_mps(speed_elem): raise ValueError(f"unsupported speed unit: {unit!r}") -# OpenDRIVE road ``type`` attribute values -> Scenic ``tags`` on ``NetworkElement``. -# Everything else (rural, bicycle) is mapped to the same value. -# This part may not even be necessary -ROAD_TYPE_MAP = {"motorway": "highway",} - -# helper function to get the primary type record -# key=lambda record: record[0] means the type record with the smallest s -def _primary_type_record(type_records): +def __primary_type_record(type_records): """Return the type record starting at the smallest s (usually s=0).""" if not type_records: return None @@ -68,39 +61,89 @@ def _primary_type_record(type_records): def speed_limit_from_type_records(type_records): """Road speed limit in m/s from parsed ````/```` records.""" - record = _primary_type_record(type_records) + record = __primary_type_record(type_records) return None if record is None else record[2] def scenic_tags_from_type_records(type_records): - """Scenic semantic tags from parsed road ```` records.""" - record = _primary_type_record(type_records) - if record is None: - return frozenset() - open_drive_type = record[1] - #normalize the open drive type to a scenic tag - tag = ROAD_TYPE_MAP.get(open_drive_type, open_drive_type) - return frozenset({tag}) if tag else frozenset() - - -_SPEED_LIMIT_PROPAGATION_TYPES = ( + """Scenic semantic tags from all parsed road ```` records.""" + tags = set() + for _, open_drive_type, _ in type_records: + if open_drive_type: + tags.add(open_drive_type) + return frozenset(tags) + + +def merge_scenic_tags(*tag_sets): + """Combine tags from road ```` records and junction metadata.""" + tags = set() + for tag_set in tag_sets: + if tag_set: + tags.update(tag_set) + return frozenset(tags) + + +def _junction_tags(junction): + """Tags taken directly from a junction's OpenDRIVE ``type`` and ``name``.""" + tags = set() + if junction.type_ and junction.type_ != "default": + tags.add(junction.type_) + if junction.name: + tags.add(junction.name) + return tags + + +def assign_semantic_tags(road_map): + """Populate ``Road.extra_tags`` from junction type and name.""" + extra = {road_id: set() for road_id in road_map.roads} + + for jid, junction in road_map.junctions.items(): + tags = _junction_tags(junction) + if not tags: + continue + for road_id, road in road_map.roads.items(): + if road.junction == jid: + extra[road_id].update(tags) + elif road.junction is None and ( + road.predecessor == jid or road.successor == jid + ): + extra[road_id].update(tags) + + for road_id, road in road_map.roads.items(): + road.extra_tags = frozenset(extra[road_id]) + + +_ROAD_PROPAGATION_TYPES = ( roadDomain.LaneGroup, roadDomain.RoadSection, roadDomain.LaneSection, roadDomain.Lane, ) - +# later - better to have types specific to portions of the road def propagate_speed_limit(speed_limit, elements): - """Copy a road speed limit down the lane hierarchy.""" + """Copy a road speed limit down to lane groups, sections, and lanes.""" if speed_limit is None: return for element in elements: - if isinstance(element, _SPEED_LIMIT_PROPAGATION_TYPES): + if isinstance(element, _ROAD_PROPAGATION_TYPES): if element.speedLimit is None: element.speedLimit = speed_limit +def propagate_tags(tags, elements): + """Copy road semantic tags down to lane groups, sections, and lanes.""" + if not tags: + return + for element in elements: + if isinstance(element, _ROAD_PROPAGATION_TYPES): + if not element.tags: + element.tags = tags + + +parse_speed_to_mps = speed_to_mps + + def buffer_union(polys, tolerance=0.01): return polygonUnion(polys, buf=tolerance, tolerance=tolerance) @@ -441,9 +484,10 @@ def __init__(self, incoming_id, connecting_id, connecting_contact, lane_links): # dict mapping incoming to connecting lane ids (empty = identity mapping) self.lane_links = lane_links - def __init__(self, id_, name): + def __init__(self, id_, name, type_="default"): self.id_ = id_ self.name = name + self.type_ = type_ self.connections = [] # Ids of roads that are paths within junction: self.paths = [] @@ -461,7 +505,7 @@ def __init__(self, name, id_, length, junction, drive_on_right=True): self.name = name self.id_ = id_ self.length = length - self.junction = junction if junction != "-1" else None + self.junction = int(junction) if junction != "-1" else None self.predecessor = None self.successor = None self.signals = [] # List of Signal objects. @@ -487,6 +531,7 @@ def __init__(self, name, id_, length, junction, drive_on_right=True): self.remappedStartLanes = None # hack for handling spurious initial lane sections self.type_records = [] # [(s, openDriveRoadType, speedLimitMps), ...] + self.extra_tags = frozenset() def get_ref_line_offset(self, s): if not self.offset: @@ -837,11 +882,14 @@ def calculate_geometry( def toScenicRoad(self, tolerance): assert self.sec_points road_speed_limit = speed_limit_from_type_records(self.type_records) - road_tags = scenic_tags_from_type_records(self.type_records) + road_tags = merge_scenic_tags( + scenic_tags_from_type_records(self.type_records), + self.extra_tags, + ) if len(self.type_records) > 1: warn( f"road {self.id_} has {len(self.type_records)} type segments;" - " using the segment with smallest s for speedLimit and tags" + " using the segment with smallest s for speedLimit" ) allElements = [] # Create lane and road sections @@ -1273,6 +1321,7 @@ def getEdges(forward): ) allElements.append(road) propagate_speed_limit(road_speed_limit, allElements) + propagate_tags(road_tags, allElements) # Set up parent references if forwardGroup: @@ -1563,7 +1612,7 @@ def parse(self, path): # parse junctions for j in root.iter("junction"): - junction = Junction(int(j.get("id")), j.get("name")) + junction = Junction(int(j.get("id")), j.get("name"), j.get("type", "default")) for c in j.iter("connection"): ty = c.get("type", "default") if ty != "default": @@ -1609,12 +1658,13 @@ def parse(self, path): else: pred_link = succ_link = None + # 2.3 Handle type and speed parsing for type_elem in r.findall("type"): s = float(type_elem.get("s")) road_type = type_elem.get("type") speed_elem = type_elem.find("speed") speed_mps = ( - parse_speed_to_mps(speed_elem) + speed_to_mps(speed_elem) if speed_elem is not None else None ) @@ -1808,6 +1858,7 @@ def popLastSectionIfShort(l): def toScenicNetwork(self): assert self.intersection_region is not None + assign_semantic_tags(self) # Prepare registry of network elements allElements = {} From 87289126e64a3f28910014f31b288f3fb16f4a44 Mon Sep 17 00:00:00 2001 From: aprabou Date: Thu, 2 Jul 2026 12:48:58 -0700 Subject: [PATCH 04/26] updated lane propagation behavior across sections --- src/scenic/formats/opendrive/xodr_parser.py | 307 +++++++++++++++++--- 1 file changed, 270 insertions(+), 37 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 9155a2627..5b960e1a9 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -33,8 +33,6 @@ class OpenDriveWarning(UserWarning): def warn(message): warnings.warn(message, OpenDriveWarning, stacklevel=2) -# Semantic tags and speed limits from OpenDRIVE / and junction metadata. - def speed_to_mps(speed_elem): """Convert an OpenDRIVE ```` element to m/s.""" raw = speed_elem.get("max") @@ -51,28 +49,140 @@ def speed_to_mps(speed_elem): else: raise ValueError(f"unsupported speed unit: {unit!r}") +def _speed_limit_ranges_from_s_speed_records(records, domain_length): + """Build ``(s_start, s_end, speed_mps)`` ranges from ``[(s, speed_mps), ...]``.""" + if not records: + return [] + + sorted_records = sorted(records, key=lambda record: record[0]) + ranges = [] + current_speed = None + + for i, (s, speed_mps) in enumerate(sorted_records): + if speed_mps is not None: + current_speed = speed_mps + s_end = ( + sorted_records[i + 1][0] + if i + 1 < len(sorted_records) + else domain_length + ) + if s_end <= s or current_speed is None: + continue + if ranges and ranges[-1][2] == current_speed and ranges[-1][1] == s: + ranges[-1] = (ranges[-1][0], s_end, current_speed) + else: + ranges.append((s, s_end, current_speed)) + + return ranges + -def __primary_type_record(type_records): - """Return the type record starting at the smallest s (usually s=0).""" +def speed_limit_ranges_from_type_records(type_records, road_length): + """Build ``(s_start, s_end, speed_mps)`` ranges from OpenDRIVE ```` records.""" if not type_records: - return None - return min(type_records, key=lambda record: record[0]) + return [] + sorted_records = sorted(type_records, key=lambda record: record[0]) + s_speed_records = [] + current_speed = None + for s, _road_type, speed_mps in sorted_records: + if speed_mps is not None: + current_speed = speed_mps + s_speed_records.append((s, current_speed)) + return _speed_limit_ranges_from_s_speed_records(s_speed_records, road_length) -def speed_limit_from_type_records(type_records): - """Road speed limit in m/s from parsed ````/```` records.""" - record = __primary_type_record(type_records) - return None if record is None else record[2] +def speed_limit_ranges_from_lane_records(speed_records, section_length): + """Build ``(s_start, s_end, speed_mps)`` ranges from lane ```` records. + + ``speed_records`` are ``(sOffset, speed_mps)`` pairs relative to the lane + section start; ``section_length`` is the lane section length in meters. + """ + return _speed_limit_ranges_from_s_speed_records(speed_records, section_length) + + +def _clip_ranges_to_interval(ranges, interval_start, interval_end): + """Clip ``(s_start, s_end, speed_mps)`` ranges to ``[interval_start, interval_end)``.""" + clipped = [] + for range_start, range_end, speed in ranges: + if range_end <= interval_start or range_start >= interval_end: + continue + clipped.append( + ( + max(range_start, interval_start), + min(range_end, interval_end), + speed, + ) + ) + return clipped -def scenic_tags_from_type_records(type_records): - """Scenic semantic tags from all parsed road ```` records.""" - tags = set() - for _, open_drive_type, _ in type_records: - if open_drive_type: - tags.add(open_drive_type) - return frozenset(tags) +def effective_speed_limit_ranges(road_ranges, lane_ranges, section_s0, section_length): + """Merge road and lane speed ranges for one OpenDRIVE lane section. + + Returns ``(s_start, s_end, speed_mps)`` ranges with *s* relative to the lane + section start. When lane speeds are present they fully replace road speeds. + """ + if lane_ranges: + return lane_ranges + + section_end = section_s0 + section_length + road_in_section = _clip_ranges_to_interval(road_ranges, section_s0, section_end) + return [ + (range_start - section_s0, range_end - section_s0, speed) + for range_start, range_end, speed in road_in_section + ] + +# function to return the minimum speed and overlapping speeds given an s-interval with multiple speed limits. +def speed_limit_for_s_interval(ranges, s_start, s_end): + """Return ``(speed_limit, overlapping_speeds)`` for ``[s_start, s_end)``.""" + if not ranges or s_end <= s_start: + return None, frozenset() + + # overlapping range is defined as such: + # if range start is less than s_end + # AND range end is greater than s_start. + overlapping = { + speed + for range_start, range_end, speed in ranges + if range_end > s_start and range_start < s_end + } + + # if no overlapping speeds, return (None, frozenset()) + # otherwise, return (min(speed), frozenset(overlapping)) + if not overlapping: + return None, frozenset() + return min(overlapping), frozenset(overlapping) + +################################################################################################################ + +def apply_lane_speed_limit(road_speed_limit, lane_speed_limit): + if lane_speed_limit is not None: + return lane_speed_limit + return road_speed_limit + + +def assign_speed_limit_from_ranges(element, ranges, warn_context=None): + """Set ``speedLimit`` and ``speedLimitRanges`` from effective speed ranges.""" + if not ranges: + return + + speeds = {speed for _, _, speed in ranges if speed is not None} + if not speeds: + return + + element.speedLimit = min(speeds) + if len(speeds) > 1: + element.speedLimitRanges = tuple(ranges) + if warn_context is not None: + speeds_text = ", ".join( + f"{speed:.4g} m/s" for speed in sorted(speeds) + ) + warn( + f"{warn_context}: spans multiple speed limits {{{speeds_text}}};" + f" using minimum {element.speedLimit:.4g} m/s" + ) + else: + element.speedLimitRanges = () def merge_scenic_tags(*tag_sets): """Combine tags from road ```` records and junction metadata.""" @@ -83,6 +193,13 @@ def merge_scenic_tags(*tag_sets): return frozenset(tags) +def lane_scenic_tags(lane_type): + """Tags for a Lane or LaneSection from the OpenDRIVE lane ``type`` attribute.""" + if not lane_type: + return frozenset() + return frozenset({lane_type}) + + def _junction_tags(junction): """Tags taken directly from a junction's OpenDRIVE ``type`` and ``name``.""" tags = set() @@ -95,32 +212,51 @@ def _junction_tags(junction): def assign_semantic_tags(road_map): """Populate ``Road.extra_tags`` from junction type and name.""" + # Create a dictionary with each road_id in the road map, initializing with an empty set for collecting tags extra = {road_id: set() for road_id in road_map.roads} + # Iterate through all junctions in the road map for jid, junction in road_map.junctions.items(): + + # Get the set of tags for this junction based on type and name tags = _junction_tags(junction) if not tags: + # If the junction has no relevant tags, skip it continue + + # Iterate through all roads to determine which should receive the junction's tags + # • If this road is directly part of the junction: + # - inherit all of its tags for road_id, road in road_map.roads.items(): if road.junction == jid: extra[road_id].update(tags) + + # For roads that are not themselves junctions: + # • If the road is not itself a junction but connects to one via predecessor or successor: + # - also inherit the tags from the junction elif road.junction is None and ( road.predecessor == jid or road.successor == jid ): extra[road_id].update(tags) + # Finally, convert each collected set of tags into an immutable frozenset, and assign to the road's extra_tags field for road_id, road in road_map.roads.items(): road.extra_tags = frozenset(extra[road_id]) -_ROAD_PROPAGATION_TYPES = ( +_ROAD_LEVEL_PROPAGATION_TYPES = ( roadDomain.LaneGroup, roadDomain.RoadSection, +) + +# later - better to have types specific to portions of the road +_LANE_PROPAGATION_TYPES = ( roadDomain.LaneSection, roadDomain.Lane, ) +_ROAD_PROPAGATION_TYPES = _ROAD_LEVEL_PROPAGATION_TYPES + _LANE_PROPAGATION_TYPES + -# later - better to have types specific to portions of the road def propagate_speed_limit(speed_limit, elements): """Copy a road speed limit down to lane groups, sections, and lanes.""" if speed_limit is None: @@ -131,12 +267,26 @@ def propagate_speed_limit(speed_limit, elements): element.speedLimit = speed_limit +def apply_lane_speed_limits(elements): + """Set lane-level scalar speed limits from their lane sections.""" + for element in elements: + if not isinstance(element, roadDomain.Lane): + continue + section_limits = [ + section.speedLimit + for section in element.sections + if section.speedLimit is not None + ] + if section_limits: + element.speedLimit = max(section_limits) + + def propagate_tags(tags, elements): - """Copy road semantic tags down to lane groups, sections, and lanes.""" + """Copy road-level semantic tags to lane groups and road sections.""" if not tags: return for element in elements: - if isinstance(element, _ROAD_PROPAGATION_TYPES): + if isinstance(element, _ROAD_LEVEL_PROPAGATION_TYPES): if not element.tags: element.tags = tags @@ -396,6 +546,8 @@ class Lane: def __init__(self, id_, type_, pred=None, succ=None): self.id_ = id_ self.width = [] # List of tuples (Poly3, int) for width and s-offset. + self.speed_records = [] # [(sOffset, speed_mps), ...] from elements. + self.speed_limit = None # First record's speed; kept for backward compatibility. self.type_ = type_ self.pred = pred self.succ = succ @@ -724,6 +876,8 @@ def calc_geometry_for_type(self, lane_types, num, tolerance, calc_gap=False): prev_id = id_ - 1 else: prev_id = id_ + 1 + if prev_id not in offsets: + continue if ( offsets[id_] == offsets[prev_id] @@ -881,25 +1035,52 @@ def calculate_geometry( def toScenicRoad(self, tolerance): assert self.sec_points - road_speed_limit = speed_limit_from_type_records(self.type_records) - road_tags = merge_scenic_tags( - scenic_tags_from_type_records(self.type_records), - self.extra_tags, + + # collect speed ranges from type records and total road length + speed_ranges = speed_limit_ranges_from_type_records( + self.type_records, self.length ) - if len(self.type_records) > 1: - warn( - f"road {self.id_} has {len(self.type_records)} type segments;" - " using the segment with smallest s for speedLimit" - ) + + type_tags = frozenset( + open_drive_type + for _, open_drive_type, _ in self.type_records + if open_drive_type + ) + road_level_tags = merge_scenic_tags(type_tags, self.extra_tags) allElements = [] # Create lane and road sections roadSections = [] last_section = None forwardSidewalks, backwardSidewalks = [], [] forwardShoulders, backwardShoulders = [], [] - for sec, pts, sec_poly, lane_polys in zip( - self.lane_secs, self.sec_points, self.sec_polys, self.sec_lane_polys + ################################################################################################################ + for sec_index, (sec, pts, sec_poly, lane_polys) in enumerate( + zip( + self.lane_secs, + self.sec_points, + self.sec_polys, + self.sec_lane_polys, + ) ): + s_start = sec.s0 + s_end = ( + self.lane_secs[sec_index + 1].s0 + if sec_index + 1 < len(self.lane_secs) + else self.length + ) + section_speed_limit, overlapping_speeds = speed_limit_for_s_interval( + speed_ranges, s_start, s_end + ) + if len(overlapping_speeds) > 1: + speeds_text = ", ".join( + f"{speed:.4g} m/s" for speed in sorted(overlapping_speeds) + ) + warn( + f"road {self.id_} section s=[{s_start},{s_end}):" + f" spans multiple speed limits {{{speeds_text}}};" + f" using minimum {section_speed_limit:.4g} m/s" + ) + ################################################################################################################ pts = [pt[:2] for pt in pts] # drop s coordinate assert sec.drivable_lanes laneSections = {} @@ -933,6 +1114,7 @@ def toScenicRoad(self, tolerance): road=None, openDriveID=id_, isForward=id_ < 0, + tags=lane_scenic_tags(lane.type_), ) section._original_lane = lane laneSections[id_] = section @@ -950,6 +1132,28 @@ def toScenicRoad(self, tolerance): ) roadSections.append(section) allElements.append(section) + ################################################################################################################ + propagate_speed_limit(section_speed_limit, [section]) + section_length = s_end - s_start + for id_, lane_section in laneSections.items(): + lane = sec.drivable_lanes[id_] + lane_ranges = speed_limit_ranges_from_lane_records( + lane.speed_records, section_length + ) + effective_ranges = effective_speed_limit_ranges( + speed_ranges, lane_ranges, s_start, section_length + ) + if effective_ranges: + assign_speed_limit_from_ranges( + lane_section, + effective_ranges, + warn_context=( + f"road {self.id_} lane {id_} section s=[{s_start},{s_end})" + ), + ) + else: + propagate_speed_limit(section_speed_limit, [lane_section]) + ################################################################################################################ last_section = section fss, bss = {}, {} @@ -1192,6 +1396,7 @@ def makeShoulder(sections, backward=False): road=None, sections=tuple(sections), successor=successorLane, # will correct inter-road links later + tags=frozenset().union(*(sec.tags for sec in sections)), ) nextID += 1 for section in sections: @@ -1302,6 +1507,16 @@ def getEdges(forward): else: leftEdge = forwardGroup.leftEdge centerline = PolylineRegion(tuple(pt[:2] for pt in self.ref_line_points)) + ################################################################################################################ + section_speed_limits = [ + section.speedLimit + for section in roadSections + if section.speedLimit is not None + ] + road_speed_limit = ( + min(section_speed_limits) if section_speed_limits else None + ) + ################################################################################################################ road = roadDomain.Road( name=self.name, uid=f"road{self.id_}", # need prefix to prevent collisions with intersections @@ -1317,11 +1532,12 @@ def getEdges(forward): signals=tuple(roadSignals), crossings=(), # TODO add these! speedLimit=road_speed_limit, - tags=road_tags, + tags=road_level_tags, ) allElements.append(road) propagate_speed_limit(road_speed_limit, allElements) - propagate_tags(road_tags, allElements) + apply_lane_speed_limits(allElements) + propagate_tags(road_level_tags, allElements) # Set up parent references if forwardGroup: @@ -1392,14 +1608,24 @@ def __init__( fill_intersections=True, drivable_lane_types=( "driving", + "stop", + "parking", "entry", "exit", - "offRamp", "onRamp", + "offRamp", "connectingRamp", + "slipLane", + "biking", + ), + sidewalk_lane_types=("walking", "sidewalk"), # sidewalk deprecated + shoulder_lane_types=( + "shoulder", + "border", + "restricted", + "curb", + "none", ), - sidewalk_lane_types=("sidewalk",), - shoulder_lane_types=("shoulder", "parking", "stop", "border"), elide_short_roads=False, ): self.tolerance = self.defaultTolerance if tolerance is None else tolerance @@ -1553,6 +1779,13 @@ def __parse_lanes(self, lanes_elem): float(w.get("d")), ) lane.width.append((w_poly, float(w.get("sOffset")))) + for speed_elem in l.iter("speed"): + lane.speed_records.append( + (float(speed_elem.get("sOffset", 0)), speed_to_mps(speed_elem)) + ) + lane.speed_records.sort(key=lambda record: record[0]) + if lane.speed_records: + lane.speed_limit = lane.speed_records[0][1] lanes[id_] = lane return lanes From 4d7338bc31d4b212368e199fd0f7fb90c7db9ed3 Mon Sep 17 00:00:00 2001 From: aprabou Date: Thu, 2 Jul 2026 18:00:21 -0700 Subject: [PATCH 05/26] complete formatting and comments --- src/scenic/formats/opendrive/xodr_parser.py | 89 +++++++-------------- 1 file changed, 28 insertions(+), 61 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 5b960e1a9..55e6cba55 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -33,6 +33,7 @@ class OpenDriveWarning(UserWarning): def warn(message): warnings.warn(message, OpenDriveWarning, stacklevel=2) + def speed_to_mps(speed_elem): """Convert an OpenDRIVE ```` element to m/s.""" raw = speed_elem.get("max") @@ -47,7 +48,8 @@ def speed_to_mps(speed_elem): elif unit == "mph": return value * 0.44704 else: - raise ValueError(f"unsupported speed unit: {unit!r}") + raise ValueError(f"unsupported speed unit: {unit!r}") + def _speed_limit_ranges_from_s_speed_records(records, domain_length): """Build ``(s_start, s_end, speed_mps)`` ranges from ``[(s, speed_mps), ...]``.""" @@ -61,11 +63,7 @@ def _speed_limit_ranges_from_s_speed_records(records, domain_length): for i, (s, speed_mps) in enumerate(sorted_records): if speed_mps is not None: current_speed = speed_mps - s_end = ( - sorted_records[i + 1][0] - if i + 1 < len(sorted_records) - else domain_length - ) + s_end = sorted_records[i + 1][0] if i + 1 < len(sorted_records) else domain_length if s_end <= s or current_speed is None: continue if ranges and ranges[-1][2] == current_speed and ranges[-1][1] == s: @@ -132,34 +130,29 @@ def effective_speed_limit_ranges(road_ranges, lane_ranges, section_s0, section_l for range_start, range_end, speed in road_in_section ] -# function to return the minimum speed and overlapping speeds given an s-interval with multiple speed limits. + def speed_limit_for_s_interval(ranges, s_start, s_end): - """Return ``(speed_limit, overlapping_speeds)`` for ``[s_start, s_end)``.""" + """Speed limit applying to the interval ``[s_start, s_end)``. + + Returns ``(speed_limit, overlapping_speeds)``, where ``overlapping_speeds`` is + the set of every limit in ``ranges`` that overlaps the interval and + ``speed_limit`` is their minimum -- the conservative choice when an interval + straddles more than one limit. Both are empty/``None`` when nothing overlaps. + """ if not ranges or s_end <= s_start: return None, frozenset() - # overlapping range is defined as such: - # if range start is less than s_end - # AND range end is greater than s_start. + # A range overlaps the interval iff it starts before s_end and ends after s_start. overlapping = { speed for range_start, range_end, speed in ranges if range_end > s_start and range_start < s_end } - # if no overlapping speeds, return (None, frozenset()) - # otherwise, return (min(speed), frozenset(overlapping)) if not overlapping: return None, frozenset() return min(overlapping), frozenset(overlapping) -################################################################################################################ - -def apply_lane_speed_limit(road_speed_limit, lane_speed_limit): - if lane_speed_limit is not None: - return lane_speed_limit - return road_speed_limit - def assign_speed_limit_from_ranges(element, ranges, warn_context=None): """Set ``speedLimit`` and ``speedLimitRanges`` from effective speed ranges.""" @@ -174,9 +167,7 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): if len(speeds) > 1: element.speedLimitRanges = tuple(ranges) if warn_context is not None: - speeds_text = ", ".join( - f"{speed:.4g} m/s" for speed in sorted(speeds) - ) + speeds_text = ", ".join(f"{speed:.4g} m/s" for speed in sorted(speeds)) warn( f"{warn_context}: spans multiple speed limits {{{speeds_text}}};" f" using minimum {element.speedLimit:.4g} m/s" @@ -184,6 +175,7 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): else: element.speedLimitRanges = () + def merge_scenic_tags(*tag_sets): """Combine tags from road ```` records and junction metadata.""" tags = set() @@ -211,35 +203,21 @@ def _junction_tags(junction): def assign_semantic_tags(road_map): - """Populate ``Road.extra_tags`` from junction type and name.""" - # Create a dictionary with each road_id in the road map, initializing with an empty set for collecting tags + """Populate each ``Road.extra_tags`` from the type/name of its junction. + + A road belonging to a junction (``road.junction`` equals the junction id) + inherits that junction's semantic tags; all other roads get no extra tags. + """ extra = {road_id: set() for road_id in road_map.roads} - # Iterate through all junctions in the road map for jid, junction in road_map.junctions.items(): - - # Get the set of tags for this junction based on type and name tags = _junction_tags(junction) if not tags: - # If the junction has no relevant tags, skip it continue - - # Iterate through all roads to determine which should receive the junction's tags - # • If this road is directly part of the junction: - # - inherit all of its tags for road_id, road in road_map.roads.items(): if road.junction == jid: extra[road_id].update(tags) - # For roads that are not themselves junctions: - # • If the road is not itself a junction but connects to one via predecessor or successor: - # - also inherit the tags from the junction - elif road.junction is None and ( - road.predecessor == jid or road.successor == jid - ): - extra[road_id].update(tags) - - # Finally, convert each collected set of tags into an immutable frozenset, and assign to the road's extra_tags field for road_id, road in road_map.roads.items(): road.extra_tags = frozenset(extra[road_id]) @@ -249,7 +227,7 @@ def assign_semantic_tags(road_map): roadDomain.RoadSection, ) -# later - better to have types specific to portions of the road + _LANE_PROPAGATION_TYPES = ( roadDomain.LaneSection, roadDomain.Lane, @@ -268,7 +246,7 @@ def propagate_speed_limit(speed_limit, elements): def apply_lane_speed_limits(elements): - """Set lane-level scalar speed limits from their lane sections.""" + """Set each Lane's scalar ``speedLimit`` to the max over its lane sections.""" for element in elements: if not isinstance(element, roadDomain.Lane): continue @@ -291,9 +269,6 @@ def propagate_tags(tags, elements): element.tags = tags -parse_speed_to_mps = speed_to_mps - - def buffer_union(polys, tolerance=0.01): return polygonUnion(polys, buf=tolerance, tolerance=tolerance) @@ -1053,7 +1028,6 @@ def toScenicRoad(self, tolerance): last_section = None forwardSidewalks, backwardSidewalks = [], [] forwardShoulders, backwardShoulders = [], [] - ################################################################################################################ for sec_index, (sec, pts, sec_poly, lane_polys) in enumerate( zip( self.lane_secs, @@ -1080,7 +1054,6 @@ def toScenicRoad(self, tolerance): f" spans multiple speed limits {{{speeds_text}}};" f" using minimum {section_speed_limit:.4g} m/s" ) - ################################################################################################################ pts = [pt[:2] for pt in pts] # drop s coordinate assert sec.drivable_lanes laneSections = {} @@ -1132,7 +1105,7 @@ def toScenicRoad(self, tolerance): ) roadSections.append(section) allElements.append(section) - ################################################################################################################ + propagate_speed_limit(section_speed_limit, [section]) section_length = s_end - s_start for id_, lane_section in laneSections.items(): @@ -1153,7 +1126,7 @@ def toScenicRoad(self, tolerance): ) else: propagate_speed_limit(section_speed_limit, [lane_section]) - ################################################################################################################ + last_section = section fss, bss = {}, {} @@ -1507,16 +1480,14 @@ def getEdges(forward): else: leftEdge = forwardGroup.leftEdge centerline = PolylineRegion(tuple(pt[:2] for pt in self.ref_line_points)) - ################################################################################################################ + section_speed_limits = [ section.speedLimit for section in roadSections if section.speedLimit is not None ] - road_speed_limit = ( - min(section_speed_limits) if section_speed_limits else None - ) - ################################################################################################################ + road_speed_limit = min(section_speed_limits) if section_speed_limits else None + road = roadDomain.Road( name=self.name, uid=f"road{self.id_}", # need prefix to prevent collisions with intersections @@ -1896,11 +1867,7 @@ def parse(self, path): s = float(type_elem.get("s")) road_type = type_elem.get("type") speed_elem = type_elem.find("speed") - speed_mps = ( - speed_to_mps(speed_elem) - if speed_elem is not None - else None - ) + speed_mps = speed_to_mps(speed_elem) if speed_elem is not None else None road.type_records.append((s, road_type, speed_mps)) if road.length < self.tolerance: From 9e2a24423f7492c67103ede179c52aca0bd0f1b0 Mon Sep 17 00:00:00 2001 From: aprabou Date: Thu, 2 Jul 2026 18:42:56 -0700 Subject: [PATCH 06/26] completed test suite for xodr_parser --- tests/formats/opendrive/test_opendrive.py | 677 ++++++++++++++++++---- 1 file changed, 570 insertions(+), 107 deletions(-) diff --git a/tests/formats/opendrive/test_opendrive.py b/tests/formats/opendrive/test_opendrive.py index 1fa2266ba..3df71dd03 100644 --- a/tests/formats/opendrive/test_opendrive.py +++ b/tests/formats/opendrive/test_opendrive.py @@ -7,16 +7,27 @@ import pytest from scenic.core.geometry import TriangulationError +from scenic.domains.driving import roads as roadDomain from scenic.formats.opendrive import OpenDriveWorkspace from scenic.formats.opendrive.xodr_parser import ( Cubic, + Junction, OpenDriveWarning, ParamCubic, RoadMap, + _junction_tags, + apply_lane_speed_limits, + assign_semantic_tags, + effective_speed_limit_ranges, + lane_scenic_tags, makeCurve, - parse_speed_to_mps, - scenic_tags_from_type_records, - speed_limit_from_type_records, + merge_scenic_tags, + propagate_speed_limit, + propagate_tags, + speed_limit_for_s_interval, + speed_limit_ranges_from_lane_records, + speed_limit_ranges_from_type_records, + speed_to_mps, ) oldDir = os.getcwd() @@ -43,20 +54,12 @@ def test_map(path, runLocally, pytestconfig): plt.close() -DEFAULT_PLAN_VIEW = """ - - - - """ - - -def write_xodr(tmp_path, plan_view, road_extras=""): +def write_xodr(tmp_path, plan_view): path = tmp_path / "test.xodr" path.write_text( f""" - {road_extras} {plan_view} @@ -78,14 +81,6 @@ def write_xodr(tmp_path, plan_view, road_extras=""): return path -def parse_scenic_network(tmp_path, road_extras="", plan_view=DEFAULT_PLAN_VIEW): - path = write_xodr(tmp_path, plan_view, road_extras=road_extras) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - return road_map.toScenicNetwork() - - def test_inconsistent_planview_length_warns(tmp_path): path = write_xodr( tmp_path, @@ -199,130 +194,598 @@ def test_make_curve_rejects_unknown_geometry_type(): makeCurve(0.0, 0.0, 0.0, 10.0, curve_elem) +DEFAULT_PLAN_VIEW = """ + + + + """ + + +def write_xodr_with_type(tmp_path, plan_view=DEFAULT_PLAN_VIEW, road_extras=""): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + + {road_extras} + {plan_view} + + + +
+ +
+ + + + + + + + + +""" + ) + return path + + +def parse_scenic_network(tmp_path, road_extras="", plan_view=DEFAULT_PLAN_VIEW): + path = write_xodr_with_type(tmp_path, plan_view, road_extras=road_extras) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + return road_map.toScenicNetwork() + + +def type_tags_from_road_extras(road_extras): + root = ET.fromstring(f"{road_extras}") + return frozenset(elem.get("type") for elem in root.iter("type") if elem.get("type")) + + +def assert_road_level_tags_propagated(road): + """Road-level tags should match lane groups and road sections only.""" + expected = road.tags + if road.forwardLanes is not None: + assert road.forwardLanes.tags == expected + if road.backwardLanes is not None: + assert road.backwardLanes.tags == expected + for section in road.sections: + assert section.tags == expected + + +def write_xodr_multi_lane(tmp_path, lanes_xml, road_extras="", junction_xml=""): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + {junction_xml} + + {road_extras} + {DEFAULT_PLAN_VIEW} + + + +
+ +
+ +{lanes_xml} + + + + + +""" + ) + return path + + +def _linear_element_stub(cls): + element = object.__new__(cls) + element.speedLimit = None + element.tags = frozenset() + return element + + @pytest.mark.parametrize( - "max_attr, unit, expected_mps", + "cls", [ - ("25", "m/s", 25.0), - ("90", "km/h", 90 / 3.6), - ("55", "mph", 55 * 0.44704), + roadDomain.RoadSection, + roadDomain.LaneGroup, ], ) -def test_parse_speed_to_mps_converts_units(max_attr, unit, expected_mps): - speed_elem = ET.fromstring(f'') - assert parse_speed_to_mps(speed_elem) == pytest.approx(expected_mps) +def test_propagate_tags(cls): + element = _linear_element_stub(cls) + tags = frozenset({"from_map"}) + propagate_tags(tags, [element, "ignored"]) + assert element.tags == tags -@pytest.mark.parametrize("max_attr", ["no limit", "undefined"]) -def test_parse_speed_to_mps_unlimited(max_attr): - speed_elem = ET.fromstring(f'') - assert parse_speed_to_mps(speed_elem) is None +def test_propagate_tags_skips_lane_elements(): + lane_section = _linear_element_stub(roadDomain.LaneSection) + lane = _linear_element_stub(roadDomain.Lane) + tags = frozenset({"from_map"}) + propagate_tags(tags, [lane_section, lane]) + assert lane_section.tags == frozenset() + assert lane.tags == frozenset() -def test_parse_speed_to_mps_rejects_unknown_unit(): - speed_elem = ET.fromstring('') - with pytest.raises(ValueError, match="unsupported speed unit"): - parse_speed_to_mps(speed_elem) +def test_propagate_tags_skips_empty(): + element = _linear_element_stub(roadDomain.LaneSection) + propagate_tags(frozenset(), [element]) + assert element.tags == frozenset() -def test_speed_limit_from_type_records(): - records = [(0.0, "town", 13.89), (10.0, "motorway", 27.78)] - assert speed_limit_from_type_records(records) == pytest.approx(13.89) +def test_propagate_tags_ignores_other_types(): + element = _linear_element_stub(roadDomain.RoadSection) + tags = frozenset({"from_map"}) + propagate_tags(tags, [object(), element]) + assert element.tags == tags -def test_speed_limit_from_type_records_empty(): - assert speed_limit_from_type_records([]) is None +def test_propagate_tags_does_not_overwrite_existing(): + element = _linear_element_stub(roadDomain.Lane) + existing = frozenset({"existing"}) + element.tags = existing + propagate_tags(frozenset({"from_map"}), [element]) + assert element.tags == existing -def test_scenic_tags_from_type_records_maps_motorway_to_highway(): - records = [(0.0, "motorway", None)] - assert scenic_tags_from_type_records(records) == frozenset({"highway"}) +def test_map_tags_propagate_to_lane_hierarchy(tmp_path): + road_extras = '' + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = network.roads[0] + assert road.tags == type_tags_from_road_extras(road_extras) + assert_road_level_tags_propagated(road) + assert road.lanes[0].tags == frozenset({"driving"}) + assert road.lanes[0].sections[0].tags == road.lanes[0].tags + + +def test_map_tags_from_all_type_segments_propagate(tmp_path): + road_extras = ( + '' + '' + ) + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = network.roads[0] + assert road.tags == type_tags_from_road_extras(road_extras) + assert_road_level_tags_propagated(road) + assert road.lanes[0].tags == frozenset({"driving"}) -def test_scenic_tags_from_type_records_passes_through_unknown_types(): - records = [(0.0, "town", None)] - assert scenic_tags_from_type_records(records) == frozenset({"town"}) +def test_lane_type_tags_are_lane_specific(tmp_path): + lanes_xml = """ + + + + + """ + road_extras = '' + path = write_xodr_multi_lane(tmp_path, lanes_xml, road_extras=road_extras) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + road = road_map.toScenicNetwork().roads[0] + tags_by_od_id = { + section.openDriveID: section.tags for section in road.sections[0].lanes + } + assert tags_by_od_id[-1] == frozenset({"driving"}) + assert tags_by_od_id[-2] == frozenset({"onRamp"}) + assert "onRamp" not in tags_by_od_id[-1] + assert "driving" not in tags_by_od_id[-2] + assert road.tags == frozenset({"motorway"}) -def test_scenic_tags_from_type_records_empty(): - assert scenic_tags_from_type_records([]) == frozenset() +def test_junction_ramp_tags_apply_only_to_ramp_lanes(tmp_path): + lanes_xml = """ + + + + + """ + road_extras = '' + path = write_xodr_multi_lane(tmp_path, lanes_xml, road_extras=road_extras) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + internal_road = road_map.roads[7] + internal_road.extra_tags = frozenset({"onramp"}) + road, _ = internal_road.toScenicRoad(road_map.tolerance) + + assert road.tags == frozenset({"motorway", "onramp"}) + tags_by_od_id = { + section.openDriveID: section.tags for section in road.sections[0].lanes + } + assert tags_by_od_id[-1] == frozenset({"driving"}) + assert tags_by_od_id[-2] == frozenset({"onRamp"}) + assert "onramp" not in tags_by_od_id[-1] + assert "onramp" not in tags_by_od_id[-2] + assert "motorway" not in tags_by_od_id[-1] + + +def test_lane_scenic_tags_uses_open_drive_values_as_is(): + assert lane_scenic_tags("onRamp") == frozenset({"onRamp"}) + assert lane_scenic_tags("driving") == frozenset({"driving"}) + assert lane_scenic_tags(None) == frozenset() + + +def test_speed_limit_ranges_from_type_records(): + records = [ + (0.0, "town", 45.0), + (7.0, "town", 30.0), + (10.0, "town", 45.0), + (15.0, "town", 45.0), + ] + assert speed_limit_ranges_from_type_records(records, 30.0) == [ + (0.0, 7.0, 45.0), + (7.0, 10.0, 30.0), + (10.0, 30.0, 45.0), + ] + + +def test_speed_limit_ranges_carry_forward_missing_speed(): + records = [ + (0.0, "town", 45.0), + (10.0, "motorway", None), + (20.0, "motorway", 30.0), + ] + assert speed_limit_ranges_from_type_records(records, 30.0) == [ + (0.0, 20.0, 45.0), + (20.0, 30.0, 30.0), + ] + + +def test_speed_limit_for_s_interval_single_and_multiple(): + ranges = [ + (0.0, 7.0, 45.0), + (7.0, 10.0, 30.0), + (10.0, 30.0, 45.0), + ] + speed, overlapping = speed_limit_for_s_interval(ranges, 10.0, 20.0) + assert speed == pytest.approx(45.0) + assert overlapping == frozenset({45.0}) + + speed, overlapping = speed_limit_for_s_interval(ranges, 0.0, 10.0) + assert speed == pytest.approx(30.0) + assert overlapping == frozenset({45.0, 30.0}) + + +def test_speed_limit_ranges_from_lane_records(): + records = [(0.0, 20.0), (5.0, 30.0)] + assert speed_limit_ranges_from_lane_records(records, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_speed_limit_ranges_from_lane_records_carry_forward(): + records = [(0.0, 20.0), (5.0, None), (8.0, 30.0)] + assert speed_limit_ranges_from_lane_records(records, 10.0) == [ + (0.0, 8.0, 20.0), + (8.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_lane_overrides_road(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 10.0, 80.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 10.0, 80.0), + ] + + +def test_effective_speed_limit_ranges_multiple_lane_speeds(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 5.0, 20.0), (5.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_lane_speed_below_road(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_road_only_uses_road_ranges(): + road_ranges = [(0.0, 5.0, 50.0), (5.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, [], 0.0, 10.0) == [ + (0.0, 5.0, 50.0), + (5.0, 10.0, 30.0), + ] + + +def test_parse_lane_speed_records(tmp_path): + lanes_xml = """ + + + + """ + path = write_xodr_lane_speeds(tmp_path, lanes_xml) + road_map = RoadMap() + road_map.parse(path) + lane = road_map.roads[7].lane_secs[0].get_lane(-1) + assert lane.speed_records == [(0.0, 20.0), (5.0, 30.0)] + assert lane.speed_limit == pytest.approx(20.0) + assert speed_limit_ranges_from_lane_records(lane.speed_records, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_lane_speed_ranges_on_scenic_lane_section(tmp_path): + lanes_xml = """ + + + + """ + path = write_xodr_lane_speeds(tmp_path, lanes_xml) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + lane_section = road_map.toScenicNetwork().roads[0].sections[0].lanes[0] -def test_road_type_records_parsed(tmp_path): - path = write_xodr( - tmp_path, - DEFAULT_PLAN_VIEW, - road_extras='', + assert lane_section.speedLimit == pytest.approx(20.0) + assert lane_section.speedLimitRanges == ( + (0.0, 5.0, 20.0), + (5.0, 20.0, 30.0), ) + assert lane_section.speed_limit_at(2.0) == pytest.approx(20.0) + assert lane_section.speed_limit_at(7.0) == pytest.approx(30.0) + + +def test_lane_speed_ranges_demo_map(getAssetPath): + path = getAssetPath("maps/demo/tags/09_lane_speed_ranges.xodr") road_map = RoadMap() road_map.parse(path) - s, road_type, speed_mps = road_map.roads[7].type_records[0] - assert s == pytest.approx(0.0) - assert road_type == "town" - assert speed_mps == pytest.approx(50 / 3.6) + road_map.calculate_geometry(num=5, calc_intersect=True) + lane_section = road_map.toScenicNetwork().roads[0].sections[0].lanes[0] + assert lane_section.speedLimit == pytest.approx(20.0) + assert lane_section.speedLimitRanges == ( + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ) -def test_scenic_road_speed_limit_and_tags(tmp_path): - network = parse_scenic_network( - tmp_path, - road_extras='', + +def test_propagate_speed_limit_sets_section_specific_values(): + road_section = _linear_element_stub(roadDomain.RoadSection) + lane_section = _linear_element_stub(roadDomain.LaneSection) + propagate_speed_limit(30.0, [road_section, lane_section]) + assert road_section.speedLimit == pytest.approx(30.0) + assert lane_section.speedLimit == pytest.approx(30.0) + + other_section = _linear_element_stub(roadDomain.RoadSection) + other_lane = _linear_element_stub(roadDomain.LaneSection) + propagate_speed_limit(45.0, [other_section, other_lane]) + assert other_section.speedLimit == pytest.approx(45.0) + assert other_lane.speedLimit == pytest.approx(45.0) + + +def test_road_speed_limits_vary_by_lane_section(): + speed_ranges = speed_limit_ranges_from_type_records( + [ + (0.0, "town", 45.0), + (7.0, "town", 30.0), + (10.0, "town", 45.0), + (15.0, "town", 45.0), + (20.0, "town", 45.0), + (30.0, "town", 45.0), + ], + 30.0, ) - road = network.roads[0] - expected = 50 / 3.6 - assert road.speedLimit == pytest.approx(expected) + section_intervals = [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)] + section_speeds = [ + speed_limit_for_s_interval(speed_ranges, s_start, s_end) + for s_start, s_end in section_intervals + ] + + assert section_speeds[0] == (pytest.approx(30.0), frozenset({45.0, 30.0})) + assert section_speeds[1] == (pytest.approx(45.0), frozenset({45.0})) + assert section_speeds[2] == (pytest.approx(45.0), frozenset({45.0})) + assert min(speed for speed, _ in section_speeds) == pytest.approx(30.0) + + +def write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=""): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + + {road_extras} + {DEFAULT_PLAN_VIEW} + + + +
+ +
+ +{lanes_xml} + + + + + +""" + ) + return path + + +def test_demo_lane_speed_limits_map(getAssetPath): + path = getAssetPath("maps/demo/tags/08_lane_speed_limits.xodr") + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + road = road_map.toScenicNetwork().roads[0] + + road_limit = 50 / 3.6 + fast_limit = 80 / 3.6 + assert road.speedLimit == pytest.approx(road_limit) assert road.tags == frozenset({"town"}) + limits_by_od_id = { + section.openDriveID: section.speedLimit for section in road.sections[0].lanes + } + tags_by_od_id = { + section.openDriveID: section.tags for section in road.sections[0].lanes + } + assert limits_by_od_id[-1] == pytest.approx(road_limit) + assert limits_by_od_id[-2] == pytest.approx(fast_limit) + assert limits_by_od_id[-3] == pytest.approx(40 / 3.6) + assert tags_by_od_id[-1] == frozenset({"driving"}) + assert tags_by_od_id[-2] == frozenset({"driving"}) + assert tags_by_od_id[-3] == frozenset({"onRamp"}) + + +def test_lane_speed_limit_overrides_when_higher(tmp_path): + lanes_xml = """ + + + + + + + + + + """ + road_extras = '' + path = write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=road_extras) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + road = road_map.toScenicNetwork().roads[0] + road_limit = 50 / 3.6 + fast_limit = 80 / 3.6 + assert road.speedLimit == pytest.approx(road_limit) + limits_by_od_id = { + section.openDriveID: section.speedLimit for section in road.sections[0].lanes + } + assert limits_by_od_id[-1] == pytest.approx(road_limit) + assert limits_by_od_id[-2] == pytest.approx(fast_limit) + assert limits_by_od_id[-3] == pytest.approx(40 / 3.6) -def test_scenic_speed_limit_propagates_to_lane_hierarchy(tmp_path): - network = parse_scenic_network( - tmp_path, - road_extras='', + +def _speed_elem(max_value, unit=None): + attrs = f'max="{max_value}"' + if unit is not None: + attrs += f' unit="{unit}"' + return ET.fromstring(f"") + + +def test_speed_to_mps_unit_conversions(): + assert speed_to_mps(_speed_elem(36, "km/h")) == pytest.approx(10.0) + assert speed_to_mps(_speed_elem(10, "m/s")) == pytest.approx(10.0) + assert speed_to_mps(_speed_elem(100, "mph")) == pytest.approx(44.704) + # A missing unit defaults to m/s. + assert speed_to_mps(_speed_elem(15)) == pytest.approx(15.0) + + +def test_speed_to_mps_unlimited_returns_none(): + assert speed_to_mps(_speed_elem("no limit", "km/h")) is None + assert speed_to_mps(_speed_elem("undefined")) is None + + +def test_speed_to_mps_rejects_unknown_unit(): + with pytest.raises(ValueError, match="unsupported speed unit"): + speed_to_mps(_speed_elem(50, "furlongs/fortnight")) + + +def test_merge_scenic_tags_combines_dedups_and_ignores_empty(): + assert merge_scenic_tags(frozenset({"a"}), frozenset({"a", "b"})) == frozenset( + {"a", "b"} ) - road = network.roads[0] - expected = 36 / 3.6 - assert road.forwardLanes.speedLimit == pytest.approx(expected) - assert road.sections[0].speedLimit == pytest.approx(expected) - assert road.lanes[0].speedLimit == pytest.approx(expected) - lane_section = road.lanes[0].sections[0] - assert lane_section.speedLimit == pytest.approx(expected) + assert merge_scenic_tags(frozenset(), None, frozenset({"x"})) == frozenset({"x"}) + assert merge_scenic_tags() == frozenset() -def test_scenic_motorway_tag_mapped_to_highway(tmp_path): - network = parse_scenic_network( - tmp_path, - road_extras='', +def test_junction_tags_from_type_and_name(): + assert _junction_tags(Junction(1, "Roundabout A", "roundabout")) == frozenset( + {"roundabout", "Roundabout A"} ) - road = network.roads[0] - assert road.tags == frozenset({"highway"}) - assert road.speedLimit == pytest.approx(120 / 3.6) + # The default junction type carries no semantic meaning and is dropped. + assert _junction_tags(Junction(2, None, "default")) == frozenset() + # type_ defaults to "default", so only the name remains. + assert _junction_tags(Junction(3, "J3")) == frozenset({"J3"}) -def test_scenic_road_without_type_has_no_limit_or_tags(tmp_path): - network = parse_scenic_network(tmp_path) - road = network.roads[0] - assert road.speedLimit is None - assert road.tags == frozenset() +class _FakeRoad: + def __init__(self, junction): + self.junction = junction + self.extra_tags = frozenset() -def test_scenic_road_no_limit_speed(tmp_path): - network = parse_scenic_network( - tmp_path, - road_extras='', +class _FakeRoadMap: + def __init__(self, roads, junctions): + self.roads = roads + self.junctions = junctions + + +def test_assign_semantic_tags_tags_only_roads_in_junction(): + road_map = _FakeRoadMap( + roads={10: _FakeRoad(junction=5), 11: _FakeRoad(junction=None)}, + junctions={5: Junction(5, "J5", "roundabout")}, ) - road = network.roads[0] - assert road.speedLimit is None - assert road.tags == frozenset({"town"}) + assign_semantic_tags(road_map) + assert road_map.roads[10].extra_tags == frozenset({"roundabout", "J5"}) + assert road_map.roads[11].extra_tags == frozenset() -def test_multiple_type_segments_use_smallest_s(tmp_path): - road_extras = """ - - - """ - with pytest.warns( - OpenDriveWarning, - match="road 7 has 2 type segments", - ): - network = parse_scenic_network(tmp_path, road_extras=road_extras) - road = network.roads[0] - assert road.speedLimit == pytest.approx(30 / 3.6) - assert road.tags == frozenset({"town"}) +def test_multi_section_lane_speed_uses_max(getAssetPath): + path = getAssetPath("maps/demo/tags/10_multi_section_lane_speeds.xodr") + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + road = road_map.toScenicNetwork().roads[0] + + # Lane -1 is linked across three sections with lane speeds 10, 20, 15 m/s; + # apply_lane_speed_limits sets the merged Lane to the max while each lane + # section keeps its own value. + lane = road.lanes[0] + assert lane.speedLimit == pytest.approx(20.0) + section_limits = sorted(sec.speedLimit for sec in lane.sections) + assert section_limits == [ + pytest.approx(10.0), + pytest.approx(15.0), + pytest.approx(20.0), + ] + + +def test_apply_lane_speed_limits_ignores_non_lane_elements(): + lane = _linear_element_stub(roadDomain.Lane) + lane.sections = [ + _linear_element_stub(roadDomain.LaneSection), + _linear_element_stub(roadDomain.LaneSection), + ] + lane.sections[0].speedLimit = 10.0 + lane.sections[1].speedLimit = 25.0 + section = _linear_element_stub(roadDomain.RoadSection) + apply_lane_speed_limits([lane, section, "ignored"]) + assert lane.speedLimit == pytest.approx(25.0) + assert section.speedLimit is None + + +def test_section_spanning_multiple_speeds_warns(tmp_path): + road_extras = ( + '' + '' + ) + # The single 20 m lane section [0,20) straddles both the 50 and 30 km/h limits. + path = write_xodr_lane_speeds( + tmp_path, + """ + + """, + road_extras=road_extras, + ) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + with pytest.warns(OpenDriveWarning, match="spans multiple speed limits"): + road_map.toScenicNetwork() From 64301437dd72def56c218d91746a8e71753f5be2 Mon Sep 17 00:00:00 2001 From: aprabou Date: Thu, 2 Jul 2026 18:58:26 -0700 Subject: [PATCH 07/26] added speed:imitAt function to get the speed limit at coordinate *s* --- src/scenic/domains/driving/roads.py | 12 +++++ tests/formats/opendrive/test_opendrive.py | 63 +---------------------- 2 files changed, 14 insertions(+), 61 deletions(-) diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index f06377c5a..9b2502a0e 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -239,9 +239,21 @@ class NetworkElement(_ElementReferencer, PolygonalRegion): vehicleTypes: FrozenSet[VehicleType] = frozenset([VehicleType.CAR]) #: Optional speed limit, which may be inherited from parent. speedLimit: Union[float, None] = None + #: Optional ``(s_start, s_end, speed_mps)`` ranges along this element's centerline. + speedLimitRanges: Tuple[Tuple[float, float, float], ...] = () #: Uninterpreted semantic tags, e.g. 'roundabout'. tags: FrozenSet[str] = frozenset() + @distributionFunction + def speedLimitAt(self, s: float) -> Union[float, None]: + """Get the speed limit at coordinate *s* along this element, in meters.""" + if self.speedLimitRanges: + for range_start, range_end, speed in self.speedLimitRanges: + if range_start <= s < range_end: + return speed + return None + return self.speedLimit + def __attrs_post_init__(self): assert self.uid is not None or self.id is not None if self.uid is None: diff --git a/tests/formats/opendrive/test_opendrive.py b/tests/formats/opendrive/test_opendrive.py index 3df71dd03..ad992d7ba 100644 --- a/tests/formats/opendrive/test_opendrive.py +++ b/tests/formats/opendrive/test_opendrive.py @@ -535,22 +535,8 @@ def test_lane_speed_ranges_on_scenic_lane_section(tmp_path): (0.0, 5.0, 20.0), (5.0, 20.0, 30.0), ) - assert lane_section.speed_limit_at(2.0) == pytest.approx(20.0) - assert lane_section.speed_limit_at(7.0) == pytest.approx(30.0) - - -def test_lane_speed_ranges_demo_map(getAssetPath): - path = getAssetPath("maps/demo/tags/09_lane_speed_ranges.xodr") - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - lane_section = road_map.toScenicNetwork().roads[0].sections[0].lanes[0] - - assert lane_section.speedLimit == pytest.approx(20.0) - assert lane_section.speedLimitRanges == ( - (0.0, 5.0, 20.0), - (5.0, 10.0, 30.0), - ) + assert lane_section.speedLimitAt(2.0) == pytest.approx(20.0) + assert lane_section.speedLimitAt(7.0) == pytest.approx(30.0) def test_propagate_speed_limit_sets_section_specific_values(): @@ -617,31 +603,6 @@ def write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=""): return path -def test_demo_lane_speed_limits_map(getAssetPath): - path = getAssetPath("maps/demo/tags/08_lane_speed_limits.xodr") - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - road = road_map.toScenicNetwork().roads[0] - - road_limit = 50 / 3.6 - fast_limit = 80 / 3.6 - assert road.speedLimit == pytest.approx(road_limit) - assert road.tags == frozenset({"town"}) - limits_by_od_id = { - section.openDriveID: section.speedLimit for section in road.sections[0].lanes - } - tags_by_od_id = { - section.openDriveID: section.tags for section in road.sections[0].lanes - } - assert limits_by_od_id[-1] == pytest.approx(road_limit) - assert limits_by_od_id[-2] == pytest.approx(fast_limit) - assert limits_by_od_id[-3] == pytest.approx(40 / 3.6) - assert tags_by_od_id[-1] == frozenset({"driving"}) - assert tags_by_od_id[-2] == frozenset({"driving"}) - assert tags_by_od_id[-3] == frozenset({"onRamp"}) - - def test_lane_speed_limit_overrides_when_higher(tmp_path): lanes_xml = """ @@ -737,26 +698,6 @@ def test_assign_semantic_tags_tags_only_roads_in_junction(): assert road_map.roads[11].extra_tags == frozenset() -def test_multi_section_lane_speed_uses_max(getAssetPath): - path = getAssetPath("maps/demo/tags/10_multi_section_lane_speeds.xodr") - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - road = road_map.toScenicNetwork().roads[0] - - # Lane -1 is linked across three sections with lane speeds 10, 20, 15 m/s; - # apply_lane_speed_limits sets the merged Lane to the max while each lane - # section keeps its own value. - lane = road.lanes[0] - assert lane.speedLimit == pytest.approx(20.0) - section_limits = sorted(sec.speedLimit for sec in lane.sections) - assert section_limits == [ - pytest.approx(10.0), - pytest.approx(15.0), - pytest.approx(20.0), - ] - - def test_apply_lane_speed_limits_ignores_non_lane_elements(): lane = _linear_element_stub(roadDomain.Lane) lane.sections = [ From baf9acbe270002b58089e33ced9b2df491cb6019 Mon Sep 17 00:00:00 2001 From: aprabou Date: Thu, 2 Jul 2026 23:41:53 -0700 Subject: [PATCH 08/26] fixed sample1.1.xodr: root cause was a type-mismatch in definition of junction in Road class. junction is now a raw string and later converted in assign_semantic_tags --- src/scenic/formats/opendrive/xodr_parser.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 55e6cba55..42fa9fcf3 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -215,7 +215,9 @@ def assign_semantic_tags(road_map): if not tags: continue for road_id, road in road_map.roads.items(): - if road.junction == jid: + # road.junction is the raw OpenDRIVE id string (or None); junctions + # are keyed by int, so convert before comparing. + if road.junction is not None and int(road.junction) == jid: extra[road_id].update(tags) for road_id, road in road_map.roads.items(): @@ -632,7 +634,7 @@ def __init__(self, name, id_, length, junction, drive_on_right=True): self.name = name self.id_ = id_ self.length = length - self.junction = int(junction) if junction != "-1" else None + self.junction = junction if junction != "-1" else None self.predecessor = None self.successor = None self.signals = [] # List of Signal objects. @@ -1580,7 +1582,6 @@ def __init__( drivable_lane_types=( "driving", "stop", - "parking", "entry", "exit", "onRamp", @@ -1595,6 +1596,7 @@ def __init__( "border", "restricted", "curb", + "parking", "none", ), elide_short_roads=False, From 171e8e0b91984f6af0fdec4c2efac2d28fb59203 Mon Sep 17 00:00:00 2001 From: aprabou Date: Tue, 7 Jul 2026 21:33:53 -0700 Subject: [PATCH 09/26] modify speedLimit: no longer returns None; s value at or beyond the end of the range is clamped to the end via last_speed; clamp negative values to zero at the top --- src/scenic/domains/driving/roads.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index 9b2502a0e..3154b79fa 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -237,7 +237,7 @@ class NetworkElement(_ElementReferencer, PolygonalRegion): #: Which types of vehicles (car, bicycle, etc.) can be here. vehicleTypes: FrozenSet[VehicleType] = frozenset([VehicleType.CAR]) - #: Optional speed limit, which may be inherited from parent. + #: Optional speed limit, which may be inherited from parent. (deprecated) speedLimit: Union[float, None] = None #: Optional ``(s_start, s_end, speed_mps)`` ranges along this element's centerline. speedLimitRanges: Tuple[Tuple[float, float, float], ...] = () @@ -247,12 +247,16 @@ class NetworkElement(_ElementReferencer, PolygonalRegion): @distributionFunction def speedLimitAt(self, s: float) -> Union[float, None]: """Get the speed limit at coordinate *s* along this element, in meters.""" + s = max(0, s) if self.speedLimitRanges: + last_speed = None for range_start, range_end, speed in self.speedLimitRanges: if range_start <= s < range_end: return speed - return None + last_speed = speed + return last_speed return self.speedLimit + def __attrs_post_init__(self): assert self.uid is not None or self.id is not None From 84f4014faf93e11118ac216cc58c59bc9675a5a2 Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 8 Jul 2026 00:43:12 -0700 Subject: [PATCH 10/26] Respecting None value as valid speedLimit, modifying relative functions to abide --- src/scenic/domains/driving/roads.py | 9 ++--- src/scenic/formats/opendrive/xodr_parser.py | 41 +++++++++++---------- 2 files changed, 26 insertions(+), 24 deletions(-) diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index 3154b79fa..48744567c 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -237,7 +237,7 @@ class NetworkElement(_ElementReferencer, PolygonalRegion): #: Which types of vehicles (car, bicycle, etc.) can be here. vehicleTypes: FrozenSet[VehicleType] = frozenset([VehicleType.CAR]) - #: Optional speed limit, which may be inherited from parent. (deprecated) + #: Optional speed limit, which may be inherited from parent. (depreciated) speedLimit: Union[float, None] = None #: Optional ``(s_start, s_end, speed_mps)`` ranges along this element's centerline. speedLimitRanges: Tuple[Tuple[float, float, float], ...] = () @@ -249,12 +249,11 @@ def speedLimitAt(self, s: float) -> Union[float, None]: """Get the speed limit at coordinate *s* along this element, in meters.""" s = max(0, s) if self.speedLimitRanges: - last_speed = None for range_start, range_end, speed in self.speedLimitRanges: if range_start <= s < range_end: - return speed - last_speed = speed - return last_speed + return speed # may be None -> genuine "no limit" stretch + # s is at/beyond the final (half-open) range's end: clamp to it. + return speed return self.speedLimit diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 42fa9fcf3..0a228077a 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -52,24 +52,30 @@ def speed_to_mps(speed_elem): def _speed_limit_ranges_from_s_speed_records(records, domain_length): - """Build ``(s_start, s_end, speed_mps)`` ranges from ``[(s, speed_mps), ...]``.""" + """Build ``(s_start, s_end, speed_mps)`` ranges from ``[(s, speed_mps), ...]``. + + Coverage is total over ``[0, domain_length)``: each record keeps its own + speed (no inheritance from a previous record), and stretches with no defined + limit are emitted as explicit ``None`` ranges. This lets ``speedLimitAt`` + distinguish a genuine "no limit" stretch from a lookup past the end. + """ if not records: return [] sorted_records = sorted(records, key=lambda record: record[0]) - ranges = [] - current_speed = None + # Guarantee coverage from 0: an undefined leading stretch stays None. + if sorted_records[0][0] > 0: + sorted_records = [(0.0, None)] + sorted_records - for i, (s, speed_mps) in enumerate(sorted_records): - if speed_mps is not None: - current_speed = speed_mps + ranges = [] + for i, (s, speed_mps) in enumerate (sorted_records): s_end = sorted_records[i + 1][0] if i + 1 < len(sorted_records) else domain_length - if s_end <= s or current_speed is None: + if s_end <= s: continue - if ranges and ranges[-1][2] == current_speed and ranges[-1][1] == s: - ranges[-1] = (ranges[-1][0], s_end, current_speed) + if ranges and ranges[-1][2] == speed_mps and ranges[-1][1] == s: + ranges[-1] = (ranges[-1][0], s_end, speed_mps) else: - ranges.append((s, s_end, current_speed)) + ranges.append((s, s_end, speed_mps)) return ranges @@ -80,12 +86,8 @@ def speed_limit_ranges_from_type_records(type_records, road_length): return [] sorted_records = sorted(type_records, key=lambda record: record[0]) - s_speed_records = [] - current_speed = None - for s, _road_type, speed_mps in sorted_records: - if speed_mps is not None: - current_speed = speed_mps - s_speed_records.append((s, current_speed)) + # Each keeps its own speed (may be None); no inheritance across types. + s_speed_records = [(s, speed_mps) for s, _road_type, speed_mps in sorted_records] return _speed_limit_ranges_from_s_speed_records(s_speed_records, road_length) @@ -146,7 +148,7 @@ def speed_limit_for_s_interval(ranges, s_start, s_end): overlapping = { speed for range_start, range_end, speed in ranges - if range_end > s_start and range_start < s_end + if range_end > s_start and range_start < s_end and speed is not None } if not overlapping: @@ -164,9 +166,10 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): return element.speedLimit = min(speeds) - if len(speeds) > 1: + has_gap = any(speed is None for _, _, speed in ranges) + if len(speeds) > 1 or has_gap: element.speedLimitRanges = tuple(ranges) - if warn_context is not None: + if len(speeds) > 1 and warn_context is not None: speeds_text = ", ".join(f"{speed:.4g} m/s" for speed in sorted(speeds)) warn( f"{warn_context}: spans multiple speed limits {{{speeds_text}}};" From 462e1dde5cc47a9e7f5ff7c82c9a6b2588eb2855 Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 8 Jul 2026 00:50:55 -0700 Subject: [PATCH 11/26] removed redundant sorting + unnecessary code/comments --- src/scenic/formats/opendrive/xodr_parser.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 0a228077a..c0cd60608 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -51,7 +51,7 @@ def speed_to_mps(speed_elem): raise ValueError(f"unsupported speed unit: {unit!r}") -def _speed_limit_ranges_from_s_speed_records(records, domain_length): +def _speed_limit_ranges_from_s_speed_records(sorted_records, domain_length): """Build ``(s_start, s_end, speed_mps)`` ranges from ``[(s, speed_mps), ...]``. Coverage is total over ``[0, domain_length)``: each record keeps its own @@ -59,10 +59,7 @@ def _speed_limit_ranges_from_s_speed_records(records, domain_length): limit are emitted as explicit ``None`` ranges. This lets ``speedLimitAt`` distinguish a genuine "no limit" stretch from a lookup past the end. """ - if not records: - return [] - - sorted_records = sorted(records, key=lambda record: record[0]) + # Already sorted in speed_limit_ranges_from_type_records # Guarantee coverage from 0: an undefined leading stretch stays None. if sorted_records[0][0] > 0: sorted_records = [(0.0, None)] + sorted_records @@ -162,8 +159,6 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): return speeds = {speed for _, _, speed in ranges if speed is not None} - if not speeds: - return element.speedLimit = min(speeds) has_gap = any(speed is None for _, _, speed in ranges) @@ -1867,7 +1862,6 @@ def parse(self, path): else: pred_link = succ_link = None - # 2.3 Handle type and speed parsing for type_elem in r.findall("type"): s = float(type_elem.get("s")) road_type = type_elem.get("type") From 8bacd2672c4b80bcba5dbb10bdfa9ea865741739 Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 8 Jul 2026 01:54:16 -0700 Subject: [PATCH 12/26] remove unncessary guardrail and crash fixes --- src/scenic/formats/opendrive/xodr_parser.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index c0cd60608..c826693ee 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -94,6 +94,9 @@ def speed_limit_ranges_from_lane_records(speed_records, section_length): ``speed_records`` are ``(sOffset, speed_mps)`` pairs relative to the lane section start; ``section_length`` is the lane section length in meters. """ + if not speed_records: + return [] + return _speed_limit_ranges_from_s_speed_records(speed_records, section_length) @@ -138,7 +141,7 @@ def speed_limit_for_s_interval(ranges, s_start, s_end): ``speed_limit`` is their minimum -- the conservative choice when an interval straddles more than one limit. Both are empty/``None`` when nothing overlaps. """ - if not ranges or s_end <= s_start: + if not ranges: return None, frozenset() # A range overlaps the interval iff it starts before s_end and ends after s_start. @@ -1116,7 +1119,9 @@ def toScenicRoad(self, tolerance): effective_ranges = effective_speed_limit_ranges( speed_ranges, lane_ranges, s_start, section_length ) - if effective_ranges: + if not effective_ranges: + propagate_speed_limit(section_speed_limit, [lane_section]) + elif any(speed is not None for _, _, speed in effective_ranges): assign_speed_limit_from_ranges( lane_section, effective_ranges, @@ -1124,8 +1129,7 @@ def toScenicRoad(self, tolerance): f"road {self.id_} lane {id_} section s=[{s_start},{s_end})" ), ) - else: - propagate_speed_limit(section_speed_limit, [lane_section]) + # else: ranges exist but define no limit; leave lane section as-is last_section = section From 34eac7035257d108d6668f3b777b5ab1fad76088 Mon Sep 17 00:00:00 2001 From: aprabou Date: Wed, 8 Jul 2026 02:19:58 -0700 Subject: [PATCH 13/26] make small changes to *_lane_type definitions --- src/scenic/formats/opendrive/xodr_parser.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index c826693ee..a1e45b5e7 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -1583,7 +1583,6 @@ def __init__( fill_intersections=True, drivable_lane_types=( "driving", - "stop", "entry", "exit", "onRamp", @@ -1597,9 +1596,9 @@ def __init__( "shoulder", "border", "restricted", - "curb", "parking", "none", + "stop", ), elide_short_roads=False, ): From 9abb699a1e125f72740af2ca5f975d0ed009b274 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 15:45:03 -0700 Subject: [PATCH 14/26] refactor and change speedLimitRanges to always have a meaningful value --- src/scenic/formats/opendrive/xodr_parser.py | 44 ++++++++------------- 1 file changed, 17 insertions(+), 27 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index a1e45b5e7..8ec7da83a 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -133,27 +133,20 @@ def effective_speed_limit_ranges(road_ranges, lane_ranges, section_s0, section_l ] -def speed_limit_for_s_interval(ranges, s_start, s_end): - """Speed limit applying to the interval ``[s_start, s_end)``. +def speed_limits_for_s_interval(ranges, s_start, s_end): + """Speed limits applying to the interval ``[s_start, s_end)``. - Returns ``(speed_limit, overlapping_speeds)``, where ``overlapping_speeds`` is - the set of every limit in ``ranges`` that overlaps the interval and - ``speed_limit`` is their minimum -- the conservative choice when an interval - straddles more than one limit. Both are empty/``None`` when nothing overlaps. + Returns a frozenset of every limit in ``ranges`` that overlaps the interval. """ if not ranges: - return None, frozenset() + return frozenset() # A range overlaps the interval iff it starts before s_end and ends after s_start. - overlapping = { + return frozenset( speed for range_start, range_end, speed in ranges if range_end > s_start and range_start < s_end and speed is not None - } - - if not overlapping: - return None, frozenset() - return min(overlapping), frozenset(overlapping) + ) def assign_speed_limit_from_ranges(element, ranges, warn_context=None): @@ -164,17 +157,13 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): speeds = {speed for _, _, speed in ranges if speed is not None} element.speedLimit = min(speeds) - has_gap = any(speed is None for _, _, speed in ranges) - if len(speeds) > 1 or has_gap: - element.speedLimitRanges = tuple(ranges) - if len(speeds) > 1 and warn_context is not None: - speeds_text = ", ".join(f"{speed:.4g} m/s" for speed in sorted(speeds)) - warn( - f"{warn_context}: spans multiple speed limits {{{speeds_text}}};" - f" using minimum {element.speedLimit:.4g} m/s" - ) - else: - element.speedLimitRanges = () + element.speedLimitRanges = tuple(ranges) + if len(speeds) > 1 and warn_context is not None: + speeds_text = ", ".join(f"{speed:.4g} m/s" for speed in sorted(speeds)) + warn( + f"{warn_context}: spans multiple speed limits {{{speeds_text}}};" + f" using minimum {element.speedLimit:.4g} m/s" + ) def merge_scenic_tags(*tag_sets): @@ -1045,9 +1034,12 @@ def toScenicRoad(self, tolerance): if sec_index + 1 < len(self.lane_secs) else self.length ) - section_speed_limit, overlapping_speeds = speed_limit_for_s_interval( + overlapping_speeds = speed_limits_for_s_interval( speed_ranges, s_start, s_end ) + section_speed_limit = ( + min(overlapping_speeds) if overlapping_speeds else None + ) if len(overlapping_speeds) > 1: speeds_text = ", ".join( f"{speed:.4g} m/s" for speed in sorted(overlapping_speeds) @@ -1589,7 +1581,6 @@ def __init__( "offRamp", "connectingRamp", "slipLane", - "biking", ), sidewalk_lane_types=("walking", "sidewalk"), # sidewalk deprecated shoulder_lane_types=( @@ -1597,7 +1588,6 @@ def __init__( "border", "restricted", "parking", - "none", "stop", ), elide_short_roads=False, From 6668665eccfec112a56cda78ae621e79ae11b00a Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 15:48:56 -0700 Subject: [PATCH 15/26] removed helper function for tag union --- src/scenic/formats/opendrive/xodr_parser.py | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 8ec7da83a..129627cd3 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -166,15 +166,6 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): ) -def merge_scenic_tags(*tag_sets): - """Combine tags from road ```` records and junction metadata.""" - tags = set() - for tag_set in tag_sets: - if tag_set: - tags.update(tag_set) - return frozenset(tags) - - def lane_scenic_tags(lane_type): """Tags for a Lane or LaneSection from the OpenDRIVE lane ``type`` attribute.""" if not lane_type: @@ -1013,7 +1004,7 @@ def toScenicRoad(self, tolerance): for _, open_drive_type, _ in self.type_records if open_drive_type ) - road_level_tags = merge_scenic_tags(type_tags, self.extra_tags) + road_level_tags = frozenset(type_tags | self.extra_tags) allElements = [] # Create lane and road sections roadSections = [] @@ -1082,7 +1073,7 @@ def toScenicRoad(self, tolerance): road=None, openDriveID=id_, isForward=id_ < 0, - tags=lane_scenic_tags(lane.type_), + tags=frozenset({lane.type_} if lane.type_ else ()), ) section._original_lane = lane laneSections[id_] = section From b5073ac999f60ee8a5b66e50a4f34a6d6745407d Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 16:12:25 -0700 Subject: [PATCH 16/26] drop helper function for applying lane speed limits; use min over lane sections rather than max; set Lane.speedLimit directly when lane is created. --- src/scenic/formats/opendrive/xodr_parser.py | 40 ++++++++++----------- 1 file changed, 18 insertions(+), 22 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 129627cd3..dbe5ab149 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -228,20 +228,6 @@ def propagate_speed_limit(speed_limit, elements): element.speedLimit = speed_limit -def apply_lane_speed_limits(elements): - """Set each Lane's scalar ``speedLimit`` to the max over its lane sections.""" - for element in elements: - if not isinstance(element, roadDomain.Lane): - continue - section_limits = [ - section.speedLimit - for section in element.sections - if section.speedLimit is not None - ] - if section_limits: - element.speedLimit = max(section_limits) - - def propagate_tags(tags, elements): """Copy road-level semantic tags to lane groups and road sections.""" if not tags: @@ -1305,6 +1291,14 @@ def makeShoulder(sections, backward=False): adj.append(lane._laneToRight) lane.adjacentLanes = tuple(adj) + section_speed_limits = [ + section.speedLimit + for section in roadSections + if section.speedLimit is not None + ] + road_speed_limit = min(section_speed_limits) if section_speed_limits else None + propagate_speed_limit(road_speed_limit, allElements) + # Gather lane sections into lanes nextID = 0 forwardLanes, backwardLanes = [], [] @@ -1346,6 +1340,11 @@ def makeShoulder(sections, backward=False): leftEdge = PolylineRegion(cleanChain(leftPoints)) rightEdge = PolylineRegion(cleanChain(rightPoints)) centerline = PolylineRegion(cleanChain(centerPoints)) + lane_section_speed_limits = [ + section.speedLimit + for section in sections + if section.speedLimit is not None + ] lane = roadDomain.Lane( id=f"road{self.id_}_lane{nextID}", polygon=ls.parent_lane_poly, @@ -1357,6 +1356,11 @@ def makeShoulder(sections, backward=False): sections=tuple(sections), successor=successorLane, # will correct inter-road links later tags=frozenset().union(*(sec.tags for sec in sections)), + speedLimit=( + min(lane_section_speed_limits) + if lane_section_speed_limits + else None + ), ) nextID += 1 for section in sections: @@ -1468,13 +1472,6 @@ def getEdges(forward): leftEdge = forwardGroup.leftEdge centerline = PolylineRegion(tuple(pt[:2] for pt in self.ref_line_points)) - section_speed_limits = [ - section.speedLimit - for section in roadSections - if section.speedLimit is not None - ] - road_speed_limit = min(section_speed_limits) if section_speed_limits else None - road = roadDomain.Road( name=self.name, uid=f"road{self.id_}", # need prefix to prevent collisions with intersections @@ -1494,7 +1491,6 @@ def getEdges(forward): ) allElements.append(road) propagate_speed_limit(road_speed_limit, allElements) - apply_lane_speed_limits(allElements) propagate_tags(road_level_tags, allElements) # Set up parent references From a610ff1fe0153800b71f253ac51a30091077e2e1 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 16:16:24 -0700 Subject: [PATCH 17/26] removed propagate_tags and set road_level tags directly on relevant elements at creation time --- src/scenic/formats/opendrive/xodr_parser.py | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index dbe5ab149..77a13c3d7 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -228,16 +228,6 @@ def propagate_speed_limit(speed_limit, elements): element.speedLimit = speed_limit -def propagate_tags(tags, elements): - """Copy road-level semantic tags to lane groups and road sections.""" - if not tags: - return - for element in elements: - if isinstance(element, _ROAD_LEVEL_PROPAGATION_TYPES): - if not element.tags: - element.tags = tags - - def buffer_union(polys, tolerance=0.01): return polygonUnion(polys, buf=tolerance, tolerance=tolerance) @@ -1074,6 +1064,7 @@ def toScenicRoad(self, tolerance): predecessor=last_section, road=None, # will set later lanesByOpenDriveID=laneSections, + tags=road_level_tags, ) roadSections.append(section) allElements.append(section) @@ -1421,6 +1412,7 @@ def getEdges(forward): bikeLane=None, shoulder=forwardShoulder, opposite=None, + tags=road_level_tags, ) allElements.append(forwardGroup) else: @@ -1442,6 +1434,7 @@ def getEdges(forward): bikeLane=None, shoulder=backwardShoulder, opposite=forwardGroup, + tags=road_level_tags, ) allElements.append(backwardGroup) if forwardGroup: @@ -1491,7 +1484,6 @@ def getEdges(forward): ) allElements.append(road) propagate_speed_limit(road_speed_limit, allElements) - propagate_tags(road_level_tags, allElements) # Set up parent references if forwardGroup: From f60d83557590c2de200ee911183f529d0d06c8ee Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 16:27:32 -0700 Subject: [PATCH 18/26] drop self.speed_limit --- src/scenic/formats/opendrive/xodr_parser.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 77a13c3d7..053d73d0b 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -481,7 +481,6 @@ def __init__(self, id_, type_, pred=None, succ=None): self.id_ = id_ self.width = [] # List of tuples (Poly3, int) for width and s-offset. self.speed_records = [] # [(sOffset, speed_mps), ...] from elements. - self.speed_limit = None # First record's speed; kept for backward compatibility. self.type_ = type_ self.pred = pred self.succ = succ @@ -1727,8 +1726,6 @@ def __parse_lanes(self, lanes_elem): (float(speed_elem.get("sOffset", 0)), speed_to_mps(speed_elem)) ) lane.speed_records.sort(key=lambda record: record[0]) - if lane.speed_records: - lane.speed_limit = lane.speed_records[0][1] lanes[id_] = lane return lanes From 9b675c69375b544a95462cb7520051469ce1c75a Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 17:23:28 -0700 Subject: [PATCH 19/26] refactor assign_semantic_tags to compute in O(junctions + roads) --- src/scenic/formats/opendrive/xodr_parser.py | 39 ++++++--------------- 1 file changed, 11 insertions(+), 28 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 053d73d0b..aeebdb05c 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -173,36 +173,21 @@ def lane_scenic_tags(lane_type): return frozenset({lane_type}) -def _junction_tags(junction): - """Tags taken directly from a junction's OpenDRIVE ``type`` and ``name``.""" - tags = set() - if junction.type_ and junction.type_ != "default": - tags.add(junction.type_) - if junction.name: - tags.add(junction.name) - return tags - - def assign_semantic_tags(road_map): - """Populate each ``Road.extra_tags`` from the type/name of its junction. + """Populate each ``Road.extra_tags`` from the type of its junction. A road belonging to a junction (``road.junction`` equals the junction id) inherits that junction's semantic tags; all other roads get no extra tags. """ - extra = {road_id: set() for road_id in road_map.roads} - - for jid, junction in road_map.junctions.items(): - tags = _junction_tags(junction) - if not tags: - continue - for road_id, road in road_map.roads.items(): - # road.junction is the raw OpenDRIVE id string (or None); junctions - # are keyed by int, so convert before comparing. - if road.junction is not None and int(road.junction) == jid: - extra[road_id].update(tags) - - for road_id, road in road_map.roads.items(): - road.extra_tags = frozenset(extra[road_id]) + junction_tags = { + jid: junction.tags for jid, junction in road_map.junctions.items() + } + for road in road_map.roads.values(): + if road.junction is not None: + # road.junction is the raw OpenDRIVE id string; junctions are keyed by int. + road.extra_tags = junction_tags.get(int(road.junction), frozenset()) + else: + road.extra_tags = frozenset() _ROAD_LEVEL_PROPAGATION_TYPES = ( @@ -573,6 +558,7 @@ def __init__(self, id_, name, type_="default"): self.id_ = id_ self.name = name self.type_ = type_ + self.tags = frozenset({type_} if type_ and type_ != "default" else ()) self.connections = [] # Ids of roads that are paths within junction: self.paths = [] @@ -809,8 +795,6 @@ def calc_geometry_for_type(self, lane_types, num, tolerance, calc_gap=False): prev_id = id_ - 1 else: prev_id = id_ + 1 - if prev_id not in offsets: - continue if ( offsets[id_] == offsets[prev_id] @@ -1088,7 +1072,6 @@ def toScenicRoad(self, tolerance): f"road {self.id_} lane {id_} section s=[{s_start},{s_end})" ), ) - # else: ranges exist but define no limit; leave lane section as-is last_section = section From 142c6271782e2d18c8d422caa97ea4ee76be2248 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 17:35:33 -0700 Subject: [PATCH 20/26] removed propagate_speed_limit helper function and added inline propagation instead --- src/scenic/formats/opendrive/xodr_parser.py | 48 +++++++++------------ 1 file changed, 21 insertions(+), 27 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index aeebdb05c..9c2d83c29 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -190,29 +190,6 @@ def assign_semantic_tags(road_map): road.extra_tags = frozenset() -_ROAD_LEVEL_PROPAGATION_TYPES = ( - roadDomain.LaneGroup, - roadDomain.RoadSection, -) - - -_LANE_PROPAGATION_TYPES = ( - roadDomain.LaneSection, - roadDomain.Lane, -) -_ROAD_PROPAGATION_TYPES = _ROAD_LEVEL_PROPAGATION_TYPES + _LANE_PROPAGATION_TYPES - - -def propagate_speed_limit(speed_limit, elements): - """Copy a road speed limit down to lane groups, sections, and lanes.""" - if speed_limit is None: - return - for element in elements: - if isinstance(element, _ROAD_PROPAGATION_TYPES): - if element.speedLimit is None: - element.speedLimit = speed_limit - - def buffer_union(polys, tolerance=0.01): return polygonUnion(polys, buf=tolerance, tolerance=tolerance) @@ -1052,7 +1029,8 @@ def toScenicRoad(self, tolerance): roadSections.append(section) allElements.append(section) - propagate_speed_limit(section_speed_limit, [section]) + if section_speed_limit is not None and section.speedLimit is None: + section.speedLimit = section_speed_limit section_length = s_end - s_start for id_, lane_section in laneSections.items(): lane = sec.drivable_lanes[id_] @@ -1063,7 +1041,11 @@ def toScenicRoad(self, tolerance): speed_ranges, lane_ranges, s_start, section_length ) if not effective_ranges: - propagate_speed_limit(section_speed_limit, [lane_section]) + if ( + section_speed_limit is not None + and lane_section.speedLimit is None + ): + lane_section.speedLimit = section_speed_limit elif any(speed is not None for _, _, speed in effective_ranges): assign_speed_limit_from_ranges( lane_section, @@ -1270,7 +1252,13 @@ def makeShoulder(sections, backward=False): if section.speedLimit is not None ] road_speed_limit = min(section_speed_limits) if section_speed_limits else None - propagate_speed_limit(road_speed_limit, allElements) + if road_speed_limit is not None: + for section in roadSections: + if section.speedLimit is None: + section.speedLimit = road_speed_limit + for lane_section in section.lanesByOpenDriveID.values(): + if lane_section.speedLimit is None: + lane_section.speedLimit = road_speed_limit # Gather lane sections into lanes nextID = 0 @@ -1465,7 +1453,13 @@ def getEdges(forward): tags=road_level_tags, ) allElements.append(road) - propagate_speed_limit(road_speed_limit, allElements) + if road_speed_limit is not None: + for group in (forwardGroup, backwardGroup): + if group is not None and group.speedLimit is None: + group.speedLimit = road_speed_limit + for lane in lanes: + if lane.speedLimit is None: + lane.speedLimit = road_speed_limit # Set up parent references if forwardGroup: From 2947383e8c0e861485855b240bffe37982389cf4 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 22:04:28 -0700 Subject: [PATCH 21/26] add Network.speedLimit method --- src/scenic/domains/driving/roads.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index 48744567c..9bcb109e4 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -1308,6 +1308,24 @@ def laneSectionAt(self, point: Vectorlike, reject=False) -> Union[LaneSection, N lane = self.laneAt(point, reject=reject) return None if lane is None else lane.sectionAt(point) + @distributionMethod + def speedLimitAt(self, point: Vectorlike, reject=False) -> Union[float, None]: + """Get the speed limit at a given point, if any. + + Looks up the `LaneSection` containing the point and returns its speed + limit there. When the section has multiple ``speedLimitRanges``, the + limit is taken from the range covering the point's projected distance + along the section centerline. + """ + point = _toVector(point) + section = self.laneSectionAt(point, reject=reject) + if section is None: + return None + if len(section.speedLimitRanges) <= 1: + return section.speedLimit + s = section.centerline.lineString.project(geometry.makeShapelyPoint(point)) + return section.speedLimitAt(s) + @distributionMethod def laneGroupAt(self, point: Vectorlike, reject=False) -> Union[LaneGroup, None]: """Get the `LaneGroup` passing through a given point.""" From 45cbaf9f7b0d7a01748fa17c46f999aa563e6216 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 22:09:04 -0700 Subject: [PATCH 22/26] Refactor OpenDRIVE speed/tag parsing and split Network-level tests --- tests/formats/opendrive/conftest.py | 145 ++++++ tests/formats/opendrive/test_opendrive.py | 552 ---------------------- tests/formats/opendrive/test_speeds.py | 251 ++++++++++ tests/formats/opendrive/test_tags.py | 116 +++++ 4 files changed, 512 insertions(+), 552 deletions(-) create mode 100644 tests/formats/opendrive/conftest.py create mode 100644 tests/formats/opendrive/test_speeds.py create mode 100644 tests/formats/opendrive/test_tags.py diff --git a/tests/formats/opendrive/conftest.py b/tests/formats/opendrive/conftest.py new file mode 100644 index 000000000..fd90bd4ee --- /dev/null +++ b/tests/formats/opendrive/conftest.py @@ -0,0 +1,145 @@ +import xml.etree.ElementTree as ET + +from scenic.formats.opendrive.xodr_parser import Junction, RoadMap + +DEFAULT_PLAN_VIEW = """ + + + + """ + + +def write_xodr_with_type(tmp_path, plan_view=DEFAULT_PLAN_VIEW, road_extras=""): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + + {road_extras} + {plan_view} + + + +
+ +
+ + + + + + + + + +""" + ) + return path + + +def write_xodr_multi_lane( + tmp_path, lanes_xml, road_extras="", junction_xml="", junction_id="-1" +): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + {junction_xml} + + {road_extras} + {DEFAULT_PLAN_VIEW} + + + +
+ +
+ +{lanes_xml} + + + + + +""" + ) + return path + + +def write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=""): + path = tmp_path / "test.xodr" + path.write_text( + f""" + + + {road_extras} + {DEFAULT_PLAN_VIEW} + + + +
+ +
+ +{lanes_xml} + + + + + +""" + ) + return path + + +def parse_scenic_network( + tmp_path, + road_extras="", + plan_view=DEFAULT_PLAN_VIEW, + *, + lanes_xml=None, + junction_xml="", + junction_id="-1", + junction_type=None, + junction_name=None, +): + if lanes_xml is None: + path = write_xodr_with_type(tmp_path, plan_view, road_extras=road_extras) + else: + path = write_xodr_multi_lane( + tmp_path, + lanes_xml, + road_extras=road_extras, + junction_xml=junction_xml, + junction_id=junction_id, + ) + road_map = RoadMap() + road_map.parse(path) + road_map.calculate_geometry(num=5, calc_intersect=True) + if junction_id != "-1": + jid = int(junction_id) + if jid not in road_map.junctions: + road_map.junctions[jid] = Junction( + jid, junction_name, junction_type or "default" + ) + return road_map.toScenicNetwork() + + +def scenic_road(network, road_id=7): + return next(road for road in network.allRoads if road.id == road_id) + + +def type_tags_from_road_extras(road_extras): + root = ET.fromstring(f"{road_extras}") + return frozenset(elem.get("type") for elem in root.iter("type") if elem.get("type")) + + +def assert_road_level_tags_propagated(road): + """Road-level tags should match lane groups and road sections only.""" + expected = road.tags + if road.forwardLanes is not None: + assert road.forwardLanes.tags == expected + if road.backwardLanes is not None: + assert road.backwardLanes.tags == expected + for section in road.sections: + assert section.tags == expected diff --git a/tests/formats/opendrive/test_opendrive.py b/tests/formats/opendrive/test_opendrive.py index ad992d7ba..62b011c59 100644 --- a/tests/formats/opendrive/test_opendrive.py +++ b/tests/formats/opendrive/test_opendrive.py @@ -7,27 +7,13 @@ import pytest from scenic.core.geometry import TriangulationError -from scenic.domains.driving import roads as roadDomain from scenic.formats.opendrive import OpenDriveWorkspace from scenic.formats.opendrive.xodr_parser import ( Cubic, - Junction, OpenDriveWarning, ParamCubic, RoadMap, - _junction_tags, - apply_lane_speed_limits, - assign_semantic_tags, - effective_speed_limit_ranges, - lane_scenic_tags, makeCurve, - merge_scenic_tags, - propagate_speed_limit, - propagate_tags, - speed_limit_for_s_interval, - speed_limit_ranges_from_lane_records, - speed_limit_ranges_from_type_records, - speed_to_mps, ) oldDir = os.getcwd() @@ -192,541 +178,3 @@ def test_make_curve_rejects_unknown_geometry_type(): with pytest.raises(NotImplementedError, match="unhandled OpenDRIVE geometry type"): makeCurve(0.0, 0.0, 0.0, 10.0, curve_elem) - - -DEFAULT_PLAN_VIEW = """ - - - - """ - - -def write_xodr_with_type(tmp_path, plan_view=DEFAULT_PLAN_VIEW, road_extras=""): - path = tmp_path / "test.xodr" - path.write_text( - f""" - - - {road_extras} - {plan_view} - - - -
- -
- - - - - - - - - -""" - ) - return path - - -def parse_scenic_network(tmp_path, road_extras="", plan_view=DEFAULT_PLAN_VIEW): - path = write_xodr_with_type(tmp_path, plan_view, road_extras=road_extras) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - return road_map.toScenicNetwork() - - -def type_tags_from_road_extras(road_extras): - root = ET.fromstring(f"{road_extras}") - return frozenset(elem.get("type") for elem in root.iter("type") if elem.get("type")) - - -def assert_road_level_tags_propagated(road): - """Road-level tags should match lane groups and road sections only.""" - expected = road.tags - if road.forwardLanes is not None: - assert road.forwardLanes.tags == expected - if road.backwardLanes is not None: - assert road.backwardLanes.tags == expected - for section in road.sections: - assert section.tags == expected - - -def write_xodr_multi_lane(tmp_path, lanes_xml, road_extras="", junction_xml=""): - path = tmp_path / "test.xodr" - path.write_text( - f""" - - {junction_xml} - - {road_extras} - {DEFAULT_PLAN_VIEW} - - - -
- -
- -{lanes_xml} - - - - - -""" - ) - return path - - -def _linear_element_stub(cls): - element = object.__new__(cls) - element.speedLimit = None - element.tags = frozenset() - return element - - -@pytest.mark.parametrize( - "cls", - [ - roadDomain.RoadSection, - roadDomain.LaneGroup, - ], -) -def test_propagate_tags(cls): - element = _linear_element_stub(cls) - tags = frozenset({"from_map"}) - propagate_tags(tags, [element, "ignored"]) - assert element.tags == tags - - -def test_propagate_tags_skips_lane_elements(): - lane_section = _linear_element_stub(roadDomain.LaneSection) - lane = _linear_element_stub(roadDomain.Lane) - tags = frozenset({"from_map"}) - propagate_tags(tags, [lane_section, lane]) - assert lane_section.tags == frozenset() - assert lane.tags == frozenset() - - -def test_propagate_tags_skips_empty(): - element = _linear_element_stub(roadDomain.LaneSection) - propagate_tags(frozenset(), [element]) - assert element.tags == frozenset() - - -def test_propagate_tags_ignores_other_types(): - element = _linear_element_stub(roadDomain.RoadSection) - tags = frozenset({"from_map"}) - propagate_tags(tags, [object(), element]) - assert element.tags == tags - - -def test_propagate_tags_does_not_overwrite_existing(): - element = _linear_element_stub(roadDomain.Lane) - existing = frozenset({"existing"}) - element.tags = existing - propagate_tags(frozenset({"from_map"}), [element]) - assert element.tags == existing - - -def test_map_tags_propagate_to_lane_hierarchy(tmp_path): - road_extras = '' - network = parse_scenic_network(tmp_path, road_extras=road_extras) - road = network.roads[0] - assert road.tags == type_tags_from_road_extras(road_extras) - assert_road_level_tags_propagated(road) - assert road.lanes[0].tags == frozenset({"driving"}) - assert road.lanes[0].sections[0].tags == road.lanes[0].tags - - -def test_map_tags_from_all_type_segments_propagate(tmp_path): - road_extras = ( - '' - '' - ) - network = parse_scenic_network(tmp_path, road_extras=road_extras) - road = network.roads[0] - assert road.tags == type_tags_from_road_extras(road_extras) - assert_road_level_tags_propagated(road) - assert road.lanes[0].tags == frozenset({"driving"}) - - -def test_lane_type_tags_are_lane_specific(tmp_path): - lanes_xml = """ - - - - - """ - road_extras = '' - path = write_xodr_multi_lane(tmp_path, lanes_xml, road_extras=road_extras) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - road = road_map.toScenicNetwork().roads[0] - - tags_by_od_id = { - section.openDriveID: section.tags for section in road.sections[0].lanes - } - assert tags_by_od_id[-1] == frozenset({"driving"}) - assert tags_by_od_id[-2] == frozenset({"onRamp"}) - assert "onRamp" not in tags_by_od_id[-1] - assert "driving" not in tags_by_od_id[-2] - assert road.tags == frozenset({"motorway"}) - - -def test_junction_ramp_tags_apply_only_to_ramp_lanes(tmp_path): - lanes_xml = """ - - - - - """ - road_extras = '' - path = write_xodr_multi_lane(tmp_path, lanes_xml, road_extras=road_extras) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - internal_road = road_map.roads[7] - internal_road.extra_tags = frozenset({"onramp"}) - road, _ = internal_road.toScenicRoad(road_map.tolerance) - - assert road.tags == frozenset({"motorway", "onramp"}) - tags_by_od_id = { - section.openDriveID: section.tags for section in road.sections[0].lanes - } - assert tags_by_od_id[-1] == frozenset({"driving"}) - assert tags_by_od_id[-2] == frozenset({"onRamp"}) - assert "onramp" not in tags_by_od_id[-1] - assert "onramp" not in tags_by_od_id[-2] - assert "motorway" not in tags_by_od_id[-1] - - -def test_lane_scenic_tags_uses_open_drive_values_as_is(): - assert lane_scenic_tags("onRamp") == frozenset({"onRamp"}) - assert lane_scenic_tags("driving") == frozenset({"driving"}) - assert lane_scenic_tags(None) == frozenset() - - -def test_speed_limit_ranges_from_type_records(): - records = [ - (0.0, "town", 45.0), - (7.0, "town", 30.0), - (10.0, "town", 45.0), - (15.0, "town", 45.0), - ] - assert speed_limit_ranges_from_type_records(records, 30.0) == [ - (0.0, 7.0, 45.0), - (7.0, 10.0, 30.0), - (10.0, 30.0, 45.0), - ] - - -def test_speed_limit_ranges_carry_forward_missing_speed(): - records = [ - (0.0, "town", 45.0), - (10.0, "motorway", None), - (20.0, "motorway", 30.0), - ] - assert speed_limit_ranges_from_type_records(records, 30.0) == [ - (0.0, 20.0, 45.0), - (20.0, 30.0, 30.0), - ] - - -def test_speed_limit_for_s_interval_single_and_multiple(): - ranges = [ - (0.0, 7.0, 45.0), - (7.0, 10.0, 30.0), - (10.0, 30.0, 45.0), - ] - speed, overlapping = speed_limit_for_s_interval(ranges, 10.0, 20.0) - assert speed == pytest.approx(45.0) - assert overlapping == frozenset({45.0}) - - speed, overlapping = speed_limit_for_s_interval(ranges, 0.0, 10.0) - assert speed == pytest.approx(30.0) - assert overlapping == frozenset({45.0, 30.0}) - - -def test_speed_limit_ranges_from_lane_records(): - records = [(0.0, 20.0), (5.0, 30.0)] - assert speed_limit_ranges_from_lane_records(records, 10.0) == [ - (0.0, 5.0, 20.0), - (5.0, 10.0, 30.0), - ] - - -def test_speed_limit_ranges_from_lane_records_carry_forward(): - records = [(0.0, 20.0), (5.0, None), (8.0, 30.0)] - assert speed_limit_ranges_from_lane_records(records, 10.0) == [ - (0.0, 8.0, 20.0), - (8.0, 10.0, 30.0), - ] - - -def test_effective_speed_limit_ranges_lane_overrides_road(): - road_ranges = [(0.0, 10.0, 50.0)] - lane_ranges = [(0.0, 10.0, 80.0)] - assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ - (0.0, 10.0, 80.0), - ] - - -def test_effective_speed_limit_ranges_multiple_lane_speeds(): - road_ranges = [(0.0, 10.0, 50.0)] - lane_ranges = [(0.0, 5.0, 20.0), (5.0, 10.0, 30.0)] - assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ - (0.0, 5.0, 20.0), - (5.0, 10.0, 30.0), - ] - - -def test_effective_speed_limit_ranges_lane_speed_below_road(): - road_ranges = [(0.0, 10.0, 50.0)] - lane_ranges = [(0.0, 10.0, 30.0)] - assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ - (0.0, 10.0, 30.0), - ] - - -def test_effective_speed_limit_ranges_road_only_uses_road_ranges(): - road_ranges = [(0.0, 5.0, 50.0), (5.0, 10.0, 30.0)] - assert effective_speed_limit_ranges(road_ranges, [], 0.0, 10.0) == [ - (0.0, 5.0, 50.0), - (5.0, 10.0, 30.0), - ] - - -def test_parse_lane_speed_records(tmp_path): - lanes_xml = """ - - - - """ - path = write_xodr_lane_speeds(tmp_path, lanes_xml) - road_map = RoadMap() - road_map.parse(path) - lane = road_map.roads[7].lane_secs[0].get_lane(-1) - assert lane.speed_records == [(0.0, 20.0), (5.0, 30.0)] - assert lane.speed_limit == pytest.approx(20.0) - assert speed_limit_ranges_from_lane_records(lane.speed_records, 10.0) == [ - (0.0, 5.0, 20.0), - (5.0, 10.0, 30.0), - ] - - -def test_lane_speed_ranges_on_scenic_lane_section(tmp_path): - lanes_xml = """ - - - - """ - path = write_xodr_lane_speeds(tmp_path, lanes_xml) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - lane_section = road_map.toScenicNetwork().roads[0].sections[0].lanes[0] - - assert lane_section.speedLimit == pytest.approx(20.0) - assert lane_section.speedLimitRanges == ( - (0.0, 5.0, 20.0), - (5.0, 20.0, 30.0), - ) - assert lane_section.speedLimitAt(2.0) == pytest.approx(20.0) - assert lane_section.speedLimitAt(7.0) == pytest.approx(30.0) - - -def test_propagate_speed_limit_sets_section_specific_values(): - road_section = _linear_element_stub(roadDomain.RoadSection) - lane_section = _linear_element_stub(roadDomain.LaneSection) - propagate_speed_limit(30.0, [road_section, lane_section]) - assert road_section.speedLimit == pytest.approx(30.0) - assert lane_section.speedLimit == pytest.approx(30.0) - - other_section = _linear_element_stub(roadDomain.RoadSection) - other_lane = _linear_element_stub(roadDomain.LaneSection) - propagate_speed_limit(45.0, [other_section, other_lane]) - assert other_section.speedLimit == pytest.approx(45.0) - assert other_lane.speedLimit == pytest.approx(45.0) - - -def test_road_speed_limits_vary_by_lane_section(): - speed_ranges = speed_limit_ranges_from_type_records( - [ - (0.0, "town", 45.0), - (7.0, "town", 30.0), - (10.0, "town", 45.0), - (15.0, "town", 45.0), - (20.0, "town", 45.0), - (30.0, "town", 45.0), - ], - 30.0, - ) - section_intervals = [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)] - section_speeds = [ - speed_limit_for_s_interval(speed_ranges, s_start, s_end) - for s_start, s_end in section_intervals - ] - - assert section_speeds[0] == (pytest.approx(30.0), frozenset({45.0, 30.0})) - assert section_speeds[1] == (pytest.approx(45.0), frozenset({45.0})) - assert section_speeds[2] == (pytest.approx(45.0), frozenset({45.0})) - assert min(speed for speed, _ in section_speeds) == pytest.approx(30.0) - - -def write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=""): - path = tmp_path / "test.xodr" - path.write_text( - f""" - - - {road_extras} - {DEFAULT_PLAN_VIEW} - - - -
- -
- -{lanes_xml} - - - - - -""" - ) - return path - - -def test_lane_speed_limit_overrides_when_higher(tmp_path): - lanes_xml = """ - - - - - - - - - - """ - road_extras = '' - path = write_xodr_lane_speeds(tmp_path, lanes_xml, road_extras=road_extras) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - road = road_map.toScenicNetwork().roads[0] - - road_limit = 50 / 3.6 - fast_limit = 80 / 3.6 - assert road.speedLimit == pytest.approx(road_limit) - limits_by_od_id = { - section.openDriveID: section.speedLimit for section in road.sections[0].lanes - } - assert limits_by_od_id[-1] == pytest.approx(road_limit) - assert limits_by_od_id[-2] == pytest.approx(fast_limit) - assert limits_by_od_id[-3] == pytest.approx(40 / 3.6) - - -def _speed_elem(max_value, unit=None): - attrs = f'max="{max_value}"' - if unit is not None: - attrs += f' unit="{unit}"' - return ET.fromstring(f"") - - -def test_speed_to_mps_unit_conversions(): - assert speed_to_mps(_speed_elem(36, "km/h")) == pytest.approx(10.0) - assert speed_to_mps(_speed_elem(10, "m/s")) == pytest.approx(10.0) - assert speed_to_mps(_speed_elem(100, "mph")) == pytest.approx(44.704) - # A missing unit defaults to m/s. - assert speed_to_mps(_speed_elem(15)) == pytest.approx(15.0) - - -def test_speed_to_mps_unlimited_returns_none(): - assert speed_to_mps(_speed_elem("no limit", "km/h")) is None - assert speed_to_mps(_speed_elem("undefined")) is None - - -def test_speed_to_mps_rejects_unknown_unit(): - with pytest.raises(ValueError, match="unsupported speed unit"): - speed_to_mps(_speed_elem(50, "furlongs/fortnight")) - - -def test_merge_scenic_tags_combines_dedups_and_ignores_empty(): - assert merge_scenic_tags(frozenset({"a"}), frozenset({"a", "b"})) == frozenset( - {"a", "b"} - ) - assert merge_scenic_tags(frozenset(), None, frozenset({"x"})) == frozenset({"x"}) - assert merge_scenic_tags() == frozenset() - - -def test_junction_tags_from_type_and_name(): - assert _junction_tags(Junction(1, "Roundabout A", "roundabout")) == frozenset( - {"roundabout", "Roundabout A"} - ) - # The default junction type carries no semantic meaning and is dropped. - assert _junction_tags(Junction(2, None, "default")) == frozenset() - # type_ defaults to "default", so only the name remains. - assert _junction_tags(Junction(3, "J3")) == frozenset({"J3"}) - - -class _FakeRoad: - def __init__(self, junction): - self.junction = junction - self.extra_tags = frozenset() - - -class _FakeRoadMap: - def __init__(self, roads, junctions): - self.roads = roads - self.junctions = junctions - - -def test_assign_semantic_tags_tags_only_roads_in_junction(): - road_map = _FakeRoadMap( - roads={10: _FakeRoad(junction=5), 11: _FakeRoad(junction=None)}, - junctions={5: Junction(5, "J5", "roundabout")}, - ) - assign_semantic_tags(road_map) - assert road_map.roads[10].extra_tags == frozenset({"roundabout", "J5"}) - assert road_map.roads[11].extra_tags == frozenset() - - -def test_apply_lane_speed_limits_ignores_non_lane_elements(): - lane = _linear_element_stub(roadDomain.Lane) - lane.sections = [ - _linear_element_stub(roadDomain.LaneSection), - _linear_element_stub(roadDomain.LaneSection), - ] - lane.sections[0].speedLimit = 10.0 - lane.sections[1].speedLimit = 25.0 - section = _linear_element_stub(roadDomain.RoadSection) - apply_lane_speed_limits([lane, section, "ignored"]) - assert lane.speedLimit == pytest.approx(25.0) - assert section.speedLimit is None - - -def test_section_spanning_multiple_speeds_warns(tmp_path): - road_extras = ( - '' - '' - ) - # The single 20 m lane section [0,20) straddles both the 50 and 30 km/h limits. - path = write_xodr_lane_speeds( - tmp_path, - """ - - """, - road_extras=road_extras, - ) - road_map = RoadMap() - road_map.parse(path) - road_map.calculate_geometry(num=5, calc_intersect=True) - with pytest.warns(OpenDriveWarning, match="spans multiple speed limits"): - road_map.toScenicNetwork() diff --git a/tests/formats/opendrive/test_speeds.py b/tests/formats/opendrive/test_speeds.py new file mode 100644 index 000000000..16b9aa7cf --- /dev/null +++ b/tests/formats/opendrive/test_speeds.py @@ -0,0 +1,251 @@ +import xml.etree.ElementTree as ET + +import pytest + +from scenic.formats.opendrive.xodr_parser import ( + OpenDriveWarning, + RoadMap, + effective_speed_limit_ranges, + speed_limits_for_s_interval, + speed_limit_ranges_from_lane_records, + speed_limit_ranges_from_type_records, + speed_to_mps, +) + +from .conftest import parse_scenic_network, scenic_road, write_xodr_lane_speeds + + +def test_speed_limit_ranges_from_type_records(): + records = [ + (0.0, "town", 45.0), + (7.0, "town", 30.0), + (10.0, "town", 45.0), + (15.0, "town", 45.0), + ] + assert speed_limit_ranges_from_type_records(records, 30.0) == [ + (0.0, 7.0, 45.0), + (7.0, 10.0, 30.0), + (10.0, 30.0, 45.0), + ] + + +def test_speed_limit_ranges_carry_forward_missing_speed(): + records = [ + (0.0, "town", 45.0), + (10.0, "motorway", None), + (20.0, "motorway", 30.0), + ] + assert speed_limit_ranges_from_type_records(records, 30.0) == [ + (0.0, 10.0, 45.0), + (10.0, 20.0, None), + (20.0, 30.0, 30.0), + ] + + +def test_speed_limits_for_s_interval_single_and_multiple(): + ranges = [ + (0.0, 7.0, 45.0), + (7.0, 10.0, 30.0), + (10.0, 30.0, 45.0), + ] + overlapping = speed_limits_for_s_interval(ranges, 10.0, 20.0) + assert overlapping == frozenset({45.0}) + assert min(overlapping) == pytest.approx(45.0) + + overlapping = speed_limits_for_s_interval(ranges, 0.0, 10.0) + assert overlapping == frozenset({45.0, 30.0}) + assert min(overlapping) == pytest.approx(30.0) + + +def test_speed_limit_ranges_from_lane_records(): + records = [(0.0, 20.0), (5.0, 30.0)] + assert speed_limit_ranges_from_lane_records(records, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_speed_limit_ranges_from_lane_records_carry_forward(): + records = [(0.0, 20.0), (5.0, None), (8.0, 30.0)] + assert speed_limit_ranges_from_lane_records(records, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 8.0, None), + (8.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_lane_overrides_road(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 10.0, 80.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 10.0, 80.0), + ] + + +def test_effective_speed_limit_ranges_multiple_lane_speeds(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 5.0, 20.0), (5.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_lane_speed_below_road(): + road_ranges = [(0.0, 10.0, 50.0)] + lane_ranges = [(0.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, lane_ranges, 0.0, 10.0) == [ + (0.0, 10.0, 30.0), + ] + + +def test_effective_speed_limit_ranges_road_only_uses_road_ranges(): + road_ranges = [(0.0, 5.0, 50.0), (5.0, 10.0, 30.0)] + assert effective_speed_limit_ranges(road_ranges, [], 0.0, 10.0) == [ + (0.0, 5.0, 50.0), + (5.0, 10.0, 30.0), + ] + + +def test_parse_lane_speed_records(tmp_path): + lanes_xml = """ + + + + """ + path = write_xodr_lane_speeds(tmp_path, lanes_xml) + road_map = RoadMap() + road_map.parse(path) + lane = road_map.roads[7].lane_secs[0].get_lane(-1) + assert lane.speed_records == [(0.0, 20.0), (5.0, 30.0)] + assert speed_limit_ranges_from_lane_records(lane.speed_records, 10.0) == [ + (0.0, 5.0, 20.0), + (5.0, 10.0, 30.0), + ] + + +def test_lane_speed_ranges_on_scenic_lane_section(tmp_path): + lanes_xml = """ + + + + """ + network = parse_scenic_network(tmp_path, lanes_xml=lanes_xml) + lane_section = scenic_road(network).sections[0].lanes[0] + + assert lane_section.speedLimit == pytest.approx(20.0) + assert lane_section.speedLimitRanges == ( + (0.0, 5.0, 20.0), + (5.0, 20.0, 30.0), + ) + assert lane_section.speedLimitAt(2.0) == pytest.approx(20.0) + assert lane_section.speedLimitAt(7.0) == pytest.approx(30.0) + + # Network.speedLimitAt projects onto the section centerline and looks up + # the matching range. + slow_point = lane_section.centerline.pointAlongBy(2.0) + fast_point = lane_section.centerline.pointAlongBy(7.0) + assert network.speedLimitAt(slow_point) == pytest.approx(20.0) + assert network.speedLimitAt(fast_point) == pytest.approx(30.0) + assert network.speedLimitAt((1000.0, 1000.0)) is None + + +def test_network_speed_limit_at_uniform_section(tmp_path): + road_extras = '' + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = scenic_road(network) + point = road.lanes[0].sections[0].centerline.pointAlongBy(5.0) + assert network.speedLimitAt(point) == pytest.approx(50 / 3.6) + + +def test_road_speed_limits_vary_by_lane_section(): + speed_ranges = speed_limit_ranges_from_type_records( + [ + (0.0, "town", 45.0), + (7.0, "town", 30.0), + (10.0, "town", 45.0), + (15.0, "town", 45.0), + (20.0, "town", 45.0), + (30.0, "town", 45.0), + ], + 30.0, + ) + section_intervals = [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)] + section_speeds = [ + speed_limits_for_s_interval(speed_ranges, s_start, s_end) + for s_start, s_end in section_intervals + ] + + assert section_speeds[0] == frozenset({45.0, 30.0}) + assert section_speeds[1] == frozenset({45.0}) + assert section_speeds[2] == frozenset({45.0}) + assert min(min(speeds) for speeds in section_speeds) == pytest.approx(30.0) + + +def test_lane_speed_limit_overrides_when_higher(tmp_path): + lanes_xml = """ + + + + + + + + + + """ + road_extras = '' + network = parse_scenic_network( + tmp_path, lanes_xml=lanes_xml, road_extras=road_extras + ) + road = scenic_road(network) + + road_limit = 50 / 3.6 + fast_limit = 80 / 3.6 + assert road.speedLimit == pytest.approx(road_limit) + limits_by_od_id = { + section.openDriveID: section.speedLimit for section in road.sections[0].lanes + } + assert limits_by_od_id[-1] == pytest.approx(road_limit) + assert limits_by_od_id[-2] == pytest.approx(fast_limit) + assert limits_by_od_id[-3] == pytest.approx(40 / 3.6) + + +def _speed_elem(max_value, unit=None): + attrs = f'max="{max_value}"' + if unit is not None: + attrs += f' unit="{unit}"' + return ET.fromstring(f"") + + +def test_speed_to_mps_unit_conversions(): + assert speed_to_mps(_speed_elem(36, "km/h")) == pytest.approx(10.0) + assert speed_to_mps(_speed_elem(10, "m/s")) == pytest.approx(10.0) + assert speed_to_mps(_speed_elem(100, "mph")) == pytest.approx(44.704) + # A missing unit defaults to m/s. + assert speed_to_mps(_speed_elem(15)) == pytest.approx(15.0) + + +def test_speed_to_mps_unlimited_returns_none(): + assert speed_to_mps(_speed_elem("no limit", "km/h")) is None + assert speed_to_mps(_speed_elem("undefined")) is None + + +def test_speed_to_mps_rejects_unknown_unit(): + with pytest.raises(ValueError, match="unsupported speed unit"): + speed_to_mps(_speed_elem(50, "furlongs/fortnight")) + + +def test_section_spanning_multiple_speeds_warns(tmp_path): + road_extras = ( + '' + '' + ) + # The single 20 m lane section [0,20) straddles both the 50 and 30 km/h limits. + lanes_xml = """ + + """ + with pytest.warns(OpenDriveWarning, match="spans multiple speed limits"): + parse_scenic_network( + tmp_path, lanes_xml=lanes_xml, road_extras=road_extras + ) diff --git a/tests/formats/opendrive/test_tags.py b/tests/formats/opendrive/test_tags.py new file mode 100644 index 000000000..80be2539e --- /dev/null +++ b/tests/formats/opendrive/test_tags.py @@ -0,0 +1,116 @@ +import pytest + +from scenic.formats.opendrive.xodr_parser import Junction, assign_semantic_tags + +from .conftest import ( + assert_road_level_tags_propagated, + parse_scenic_network, + scenic_road, + type_tags_from_road_extras, +) + + +def test_map_tags_propagate_to_lane_hierarchy(tmp_path): + road_extras = '' + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = network.roads[0] + assert road.tags == type_tags_from_road_extras(road_extras) + assert_road_level_tags_propagated(road) + assert road.lanes[0].tags == frozenset({"driving"}) + assert road.lanes[0].sections[0].tags == road.lanes[0].tags + + +def test_map_tags_from_all_type_segments_propagate(tmp_path): + road_extras = ( + '' + '' + ) + network = parse_scenic_network(tmp_path, road_extras=road_extras) + road = network.roads[0] + assert road.tags == type_tags_from_road_extras(road_extras) + assert_road_level_tags_propagated(road) + assert road.lanes[0].tags == frozenset({"driving"}) + + +def test_lane_type_tags_are_lane_specific(tmp_path): + lanes_xml = """ + + + + + """ + road_extras = '' + network = parse_scenic_network( + tmp_path, lanes_xml=lanes_xml, road_extras=road_extras + ) + road = scenic_road(network) + + tags_by_od_id = { + section.openDriveID: section.tags for section in road.sections[0].lanes + } + assert tags_by_od_id[-1] == frozenset({"driving"}) + assert tags_by_od_id[-2] == frozenset({"onRamp"}) + assert "onRamp" not in tags_by_od_id[-1] + assert "driving" not in tags_by_od_id[-2] + assert road.tags == frozenset({"motorway"}) + + +def test_junction_ramp_tags_apply_only_to_ramp_lanes(tmp_path): + lanes_xml = """ + + + + + """ + road_extras = '' + network = parse_scenic_network( + tmp_path, + lanes_xml=lanes_xml, + road_extras=road_extras, + junction_id="5", + junction_type="onramp", + junction_name="J5", + ) + road = scenic_road(network) + + assert road.tags == frozenset({"motorway", "onramp"}) + tags_by_od_id = { + section.openDriveID: section.tags for section in road.sections[0].lanes + } + assert tags_by_od_id[-1] == frozenset({"driving"}) + assert tags_by_od_id[-2] == frozenset({"onRamp"}) + assert "onramp" not in tags_by_od_id[-1] + assert "onramp" not in tags_by_od_id[-2] + assert "motorway" not in tags_by_od_id[-1] + + +def test_junction_tags_from_type(): + assert Junction(1, "Roundabout A", "roundabout").tags == frozenset( + {"roundabout"} + ) + # The default junction type carries no semantic meaning and is dropped. + assert Junction(2, None, "default").tags == frozenset() + # type_ defaults to "default"; junction names are not semantic tags. + assert Junction(3, "J3").tags == frozenset() + + +class _FakeRoad: + def __init__(self, junction): + self.junction = junction + self.extra_tags = frozenset() + + +class _FakeRoadMap: + def __init__(self, roads, junctions): + self.roads = roads + self.junctions = junctions + + +def test_assign_semantic_tags_tags_only_roads_in_junction(): + road_map = _FakeRoadMap( + roads={10: _FakeRoad(junction=5), 11: _FakeRoad(junction=None)}, + junctions={5: Junction(5, "J5", "roundabout")}, + ) + assign_semantic_tags(road_map) + assert road_map.roads[10].extra_tags == frozenset({"roundabout"}) + assert road_map.roads[11].extra_tags == frozenset() From e345b5e9d830e8208ebf946658ecbb7caed0e132 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 22:28:46 -0700 Subject: [PATCH 23/26] unified lane-section speed assignment --- src/scenic/formats/opendrive/xodr_parser.py | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 9c2d83c29..24b48c87c 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -65,7 +65,7 @@ def _speed_limit_ranges_from_s_speed_records(sorted_records, domain_length): sorted_records = [(0.0, None)] + sorted_records ranges = [] - for i, (s, speed_mps) in enumerate (sorted_records): + for i, (s, speed_mps) in enumerate(sorted_records): s_end = sorted_records[i + 1][0] if i + 1 < len(sorted_records) else domain_length if s_end <= s: continue @@ -156,7 +156,7 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): speeds = {speed for _, _, speed in ranges if speed is not None} - element.speedLimit = min(speeds) + element.speedLimit = min(speeds) if speeds else None element.speedLimitRanges = tuple(ranges) if len(speeds) > 1 and warn_context is not None: speeds_text = ", ".join(f"{speed:.4g} m/s" for speed in sorted(speeds)) @@ -166,13 +166,6 @@ def assign_speed_limit_from_ranges(element, ranges, warn_context=None): ) -def lane_scenic_tags(lane_type): - """Tags for a Lane or LaneSection from the OpenDRIVE lane ``type`` attribute.""" - if not lane_type: - return frozenset() - return frozenset({lane_type}) - - def assign_semantic_tags(road_map): """Populate each ``Road.extra_tags`` from the type of its junction. @@ -1040,13 +1033,7 @@ def toScenicRoad(self, tolerance): effective_ranges = effective_speed_limit_ranges( speed_ranges, lane_ranges, s_start, section_length ) - if not effective_ranges: - if ( - section_speed_limit is not None - and lane_section.speedLimit is None - ): - lane_section.speedLimit = section_speed_limit - elif any(speed is not None for _, _, speed in effective_ranges): + if effective_ranges: assign_speed_limit_from_ranges( lane_section, effective_ranges, From 8b22787053289dab5101dd34ced763a5ad72dd65 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 22:30:19 -0700 Subject: [PATCH 24/26] drop open_drive_type checker --- src/scenic/formats/opendrive/xodr_parser.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 24b48c87c..5872c844f 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -931,7 +931,7 @@ def toScenicRoad(self, tolerance): type_tags = frozenset( open_drive_type for _, open_drive_type, _ in self.type_records - if open_drive_type + # if open_drive_type ) road_level_tags = frozenset(type_tags | self.extra_tags) allElements = [] From 071a234381e546eb013a7701e8611ab7e3cd57c4 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 22:41:17 -0700 Subject: [PATCH 25/26] added test for no-limit speeds --- tests/formats/opendrive/test_speeds.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/formats/opendrive/test_speeds.py b/tests/formats/opendrive/test_speeds.py index 16b9aa7cf..19c186ddb 100644 --- a/tests/formats/opendrive/test_speeds.py +++ b/tests/formats/opendrive/test_speeds.py @@ -158,6 +158,18 @@ def test_network_speed_limit_at_uniform_section(tmp_path): assert network.speedLimitAt(point) == pytest.approx(50 / 3.6) +def test_no_limit_speed_sets_ranges_with_none(tmp_path): + road_extras = ( + '' + ) + network = parse_scenic_network(tmp_path, road_extras=road_extras) + lane_section = scenic_road(network).lanes[0].sections[0] + assert lane_section.speedLimit is None + assert lane_section.speedLimitRanges == ((0.0, 20.0, None),) + point = lane_section.centerline.pointAlongBy(5.0) + assert network.speedLimitAt(point) is None + + def test_road_speed_limits_vary_by_lane_section(): speed_ranges = speed_limit_ranges_from_type_records( [ From 59d2264dba6b324988fbf328a77bd88ab88413d5 Mon Sep 17 00:00:00 2001 From: aprabou Date: Sun, 12 Jul 2026 23:19:08 -0700 Subject: [PATCH 26/26] proper reformat --- src/scenic/domains/driving/roads.py | 1 - src/scenic/formats/opendrive/xodr_parser.py | 12 +++--------- tests/formats/opendrive/test_speeds.py | 14 ++++---------- tests/formats/opendrive/test_tags.py | 8 ++------ 4 files changed, 9 insertions(+), 26 deletions(-) diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index 9bcb109e4..c0928286b 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -255,7 +255,6 @@ def speedLimitAt(self, s: float) -> Union[float, None]: # s is at/beyond the final (half-open) range's end: clamp to it. return speed return self.speedLimit - def __attrs_post_init__(self): assert self.uid is not None or self.id is not None diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 5872c844f..30aee57ae 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -172,9 +172,7 @@ def assign_semantic_tags(road_map): A road belonging to a junction (``road.junction`` equals the junction id) inherits that junction's semantic tags; all other roads get no extra tags. """ - junction_tags = { - jid: junction.tags for jid, junction in road_map.junctions.items() - } + junction_tags = {jid: junction.tags for jid, junction in road_map.junctions.items()} for road in road_map.roads.values(): if road.junction is not None: # road.junction is the raw OpenDRIVE id string; junctions are keyed by int. @@ -954,12 +952,8 @@ def toScenicRoad(self, tolerance): if sec_index + 1 < len(self.lane_secs) else self.length ) - overlapping_speeds = speed_limits_for_s_interval( - speed_ranges, s_start, s_end - ) - section_speed_limit = ( - min(overlapping_speeds) if overlapping_speeds else None - ) + overlapping_speeds = speed_limits_for_s_interval(speed_ranges, s_start, s_end) + section_speed_limit = min(overlapping_speeds) if overlapping_speeds else None if len(overlapping_speeds) > 1: speeds_text = ", ".join( f"{speed:.4g} m/s" for speed in sorted(overlapping_speeds) diff --git a/tests/formats/opendrive/test_speeds.py b/tests/formats/opendrive/test_speeds.py index 19c186ddb..2f1f50ec1 100644 --- a/tests/formats/opendrive/test_speeds.py +++ b/tests/formats/opendrive/test_speeds.py @@ -6,9 +6,9 @@ OpenDriveWarning, RoadMap, effective_speed_limit_ranges, - speed_limits_for_s_interval, speed_limit_ranges_from_lane_records, speed_limit_ranges_from_type_records, + speed_limits_for_s_interval, speed_to_mps, ) @@ -159,9 +159,7 @@ def test_network_speed_limit_at_uniform_section(tmp_path): def test_no_limit_speed_sets_ranges_with_none(tmp_path): - road_extras = ( - '' - ) + road_extras = '' network = parse_scenic_network(tmp_path, road_extras=road_extras) lane_section = scenic_road(network).lanes[0].sections[0] assert lane_section.speedLimit is None @@ -207,9 +205,7 @@ def test_lane_speed_limit_overrides_when_higher(tmp_path): """ road_extras = '' - network = parse_scenic_network( - tmp_path, lanes_xml=lanes_xml, road_extras=road_extras - ) + network = parse_scenic_network(tmp_path, lanes_xml=lanes_xml, road_extras=road_extras) road = scenic_road(network) road_limit = 50 / 3.6 @@ -258,6 +254,4 @@ def test_section_spanning_multiple_speeds_warns(tmp_path): """ with pytest.warns(OpenDriveWarning, match="spans multiple speed limits"): - parse_scenic_network( - tmp_path, lanes_xml=lanes_xml, road_extras=road_extras - ) + parse_scenic_network(tmp_path, lanes_xml=lanes_xml, road_extras=road_extras) diff --git a/tests/formats/opendrive/test_tags.py b/tests/formats/opendrive/test_tags.py index 80be2539e..dcd1d21d3 100644 --- a/tests/formats/opendrive/test_tags.py +++ b/tests/formats/opendrive/test_tags.py @@ -40,9 +40,7 @@ def test_lane_type_tags_are_lane_specific(tmp_path): """ road_extras = '' - network = parse_scenic_network( - tmp_path, lanes_xml=lanes_xml, road_extras=road_extras - ) + network = parse_scenic_network(tmp_path, lanes_xml=lanes_xml, road_extras=road_extras) road = scenic_road(network) tags_by_od_id = { @@ -85,9 +83,7 @@ def test_junction_ramp_tags_apply_only_to_ramp_lanes(tmp_path): def test_junction_tags_from_type(): - assert Junction(1, "Roundabout A", "roundabout").tags == frozenset( - {"roundabout"} - ) + assert Junction(1, "Roundabout A", "roundabout").tags == frozenset({"roundabout"}) # The default junction type carries no semantic meaning and is dropped. assert Junction(2, None, "default").tags == frozenset() # type_ defaults to "default"; junction names are not semantic tags.