simulation.src.simulation_groundtruth.src.groundtruth package

Subpackages

• simulation.src.simulation_groundtruth.src.groundtruth.test package
  • Submodules
  • simulation.src.simulation_groundtruth.src.groundtruth.test.mock_node module
    • create_straight_section()
    • create_parking_section()
    • create_intersection()
    • GroundtruthMockNode
      • GroundtruthMockNode._load_road()
  • simulation.src.simulation_groundtruth.src.groundtruth.test.road_section_mocks module
    • _generic_mock()
    • mock_generic_section()
    • _mock_spots()
    • _mock_lots()
    • mock_parking_section()
    • mock_intersection()
  • Module contents

Submodules

simulation.src.simulation_groundtruth.src.groundtruth.groundtruth module

Functions:

_lane_msg_from_lines(left, middle, right)

Create a lane message from three lines.

Classes:

IntersectionTuple(turn, rule, south, west, ...)	Container for information about an intersection.
Groundtruth(road)	Provide functionality to extract groundtruth information as ROS messages.

_lane_msg_from_lines(left: Line, middle: Line, right: Line) → Lane

Create a lane message from three lines.

class IntersectionTuple(turn, rule, south, west, east, north)

Bases: tuple

Container for information about an intersection.

Methods:

_asdict()	Return a new dict which maps field names to their values.
_make(iterable)	Make a new IntersectionTuple object from a sequence or iterable
_replace(**kwds)	Return a new IntersectionTuple object replacing specified fields with new values

Attributes:

_field_defaults
_fields
east	Lane markings on the right side when coming from south.
north	Lane markings straight ahead when coming from south.
rule	Priority rule.
south	Lane markings approaching the intersection.
turn	Direction the road follows.
west	Lane markings on the left side when coming from south.

_asdict()

Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('turn', 'rule', 'south', 'west', 'east', 'north')

classmethod _make(iterable)

Make a new IntersectionTuple object from a sequence or iterable

_replace(**kwds)

Return a new IntersectionTuple object replacing specified fields with new values

east

Lane markings on the right side when coming from south.

Type:

LaneMsg

north

Lane markings straight ahead when coming from south.

Type:

LaneMsg

rule

Priority rule.

Type:

int

south

Lane markings approaching the intersection.

Type:

LaneMsg

turn

Direction the road follows.

Type:

int

west

Lane markings on the left side when coming from south.

Type:

LaneMsg

class Groundtruth(road: Road)

Bases: object

Provide functionality to extract groundtruth information as ROS messages.

Attributes:

road	Store the current road.

Methods:

get_section_msgs()	Section message for all sections.
get_lane_msg(id)	Lane message for requested road section.
get_intersection_msg(id)	Intersection tuple for requested road section.
get_parking_msg(id)	Parking message for the left and right side of the requested road section.
get_surface_marking_msgs(id)	Labeled polygon msg for each surface marking in the requested road section.
get_traffic_sign_msgs(id)	Labeled polygon msg for each traffic sign in the requested road section.

road: Road

Store the current road.

get_section_msgs() → list[Section]

Section message for all sections.

get_lane_msg(id: int) → Lane

Lane message for requested road section.

Parameters:

id – section id

get_intersection_msg(id: int) → IntersectionTuple

Intersection tuple for requested road section.

Parameters:

id – section id

get_parking_msg(id: int) → tuple[Parking, Parking]

Parking message for the left and right side of the requested road section.

Parameters:

id – section id

get_surface_marking_msgs(id: int) → list[LabeledPolygon]

Labeled polygon msg for each surface marking in the requested road section.

Parameters:

id – section id

get_traffic_sign_msgs(id: int) → list[LabeledPolygon]

Labeled polygon msg for each traffic sign in the requested road section.

Parameters:

id – section id

simulation.src.simulation_groundtruth.src.groundtruth.labeled_polygon_service module

Classes:

LabeledPolygonService(node, topic, callback)

Small wrapper class to easily create ROS services that offer LabeledPolygonMsgs.

class LabeledPolygonService(node: Node, topic: str, callback: Callable[[int], list[LabeledPolygon]])

Bases: object

Small wrapper class to easily create ROS services that offer LabeledPolygonMsgs.

Attributes:

node
topic
callback

Methods:

request(request, response)	Answer service request.
shutdown()

node: Node

topic: str

callback: Callable[[int], list[LabeledPolygon]]

request(request: LabeledPolygonSrv_Request, response: LabeledPolygonSrv_Response) → LabeledPolygonSrv_Response

Answer service request.

shutdown()

