Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Redesign] Add type annotations to operator and component creator #278

Open
wants to merge 3 commits into
base: redesign
Choose a base branch
from
Open

[Redesign] Add type annotations to operator and component creator #278

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ab8d23e
Adding type annotations to operator_creater.py
HivaMohammadzadeh1 Jul 28, 2022
e03b4dd
Added type annotations of operator_creator.py
HivaMohammadzadeh1 Jul 28, 2022
b674f54
Adding annotations for component_creator.py
HivaMohammadzadeh1 Jul 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 25 additions & 17 deletions pylot/component_creator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,14 @@
from absl import flags

from erdos import Stream
from pylot import prediction
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
from pylot import prediction


import pylot.operator_creator
from pylot.drivers.sensor_setup import DepthCameraSetup, RGBCameraSetup, \
from pylot.drivers.sensor_setup import DepthCameraSetup, IMUSetup, LidarSetup, RGBCameraSetup, \
SegmentedCameraSetup, CameraSetup
from pylot.perception.camera_frame import CameraFrame
from pylot.perception.depth_frame import DepthFrame
from pylot.perception.detection import obstacle
from pylot.perception.detection.lane import Lane
from pylot.perception.messages import (ObstacleTrajectoriesMessageTuple,
ObstaclesMessageTuple,
Expand Down Expand Up @@ -438,14 +440,15 @@ def add_segmentation(center_camera_stream: Stream[CameraFrame],
return segmented_stream


def add_prediction(obstacles_tracking_stream,
vehicle_id_stream,
time_to_decision_stream,
camera_transform=None,
release_sensor_stream=None,
pose_stream=None,
point_cloud_stream=None,
lidar_setup=None):
def add_prediction(obstacles_tracking_stream: Stream[ObstaclesMessageTuple],
vehicle_id_stream: Stream[int],
release_sensor_stream: Stream[None],
point_cloud_stream: Stream[PointCloud],
lidar_setup: Stream[LidarSetup],
pose_stream: Optional[Stream[pylot.utils.Pose]] = None,
Camera_transform = None,
time_to_decision_stream: Optional[Stream[float]] = None
) -> Tuple[Stream[prediction], Stream[int], Stream]:
Comment on lines +443 to +451
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you make sure that the arguments remain in the same order as before?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, the return type should be Tuple[Stream[List[ObstaclePrediction]], Stream[SegmentedFrame], Stream[None]]

"""Adds prediction operators.

Args:
Expand Down Expand Up @@ -508,9 +511,14 @@ def add_prediction(obstacles_tracking_stream,
notify_reading_stream)


def add_planning(goal_location, pose_stream, prediction_stream,
traffic_lights_stream, lanes_stream, open_drive_stream,
global_trajectory_stream, time_to_decision_stream):
def add_planning(goal_location,
traffic_lights_stream: Stream[TrafficLight],
lanes_stream: Stream[Lane],
open_drive_stream: Stream,
global_trajectory_stream,
prediction_stream: Stream[prediction],
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
prediction_stream: Stream[prediction],
prediction_stream: Stream[List[ObstaclePrediction]],

time_to_decision_stream: Optional[Stream[float]] = None,
pose_stream: Optional[Stream[pylot.utils.Pose]] = None) -> Stream:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please annotate the return type of the stream -- it should carry waypoints.

Comment on lines +514 to +521
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please make sure the arguments stay in the same order.

"""Adds planning operators.

Args:
Expand Down Expand Up @@ -544,10 +552,10 @@ def add_planning(goal_location, pose_stream, prediction_stream,
return waypoints_stream


def add_control(pose_stream,
waypoints_stream,
ground_vehicle_id_stream=None,
perfect_obstacles_stream=None):
def add_control(waypoints_stream,
ground_vehicle_id_stream: Stream[int],
perfect_obstacles_stream: Stream[obstacle],
pose_stream: Optional[Stream[pylot.utils.Pose]] = None) -> Tuple[Stream, Stream]:
Comment on lines +555 to +558
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please make sure the arguments stay in the same order. Also, the return type is only a single stream.

"""Adds ego-vehicle control operators.

Args:
Expand Down Expand Up @@ -597,7 +605,7 @@ def add_control(pose_stream,
return control_stream


def add_evaluation(vehicle_id_stream, pose_stream, imu_stream):
def add_evaluation(vehicle_id_stream: Stream[int], pose_stream: Optional[Stream[pylot.utils.Pose]] = None, imu_stream: Optional[Stream[IMUSetup]] = None) -> Stream:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Return type is None. None of the arguments should be optional. imu_stream is Stream[IMUMessageTuple].

if FLAGS.evaluation:
logger.debug('Adding collision logging sensor...')
collision_stream = pylot.operator_creator.add_collision_sensor(
Expand Down
Loading