diff --git a/src/scenic/domains/driving/roads.py b/src/scenic/domains/driving/roads.py index f06377c5a..a2f7ef006 100644 --- a/src/scenic/domains/driving/roads.py +++ b/src/scenic/domains/driving/roads.py @@ -27,7 +27,7 @@ import attr import shapely -from shapely.geometry import MultiPolygon, Polygon +from shapely.geometry import MultiPolygon, Point as ShapelyPoint, Polygon from scenic.core.distributions import ( RejectionException, @@ -147,6 +147,34 @@ def guessTypeFromLanes( return ManeuverType.STRAIGHT +@enum.unique +class SignalPriorityType(enum.Enum): + """OpenDRIVE ``e_signals_semantics_priority`` literals.""" + + FOUR_WAY = "4way" + KEEP_CLEAR_LINE = "keepClearLine" + NO_PARKING_LINE = "noParkingLine" + NO_TURN_ON_RED = "noTurnOnRed" + PRIORITY_ROAD_END = "priorityRoadEnd" + PRIORITY_ROAD = "priorityRoad" + PRIORITY_TO_THE_RIGHT_RULE = "priorityToTheRightRule" + STOP_LINE = "stopLine" + STOP = "stop" + TRAFFIC_LIGHT = "trafficLight" + TURN_ON_RED_ALLOWED = "turnOnRedAllowed" + WAITING_LINE = "waitingLine" + YIELD = "yield" + UNKNOWN = "unknown" + + @classmethod + def fromOpenDrive(cls, type_str: str) -> SignalPriorityType: + """Map an OpenDRIVE priority string to an enum member.""" + for member in cls: + if member is not cls.UNKNOWN and member.value == type_str: + return member + return cls.UNKNOWN + + @attr.s(auto_attribs=True, kw_only=True, eq=False) class Maneuver(_ElementReferencer): """Maneuver() @@ -571,6 +599,22 @@ def sectionAt(self, point: Vectorlike, reject=False) -> Union[LaneSection, None] """Get the LaneSection passing through a given point.""" return self.network.findPointIn(point, self.sections, reject) + def haltPointAhead(self, position, lookahead: float = 80.0) -> Optional[Vector]: + """Nearest `Signal.stoppingPointOn` this lane still ahead of ``position``.""" + if self.road is None: + return None + here = self.centerline.lineString.project(ShapelyPoint(position.x, position.y)) + best = None + best_ahead = None + for sig in self.road.signals: + pt = sig.stoppingPointOn(self) + if pt is None: + continue + ahead = self.centerline.lineString.project(ShapelyPoint(pt.x, pt.y)) - here + if 0 <= ahead <= lookahead and (best_ahead is None or ahead < best_ahead): + best, best_ahead = pt, ahead + return best + @attr.s(auto_attribs=True, kw_only=True, repr=False, eq=False) class RoadSection(LinearElement): @@ -809,6 +853,20 @@ def nominalDirectionsAt(self, point: Vectorlike) -> Tuple[Orientation]: return tuple(m.connectingLane.orientation[point] for m in maneuvers) +@attr.s(auto_attribs=True, frozen=True, kw_only=True) +class SignalLink: + """OpenDRIVE ```` from this signal to another signal or object. + + Distinct from ````, which re-applies the same signal on + another road. A typical 1.8+ use is a traffic light linking to a stop line + (``elementType="signal"``, ``type="stopline"``). + """ + + elementId: str + elementType: str + type: Optional[str] = None + + @attr.s(auto_attribs=True, kw_only=True, repr=False, eq=False) class Signal: """Traffic lights, stop signs, etc. @@ -825,12 +883,204 @@ class Signal: country: str #: Type identifier according to country code. type: str + #: Subtype identifier according to country code (``None`` if absent). + subtype: Optional[str] = None + #: OpenDRIVE ``e_signals_semantics_priority`` entries (empty if unknown / pre-1.8). + #: All ```` children are kept; they are not collapsed to a single type. + priorities: Tuple[SignalPriorityType, ...] = () + #: Longitudinal s-coordinate along the road reference line (OpenDRIVE ``s``). + #: Physical pole location in 1.8+; logical effect station if `sIsLogical`. + s: Optional[float] = None + #: Lateral t-coordinate from the reference line (OpenDRIVE ``t``; +t = left). + t: Optional[float] = None + #: OpenDRIVE signal orientation: ``"+"``, ``"-"``, or ``"none"``. + orientation: Optional[str] = None + #: OpenDRIVE ```` lane range ``(fromLane, toLane)``, if any. + validity: Optional[Tuple[int, int]] = None + #: OpenDRIVE ```` links (e.g. light → stop line). + references: Tuple[SignalLink, ...] = () + #: True when deprecated ```` / ```` is present, + #: so `s` is the logical effect station rather than the pole. + sIsLogical: bool = False + #: Station along the parent road where an ego should halt for this signal. + #: May differ from `s` (e.g. a traffic light's pole vs its stop line). + #: ``None`` if we cannot derive one (typical for a connector-only light). + stoppingS: Optional[float] = None + #: Placeholder for a future world-space device/pole position. Halt decisions + #: intentionally use `stoppingS` and `stoppingPointOn`, not this field. + position: Optional[Vector] = None + #: Maneuvers that require this signal to be green (empty if unknown). + controlledManeuvers: Tuple[Maneuver, ...] = () + + def hasPriority(self, priority: SignalPriorityType) -> bool: + """Whether this signal lists the given OpenDRIVE priority semantic.""" + return priority in self.priorities + + def affects(self, lane: Lane) -> bool: + """Whether this signal applies to ``lane``. + + Uses OpenDRIVE ``validity`` when present, then ``orientation`` against + whether the lane travels with the road (+s). A validity range that + matches no driving lane (CARLA's ``0–0``) is ignored; those files + encode lamp facing, so a junction-contact light then applies only to + the arriving side, not the road you turn into. + """ + validity_is_dummy = False + if self.validity is not None: + lo, hi = min(self.validity), max(self.validity) + + def validity_hits(candidate): + return any(lo <= sec.openDriveID <= hi for sec in candidate.sections) + + validity_hits_lane = validity_hits(lane) + if not validity_hits_lane: + road = lane.road + if road is not None: + validity_is_dummy = not any( + validity_hits(other) for other in road.lanes + ) + else: + validity_is_dummy = lo == 0 and hi == 0 + if not validity_is_dummy and not validity_hits_lane: + return False + if validity_is_dummy: + # CARLA 0–0 is not a lane range; orientation is lamp facing. + # Junction-contact lights still only apply to the arriving side. + return self._lane_arrives_at_halt(lane) + if self.orientation in (None, "none"): + return True + is_forward = any(sec.isForward for sec in lane.sections) + if self.orientation == "+": + return is_forward + if self.orientation == "-": + return not is_forward + return True + + def _isHaltLocation(self) -> bool: + return ( + self.isStop + or self.isYield + or self.isStopLine + or self.isWaitingLine + or self.isKeepClearLine + ) + + def resolveStoppingS(self, by_id, plus_contact, minus_contact): + """Pick the halt station from references, logical ``s``, or junction contact. + + ``by_id`` maps OpenDRIVE signal id → `Signal` on the same road. + ``plus_contact`` / ``minus_contact`` are that road's junction stations + for +s / −s travel, or ``None`` if that end is not a junction. + """ + for link in self.references: + if link.elementType != "signal": + continue + target = by_id.get(link.elementId) + if target is None: + continue + kind = (link.type or "").replace("_", "").lower() + if kind == "stopline" or target._isHaltLocation(): + return target.s + if self.sIsLogical or self._isHaltLocation(): + return self.s + if self.isTrafficLight: + candidates = [ + contact + for contact in (plus_contact, minus_contact) + if contact is not None + ] + if not candidates: + return None + # CARLA poles sit next to the junction they serve; orientation + # often names the other end (or a non-junction end). + if self.s is not None: + return min(candidates, key=lambda contact: abs(contact - self.s)) + if self.orientation == "+": + return plus_contact + if self.orientation == "-": + return minus_contact + return candidates[0] + return None + + def _lane_arrives_at_halt(self, lane: Lane) -> bool: + """Whether ``lane`` is still traveling into a junction-contact halt.""" + road = lane.road + if road is None or self.stoppingS is None: + return True + if not self.isTrafficLight or self.sIsLogical or self._isHaltLocation(): + return True + for link in self.references: + if link.elementType == "signal": + kind = (link.type or "").replace("_", "").lower() + if kind == "stopline": + return True + length = road.centerline.length + s = self.stoppingS + if abs(s) > 1e-4 and abs(s - length) > 1e-4: + return True + is_forward = any(sec.isForward for sec in lane.sections) + at_start = abs(self.stoppingS) <= 1e-4 + return (not is_forward) if at_start else is_forward + + def stoppingPointOn(self, lane: Lane) -> Optional[Vector]: + """Point on ``lane`` where an ego should halt for this signal, if any. + + Takes `stoppingS` along the parent road's +s centerline, then the + nearest point on ``lane``. A junction-contact halt is only returned + for the direction still arriving at that end — not the lane you enter + after turning through the light. ``None`` if the signal does not + `affects` the lane, has no `stoppingS`, or does not belong to + ``lane.road``. + """ + if self.stoppingS is None or not self.affects(lane): + return None + road = lane.road + if road is None or self not in road.signals: + return None + if not self._lane_arrives_at_halt(lane): + return None + length = road.centerline.length + if length == 0: + return None + s = min(max(self.stoppingS, 0.0), length) + return lane.centerline.project(road.centerline.pointAlongBy(s)) @property def isTrafficLight(self) -> bool: - """Whether or not this signal is a traffic light.""" + """Whether this signal is a traffic light.""" + if self.priorities: + return self.hasPriority(SignalPriorityType.TRAFFIC_LIGHT) return self.type == "1000001" + @property + def isStop(self) -> bool: + """Whether this signal is a stop sign.""" + if self.priorities: + return self.hasPriority(SignalPriorityType.STOP) + return self.type == "206" + + @property + def isYield(self) -> bool: + """Whether this signal is a yield sign.""" + if self.priorities: + return self.hasPriority(SignalPriorityType.YIELD) + return self.type == "205" + + @property + def isStopLine(self) -> bool: + """Whether this signal marks a stop line.""" + return self.hasPriority(SignalPriorityType.STOP_LINE) + + @property + def isWaitingLine(self) -> bool: + """Whether this signal marks a waiting line.""" + return self.hasPriority(SignalPriorityType.WAITING_LINE) + + @property + def isKeepClearLine(self) -> bool: + """Whether this signal marks a keep-clear line.""" + return self.hasPriority(SignalPriorityType.KEEP_CLEAR_LINE) + @attr.s(auto_attribs=True, kw_only=True, repr=False, eq=False) class Network: diff --git a/src/scenic/formats/opendrive/xodr_parser.py b/src/scenic/formats/opendrive/xodr_parser.py index 8175d2c83..8112c1a0f 100644 --- a/src/scenic/formats/opendrive/xodr_parser.py +++ b/src/scenic/formats/opendrive/xodr_parser.py @@ -1159,14 +1159,32 @@ def getEdges(forward): # Create signal roadSignals = [] for i, signal_ in enumerate(self.signals): + validity = None if signal_.validity is None else tuple(signal_.validity) signal = roadDomain.Signal( uid=f"signal{signal_.id_}_{self.id_}_{i}", openDriveID=signal_.id_, country=signal_.country, type=signal_.type_, + subtype=signal_.subtype, + priorities=signal_.priorities, + s=signal_.s, + t=signal_.t, + orientation=signal_.orientation, + validity=validity, + references=signal_.references, + sIsLogical=signal_.sIsLogical, + # Placeholder: physical pole coordinates are not needed for halt + # decisions, which use stoppingS and stoppingPointOn instead. + position=None, ) roadSignals.append(signal) + by_id = {str(sig.openDriveID): sig for sig in roadSignals} + plus_contact = self.length if self.successor is not None else None + minus_contact = 0.0 if self.predecessor is not None else None + for sig in roadSignals: + sig.stoppingS = sig.resolveStoppingS(by_id, plus_contact, minus_contact) + # Create road assert forwardGroup or backwardGroup if forwardGroup: @@ -1233,25 +1251,49 @@ def getEdges(forward): class Signal: """Traffic lights, stop signs, etc.""" - def __init__(self, id_, country, type_, subtype, orientation, validity=None): + def __init__( + self, + id_, + country, + type_, + subtype, + orientation, + s, + t, + validity=None, + priorities=(), + references=(), + sIsLogical=False, + ): self.id_ = id_ self.country = country self.type_ = type_ self.subtype = subtype self.orientation = orientation + self.s = s + self.t = t self.validity = validity + #: Tuple of `roadDomain.SignalPriorityType` from ````. + self.priorities = tuple(priorities) + #: Tuple of `roadDomain.SignalLink` from ````. + self.references = tuple(references) + self.sIsLogical = sIsLogical def is_valid(self): + """Whether ``validity`` names a real lane (not CARLA's dummy ``0–0``).""" return self.validity is None or self.validity != [0, 0] class SignalReference: - def __init__(self, id_, orientation, validity=None): + def __init__(self, id_, orientation, s, t, validity=None): self.id_ = id_ - self.validity = validity self.orientation = orientation + self.s = s + self.t = t + self.validity = validity def is_valid(self): + """Whether ``validity`` names a real lane (not CARLA's dummy ``0–0``).""" return self.validity is None or self.validity != [0, 0] @@ -1459,6 +1501,73 @@ def __parse_signal_validity(self, validity_elem): return None return [int(validity_elem.get("fromLane")), int(validity_elem.get("toLane"))] + # OpenDRIVE / CARLA country="OpenDRIVE" type codes with a known priority meaning. + _LEGACY_TYPE_TO_PRIORITY = { + "1000001": roadDomain.SignalPriorityType.TRAFFIC_LIGHT, + "206": roadDomain.SignalPriorityType.STOP, + "205": roadDomain.SignalPriorityType.YIELD, + } + + def __warn_priority_type_disagreement(self, signal): + """Warn if a known legacy ``type`` conflicts with ```` semantics.""" + legacy = self._LEGACY_TYPE_TO_PRIORITY.get(signal.type_) + if legacy is not None and signal.priorities and legacy not in signal.priorities: + listed = ", ".join(p.value for p in signal.priorities) + warn( + f'signal {signal.id_} has OpenDRIVE type "{signal.type_}" ' + f"(legacy {legacy.value}) but lists [{listed}]; " + f"using priorities for classification" + ) + + def __parse_signal_priorities(self, signal_elem): + """Parse ```` children (OpenDRIVE 1.8+). + + Returns a tuple of `roadDomain.SignalPriorityType`. Unknown literals are + mapped to `UNKNOWN` and emit an `OpenDriveWarning`. Other semantic + categories (````, ````, …) are ignored for now. + """ + semantics_elem = signal_elem.find("semantics") + if semantics_elem is None: + return () + priorities = [] + for priority_elem in semantics_elem.findall("priority"): + type_str = priority_elem.get("type") + if type_str is None: + warn( + f'signal {signal_elem.get("id")} has without type; ' + "skipping it" + ) + continue + priority = roadDomain.SignalPriorityType.fromOpenDrive(type_str) + if priority is roadDomain.SignalPriorityType.UNKNOWN: + warn( + f'signal {signal_elem.get("id")} has unrecognized ' + f'priority type "{type_str}"; storing as UNKNOWN' + ) + priorities.append(priority) + return tuple(priorities) + + def __parse_signal_links(self, signal_elem): + """Parse ```` children (light → stop line, etc.).""" + links = [] + for ref in signal_elem.findall("reference"): + element_id = ref.get("elementId") + element_type = ref.get("elementType") + if not element_id or not element_type: + warn( + f'signal {signal_elem.get("id")} has ' + "without elementId or elementType; skipping it" + ) + continue + links.append( + roadDomain.SignalLink( + elementId=element_id, + elementType=element_type, + type=ref.get("type"), + ) + ) + return tuple(links) + def __parse_signal(self, signal_elem): return Signal( signal_elem.get("id"), @@ -1466,13 +1575,24 @@ def __parse_signal(self, signal_elem): signal_elem.get("type"), signal_elem.get("subtype"), signal_elem.get("orientation"), + float(signal_elem.get("s")), + float(signal_elem.get("t")), + # other required fields not parsed: + # dynamic signal_elem.get("dynamic"), + # zOffset signal_elem.get("zOffset"), self.__parse_signal_validity(signal_elem.find("validity")), + self.__parse_signal_priorities(signal_elem), + self.__parse_signal_links(signal_elem), + signal_elem.find("positionRoad") is not None + or signal_elem.find("positionInertial") is not None, ) def __parse_signal_reference(self, signal_reference_elem): return SignalReference( signal_reference_elem.get("id"), signal_reference_elem.get("orientation"), + float(signal_reference_elem.get("s")), + float(signal_reference_elem.get("t")), self.__parse_signal_validity(signal_reference_elem.find("validity")), ) @@ -1676,22 +1796,29 @@ def popLastSectionIfShort(l): if signals is not None: for signal_elem in signals.iter("signal"): signal = self.__parse_signal(signal_elem) - if signal.is_valid(): - road.signals.append(signal) + # Do not drop CARLA dummy validity [0, 0]. Those are real + # lights; maneuver matching still ignores a 0-0 range. + self.__warn_priority_type_disagreement(signal) + road.signals.append(signal) for signal_ref_elem in signals.iter("signalReference"): signalReference = self.__parse_signal_reference(signal_ref_elem) - if signalReference.is_valid(): - referencedSignal = _temp_signals[signalReference.id_] - signal = Signal( - referencedSignal.id_, - referencedSignal.country, - referencedSignal.type_, - referencedSignal.subtype, - signalReference.orientation, - signalReference.validity, - ) - road.signals.append(signal) + referencedSignal = _temp_signals[signalReference.id_] + # Semantics come from the canonical ; placement + # (s/t/orientation/validity) is this road's . + signal = Signal( + referencedSignal.id_, + referencedSignal.country, + referencedSignal.type_, + referencedSignal.subtype, + signalReference.orientation, + signalReference.s, + signalReference.t, + signalReference.validity, + referencedSignal.priorities, + referencedSignal.references, + ) + road.signals.append(signal) if len(road.lane_secs) > 1: popLastSectionIfShort(road.length - s) diff --git a/tests/formats/opendrive/test_signal_halt_encodings.py b/tests/formats/opendrive/test_signal_halt_encodings.py new file mode 100644 index 000000000..c5b783f42 --- /dev/null +++ b/tests/formats/opendrive/test_signal_halt_encodings.py @@ -0,0 +1,412 @@ +"""Halt-point encodings: deprecated logical-s, signalReference, and CARLA dummy validity. + +Junction-contact traffic lights must stop only the arriving direction. The lane +you turn into at the far side of a green light must not inherit a halt at s=0. +""" + +from pathlib import Path + +import pytest + +from scenic.core.vectors import Vector +from scenic.domains.driving.roads import Network + +# Shared two-way section: +1 travels −s, −1 travels +s. +_TWOWAY_LANES = """\ + + + + + + + +
+ + + + + + +""" + +# Pre-1.8: @s is the effect station; / is the pole. +MAP_DEPRECATED_LOGICAL = f""" + +
+ + + + + +{_TWOWAY_LANES} + + + + + + + + + + + + + + +""" + +# Canonical light on the west approach; same id re-applied on the east road +# (the road you turn into) via . +MAP_SIGNAL_REFERENCE = f""" + +
+ + + + + + +{_TWOWAY_LANES} + + + + + + + + + + + +{_TWOWAY_LANES} + + + + + + + + + + + + + + + + +
+ + + + + + + + + + + + + + + +""" + +# Same geometry as MAP_SIGNAL_REFERENCE, but CARLA-undefined: dummy 0–0 validity +# and lamp-facing orientation (often the opposite of the arriving lane). +MAP_CARLA_TWOWAY = f""" + +
+ + + + + + +{_TWOWAY_LANES} + + + + + + + + + + + + + +{_TWOWAY_LANES} + + + + + + + + + + + + + + + + + + +
+ + + + + + + + + + + + + + + +""" + +# CARLA-style dummy validity on a (Town10HD hybrid). +MAP_CARLA_SIGNAL_REFERENCE = MAP_SIGNAL_REFERENCE.replace( + '', + '\n' + ' \n' + " ", +) + + +def load_network(tmp_path: Path, xml: str) -> Network: + path = tmp_path / "map.xodr" + path.write_text(xml) + return Network.fromFile(path, useCache=False) + + +def od_id(lane) -> int: + ids = {sec.openDriveID for sec in lane.sections} + assert len(ids) == 1, lane + return next(iter(ids)) + + +def lanes_by_id(road): + return {od_id(lane): lane for lane in road.lanes} + + +def road_by_id(network, road_id): + return next(road for road in network.roads if road.id == road_id) + + +def signal_on(road, open_drive_id): + matches = [sig for sig in road.signals if str(sig.openDriveID) == str(open_drive_id)] + assert len(matches) == 1, (road.uid, open_drive_id, matches) + return matches[0] + + +def _xy(pt, x, y, tol=0.15): + assert pt is not None + assert abs(pt.x - x) < tol, (pt.x, x) + assert abs(pt.y - y) < tol, (pt.y, y) + + +# --- deprecated pre-1.8 logical @s --- + + +def test_deprecated_position_road_halts_at_logical_s_not_pole(tmp_path): + network = load_network(tmp_path, MAP_DEPRECATED_LOGICAL) + road = road_by_id(network, 1) + lanes = lanes_by_id(road) + stop = signal_on(road, 10) + assert stop.sIsLogical + assert stop.s == 12.0 + assert stop.stoppingS == 12.0 + assert stop.stoppingPointOn(lanes[-1]) is None + _xy(stop.stoppingPointOn(lanes[1]), 12.0, 1.75) + + +def test_deprecated_position_inertial_halts_at_logical_s(tmp_path): + network = load_network(tmp_path, MAP_DEPRECATED_LOGICAL) + road = road_by_id(network, 1) + lanes = lanes_by_id(road) + yield_sig = signal_on(road, 11) + assert yield_sig.sIsLogical + assert yield_sig.stoppingS == 28.0 + _xy(yield_sig.stoppingPointOn(lanes[-1]), 28.0, -1.75) + assert yield_sig.stoppingPointOn(lanes[1]) is None + + +def test_deprecated_midroad_stopline_still_both_directions(tmp_path): + """A painted line at mid-s is not a junction entry; both directions halt.""" + network = load_network(tmp_path, MAP_DEPRECATED_LOGICAL) + road = road_by_id(network, 1) + lanes = lanes_by_id(road) + line = signal_on(road, 12) + assert not line.sIsLogical + assert line.stoppingS == 20.0 + _xy(line.stoppingPointOn(lanes[-1]), 20.0, -1.75) + _xy(line.stoppingPointOn(lanes[1]), 20.0, 1.75) + + +def test_deprecated_halt_point_ahead_skips_opposite_direction(tmp_path): + network = load_network(tmp_path, MAP_DEPRECATED_LOGICAL) + road = road_by_id(network, 1) + lanes = lanes_by_id(road) + # +s traffic: stop line at 20, then yield at 28. Opposite stop at 12 is behind. + first = lanes[-1].haltPointAhead(Vector(1, -1.75)) + _xy(first, 20.0, -1.75) + second = lanes[-1].haltPointAhead(Vector(21, -1.75)) + _xy(second, 28.0, -1.75) + assert lanes[-1].haltPointAhead(Vector(29, -1.75)) is None + + +# --- (same signal, other road) --- + + +def test_signal_reference_keeps_this_roads_placement(tmp_path): + network = load_network(tmp_path, MAP_SIGNAL_REFERENCE) + west = road_by_id(network, 1) + east = road_by_id(network, 2) + canonical = signal_on(west, 201) + clone = signal_on(east, 201) + assert canonical.type == clone.type == "1000001" + assert canonical.s == 18.0 + assert clone.s == 1.5 + assert clone.orientation == "-" + assert canonical.uid != clone.uid + + +def test_signal_reference_on_exit_does_not_stop_outgoing(tmp_path): + """Turning onto the east road must not halt at that road's s≈0 clone.""" + network = load_network(tmp_path, MAP_SIGNAL_REFERENCE) + west = road_by_id(network, 1) + east = road_by_id(network, 2) + west_lanes = lanes_by_id(west) + east_lanes = lanes_by_id(east) + canonical = signal_on(west, 201) + clone = signal_on(east, 201) + + _xy(canonical.stoppingPointOn(west_lanes[-1]), 20.0, -1.75) + assert canonical.stoppingPointOn(west_lanes[1]) is None + + # Backward on the east road is still arriving at the junction. + _xy(clone.stoppingPointOn(east_lanes[1]), 32.0, 1.75) + # Forward on the east road just left the junction. + assert clone.stoppingPointOn(east_lanes[-1]) is None + assert east_lanes[-1].haltPointAhead(Vector(33.0, -1.75)) is None + + +def test_signal_reference_connector_has_no_invented_halt(tmp_path): + network = load_network(tmp_path, MAP_SIGNAL_REFERENCE) + connector = next(road for road in network.connectingRoads if road.id == 10) + for sig in connector.signals: + for lane in connector.lanes: + assert sig.stoppingPointOn(lane) is None + + +def test_carla_dummy_signal_reference_on_exit(tmp_path): + """Town10HD-style: dummy validity on the reference still must not stop the exit.""" + network = load_network(tmp_path, MAP_CARLA_SIGNAL_REFERENCE) + east = road_by_id(network, 2) + lanes = lanes_by_id(east) + clone = signal_on(east, 201) + assert clone.validity == (0, 0) + assert clone.orientation == "+" # lamp facing; would wrongly pick the exit + _xy(clone.stoppingPointOn(lanes[1]), 32.0, 1.75) + assert clone.stoppingPointOn(lanes[-1]) is None + assert lanes[-1].haltPointAhead(Vector(33.0, -1.75)) is None + + +# --- CARLA / RoadRunner undefined (dummy 0–0, pole only) --- + + +def test_carla_twoway_arriving_side_halts_leaving_side_does_not(tmp_path): + network = load_network(tmp_path, MAP_CARLA_TWOWAY) + west = road_by_id(network, 1) + east = road_by_id(network, 2) + west_lanes = lanes_by_id(west) + east_lanes = lanes_by_id(east) + west_light = signal_on(west, 362) + east_light = signal_on(east, 360) + + assert west_light.validity == (0, 0) + assert east_light.validity == (0, 0) + assert west_light.stoppingS == 20.0 + assert east_light.stoppingS == 0.0 + + # West +s is arriving at the junction. + _xy(west_light.stoppingPointOn(west_lanes[-1]), 20.0, -1.75) + assert west_light.stoppingPointOn(west_lanes[1]) is None + + # East +s is the road you turn into; −s is the opposite approach. + assert east_light.stoppingPointOn(east_lanes[-1]) is None + _xy(east_light.stoppingPointOn(east_lanes[1]), 32.0, 1.75) + + +def test_carla_green_light_then_no_stop_on_exit_lane(tmp_path): + """After the west stop, haltPointAhead on the east exit lane is empty.""" + network = load_network(tmp_path, MAP_CARLA_TWOWAY) + west = road_by_id(network, 1) + east = road_by_id(network, 2) + west_fwd = lanes_by_id(west)[-1] + east_fwd = lanes_by_id(east)[-1] + _xy(west_fwd.haltPointAhead(Vector(1.0, -1.75)), 20.0, -1.75) + assert east_fwd.haltPointAhead(Vector(33.0, -1.75)) is None + # Exactly at the east road start (s=0) must not count as a stop either. + assert east_fwd.haltPointAhead(Vector(32.0, -1.75)) is None + + +def test_carla_connector_lights_do_not_invent_a_stop(tmp_path): + network = load_network(tmp_path, MAP_CARLA_TWOWAY) + for road in network.connectingRoads: + for sig in road.signals: + assert sig.stoppingS is None + for lane in road.lanes: + assert sig.stoppingPointOn(lane) is None + + +@pytest.mark.slow +def test_town01_exit_lane_does_not_halt_at_entry(getAssetPath): + """Town01 intersection26: road 1 +s leaves the junction and must not stop there.""" + from scenic.core.geometry import TriangulationError + + path = Path(getAssetPath("maps/CARLA/Town01.xodr")) + if not path.exists(): + pytest.skip("Town01.xodr not shipped") + try: + network = Network.fromOpenDrive(path, ref_points=40, tolerance=0.05) + except TriangulationError: + pytest.skip("need better triangulation library") + + road1 = road_by_id(network, 1) + lanes = lanes_by_id(road1) + # OpenDRIVE +1 is −s (arriving at junction 26); −1 is +s (leaving it). + arriving = lanes[1] + leaving = lanes[-1] + entry_light = next( + sig + for sig in road1.signals + if sig.stoppingS is not None and abs(sig.stoppingS) <= 1e-4 + ) + assert entry_light.stoppingPointOn(arriving) is not None + assert entry_light.stoppingPointOn(leaving) is None + start = leaving.centerline.pointAlongBy(1.0) + assert leaving.haltPointAhead(start, lookahead=15.0) is None