simulation.src.simulation_groundtruth.src.groundtruth.node module

Data:

_GAZEBO_STARTUP_DELAY

Seconds to wait for the gz-sim server before spawning tile models (see GroundtruthNode._ensure_gazebo_startup()).

Classes:

GroundtruthNode([name])

ROS node providing services to access the road's groundtruth.

Functions:

main([args])

Console-script entry point for the groundtruth node.

_GAZEBO_STARTUP_DELAY = 2.0

Seconds to wait for the gz-sim server before spawning tile models (see GroundtruthNode._ensure_gazebo_startup()).

class GroundtruthNode(name='groundtruth_node')

Bases: NodeBase

ROS node providing services to access the road’s groundtruth.

groundtruth

Stores road sections and allows to access groundtruth information.

Type:

Groundtruth

renderer

Render and update the road in Gazebo.

Type:

Renderer

section_srv

ROS service to access section ids and types.

lane_srv

ROS service to access the lanes of a road section.

parking_srv

ROS service to access parking spots and other parking information.

intersection_srv

ROS service to access information of an intersection.

Methods:

_load_road()	Load road sections from current road.
_ensure_gazebo_startup()	Wait until Gazebo is ready to accept spawned tile models.
start()	Called when activating the node.
stop()	Turn off.
load()	Load the groundtruth including renderer and object controller.
get_sections(request, response)	Answer section service request.
get_lanes(request, response)	Answer lane service request.
get_parking(request, response)	Answer parking service request.
get_intersection(request, response)	Answer intersection service request.
receive_model_states(model_states)
_spawn_model(model_xml)	Spawn a model in Gazebo.
_remove_model(name)	Remove a model from Gazebo.
_pause_gazebo()	Pause Gazebo.
_unpause_gazebo()	Request to start Gazebo.
receive_reload_request(_)	Receive requests to reload the world.
receive_interrupt_request(_)	Receive requests to stop rendering processes.
update_groundtruth_status([status, ...])

_load_road() → Road

Load road sections from current road.

_ensure_gazebo_startup()

Wait until Gazebo is ready to accept spawned tile models.

gz-sim Harmonic has no gazebo_msgs/ModelStates topic, so the old spawn-a-probe-and-wait-for-it-in-ModelStates handshake cannot be used. The tile models are spawned via the WorldPluginLink topic, which is processed once the gz-sim server is running; a fixed short delay is sufficient and avoids depending on a topic that never publishes. The world is brought up before this node anyway.

start()

Called when activating the node.

stop()

Turn off.

load()

Load the groundtruth including renderer and object controller.

get_sections(request: SectionSrv_Request, response: SectionSrv_Response) → SectionSrv_Response

Answer section service request.

get_lanes(request: LaneSrv_Request, response: LaneSrv_Response) → LaneSrv_Response

Answer lane service request.

get_parking(request: ParkingSrv_Request, response: ParkingSrv_Response) → ParkingSrv_Response

Answer parking service request.

get_intersection(request: IntersectionSrv_Request, response: IntersectionSrv_Response) → IntersectionSrv_Response

Answer intersection service request.

receive_model_states(model_states: None)

_spawn_model(model_xml: str)

Spawn a model in Gazebo.

Parameters:

model_xml – XML String that defines the model. (Without sdf declaration!)

_remove_model(name)

Remove a model from Gazebo.

Parameters:

name – Name of the model in Gazebo.

_pause_gazebo()

Pause Gazebo.

_unpause_gazebo()

Request to start Gazebo.

receive_reload_request(_: Empty)

Receive requests to reload the world.

receive_interrupt_request(_: Empty)

Receive requests to stop rendering processes.

update_groundtruth_status(status=None, processed_tiles=None, number_of_tiles=None)

main(args=None)

Console-script entry point for the groundtruth node.

simulation.src.simulation_groundtruth.src.groundtruth.object_controller_node module

Classes:

ObjectControllerNode(*[, name])

ROS node to place obstacles and other objects (e.g. traffic signs) in Gazebo.

Functions:

main([args])

Console-script entry point for the object controller node.

class ObjectControllerNode(*, name='object_controller_node')

Bases: NodeBase

ROS node to place obstacles and other objects (e.g. traffic signs) in Gazebo.

Attributes:

active
road
start_box
car_state
set_model_pose_publishers
obstacle_prefix
sign_prefix
start_gate
model_names
obstacles
dynamic_obstacles
signs

Methods:

_now()
receive_model_states(model_states)
receive_car_state(car_state)
receive_status(status_msg)
receive_reload_request(_)	Receive requests to reload the world.
receive_mission_mode(mission_mode)
_load_road()	Load road sections from current road.
load()	Reload all objects.
start()	Called when activating the node.
stop()	Called when deactivating or shutting down the node.
_set_model_pose(name, pose)
_spawn_model(model_xml)	Spawn a model in Gazebo.
_remove_model(name)	Remove a model from Gazebo.
_load_and_name(*, objects, prefix)
_remove_old_objects(model_names)
update()	Update the object controller.
get_obstacle_msgs(id)	Labeled polygon msg for each obstacle in the requested road section.

active: bool = False

road: Road = Road(_name=None, _seed=None, use_seed=True, sections=[], length=0, start_box=False)

start_box: StartBox

car_state: CarState = gazebo_simulation_msgs.msg.CarState(pose=geometry_msgs.msg.Pose(position=geometry_msgs.msg.Point(x=0.0, y=0.0, z=0.0), orientation=geometry_msgs.msg.Quaternion(x=0.0, y=0.0, z=0.0, w=1.0)), twist=geometry_msgs.msg.Twist(linear=geometry_msgs.msg.Vector3(x=0.0, y=0.0, z=0.0), angular=geometry_msgs.msg.Vector3(x=0.0, y=0.0, z=0.0)), frame=geometry_msgs.msg.Polygon(points=[]), view_cone=geometry_msgs.msg.Polygon(points=[]))

set_model_pose_publishers: dict = {}

obstacle_prefix: str = 'obstacle'

sign_prefix: str = 'sign'

start_gate: float | None = None

model_names = None

_now() → float

receive_model_states(model_states: None)

receive_car_state(car_state: CarState)

receive_status(status_msg: GroundtruthStatus)

receive_reload_request(_: Empty)

Receive requests to reload the world.

receive_mission_mode(mission_mode: MissionMode)

_load_road() → Road

Load road sections from current road.

load()

Reload all objects.

start()

Called when activating the node.

stop()

Called when deactivating or shutting down the node.

_set_model_pose(name, pose: Pose)

_spawn_model(model_xml: str)

Spawn a model in Gazebo.

Parameters:

model_xml – XML String that defines the model. (Without sdf declaration!)

_remove_model(name)

Remove a model from Gazebo.

Parameters:

name – Name of the model in Gazebo.

_load_and_name(*, objects: Iterable[Any], prefix: str)

property obstacles: list[tuple[str, StaticObstacle]]

property dynamic_obstacles: list[tuple[str, DynamicObstacle]]

property signs: list[tuple[str, TrafficSign]]

_remove_old_objects(model_names: list[str])

update()

Update the object controller.

Use this function to move dynamic obstacles around and open the startbox gate.

Parameters:

car_state – Current car state containing the position of the car.

get_obstacle_msgs(id: int) → list[LabeledPolygon]

Labeled polygon msg for each obstacle in the requested road section.

Parameters:

id – section id

_parameters: dict

_publishers: List[Publisher]

_subscriptions: List[Subscription]

_clients: List[Client]

_services: List[Service]

_timers: List[Timer]

_guards: List[GuardCondition]

__waitables: List[Waitable]

_pre_set_parameters_callbacks: List[Callable[[List[Parameter]], List[Parameter]]]

_on_set_parameters_callbacks: List[Callable[[List[Parameter]], SetParametersResult]]

_post_set_parameters_callbacks: List[Callable[[List[Parameter]], None]]

main(args=None)

Console-script entry point for the object controller node.

simulation.src.simulation_groundtruth.src.groundtruth.renderer module

Functions:

current_pretty_time()

Classes:

PreviousRendering(seed, tile_size, float], ...)
Renderer(road, remove_model, spawn_model, ...)

current_pretty_time()

class PreviousRendering(seed: str, tile_size: tuple[float, float], tile_resolution: tuple[float, float], tiles: dict[str, str] = <factory>, time: float = <factory>, pretty_time: str = <factory>)

Bases: object

Attributes:

DEFAULT_FILE_NAME
seed
tile_size
tile_resolution
tiles
time
pretty_time

Methods:

load(dir)	Load information about previous rendering from file.
save(dir)	Store information about the current rendering.
delete(dir)	Delete directory containing all materials of previous renderings.

DEFAULT_FILE_NAME = 'previous_rendering.yaml'

seed: str

tile_size: tuple[float, float]

tile_resolution: tuple[float, float]

tiles: dict[str, str]

time: float

pretty_time: str

classmethod load(dir: str) → PreviousRendering

Load information about previous rendering from file.

Parameters:

dir – Path to directory that contains previous rendering.

save(dir: str)

Store information about the current rendering.

Parameters:

dir – Path to directory that contains current rendering.

classmethod delete(dir: str)

Delete directory containing all materials of previous renderings.

Parameters:

dir – Path to directory that contains previous rendering.

class Renderer(road: simulation.utils.road.road.Road, remove_model: collections.abc.Callable[[str], None], spawn_model: collections.abc.Callable[[str], None], pause_gazebo: collections.abc.Callable[[], None], unpause_gazebo: collections.abc.Callable[[], None], info_callback: collections.abc.Callable[[int, int, int], None], logger: rclpy.impl.rcutils_logger.RcutilsLogger = None, tile_size: kitcar_utils.geometry.vector.Vector = Vector(2.0, 2.0, 0.0), tile_resolution: kitcar_utils.geometry.vector.Vector = Vector(512.0, 512.0, 0.0), force_reload: bool = False)

Bases: object

Attributes:

road
remove_model
spawn_model
pause_gazebo
unpause_gazebo
info_callback	Function that is called, when the renderer's state changes.
logger
tile_size
tile_resolution
force_reload
roads_path
materials_path
prev_rendering

Methods:

_prev_rendering_available()
_load_prev_tiles()
save_state(tiles)
_create_new_tiles()
_remove_displayed_tiles(model_names)
interrupt()
load(model_names)

road: Road

remove_model: Callable[[str], None]

spawn_model: Callable[[str], None]

pause_gazebo: Callable[[], None]

unpause_gazebo: Callable[[], None]

info_callback: Callable[[int, int, int], None]

Function that is called, when the renderer’s state changes.

logger: RcutilsLogger = None

tile_size: Vector = Vector(2.0, 2.0, 0.0)

tile_resolution: Vector = Vector(512.0, 512.0, 0.0)

force_reload: bool = False

property roads_path: str

property materials_path: str

property prev_rendering: PreviousRendering

_prev_rendering_available() → bool

_load_prev_tiles() → list[Tile]

save_state(tiles: list[Tile])

_create_new_tiles() → list[Tile]

_remove_displayed_tiles(model_names: list[str])

interrupt()

load(model_names: list[str])

simulation.src.simulation_groundtruth.src.groundtruth.visualization module

Node that publishes groundtruth information for rviz.

Classes:

GroundtruthVisualizationNode()

ROS node to visualize groundtruth objects in RVIZ.

Functions:

main([args])

Console-script entry point for the groundtruth visualization node.

class GroundtruthVisualizationNode

Bases: NodeBase

ROS node to visualize groundtruth objects in RVIZ.

publishers

Automatically generated dictionary containing all publishers. A publisher is created for every visualization topic listed in the parameter file.

Type:

Dict[str, Publisher]

get_sections, get_lane, get_parking, get_obstacles, get_intersection

Can be called for groundtruth information.

Type:

Client

Methods:

start()	Start the node by creating publishers and service clients.
stop()	Stop the node by destroying publishers.
_call(client, **kwargs)	Call a service synchronously and return the response.
_publish_point_marker(points, ...[, ns])	Publish an RVIZ marker on the publisher's topic.
_update_labeled_polygon_topic(client, ...)	Publish polygons for all elements.
_show_lane_markers()	Publish markers for all lane markings of road.
_show_parking_markers()	Publish markers for all parking lines and spots of road.
_show_intersection_markers()	Publish lanes and surface markings of all intersections.
update()	Update all markers.

start()

Start the node by creating publishers and service clients.

stop()

Stop the node by destroying publishers.

static _call(client: Client, **kwargs)

Call a service synchronously and return the response.

_publish_point_marker(points: list[Point32], publisher_name: str, id: int, rgba: list[float], ns='simulation/groundtruth')

Publish an RVIZ marker on the publisher’s topic.

Parameters:

• points – Points to be published.

• publisher_name – Key of the publisher in the publisher dictionary.

• id – RVIZ marker id.

• rgba – List of the marker color.

• ns – RVIZ namespace of the marker.

_update_labeled_polygon_topic(client: Client, publisher_name: str, marker_color: tuple[float, float, float, float])

Publish polygons for all elements.

_show_lane_markers()

Publish markers for all lane markings of road.

_show_parking_markers()

Publish markers for all parking lines and spots of road.

_show_intersection_markers()

Publish lanes and surface markings of all intersections.

update()

Update all markers.

main(args=None)

Console-script entry point for the groundtruth visualization node.

Module contents