code: """Implements an operator that detects obstacles."""
  import logging
  import time
  
  import erdos
  
  import numpy as np
  
  import pylot.utils
  from pylot.perception.detection.obstacle import Obstacle
  from pylot.perception.detection.utils import BoundingBox2D, \
      OBSTACLE_LABELS, load_coco_bbox_colors, load_coco_labels
  from pylot.perception.messages import ObstaclesMessage
  
  import tensorflow as tf
  
  
  class DetectionOperator(erdos.Operator):
      """Detects obstacles using a TensorFlow model.
  
      The operator receives frames on a camera stream, and runs a model for each
      frame.
  
      Args:
          camera_stream (:py:class:`erdos.ReadStream`): The stream on which
              camera frames are received.
          obstacles_stream (:py:class:`erdos.WriteStream`): Stream on which the
              operator sends
              :py:class:`~pylot.perception.messages.ObstaclesMessage` messages.
          model_path(:obj:`str`): Path to the model pb file.
          flags (absl.flags): Object to be used to access absl flags.
      """
      def __init__(self, camera_stream, time_to_decision_stream,
                   obstacles_stream, model_path, flags):
          camera_stream.add_callback(self.on_msg_camera_stream,
                                     [obstacles_stream])
          time_to_decision_stream.add_callback(self.on_time_to_decision_update)
          self._flags = flags
          self._logger = erdos.utils.setup_logging(self.config.name,
                                                   self.config.log_file_name)
          self._detection_graph = tf.Graph()
          # Load the model from the model file.
          pylot.utils.set_tf_loglevel(logging.ERROR)
          with self._detection_graph.as_default():
              od_graph_def = tf.GraphDef()
              with tf.gfile.GFile(model_path, 'rb') as fid:
                  serialized_graph = fid.read()
                  od_graph_def.ParseFromString(serialized_graph)
                  tf.import_graph_def(od_graph_def, name='')
  
          self._gpu_options = tf.GPUOptions(
              per_process_gpu_memory_fraction=flags.
              obstacle_detection_gpu_memory_fraction)
          # Create a TensorFlow session.
          self._tf_session = tf.Session(
              graph=self._detection_graph,
              config=tf.ConfigProto(gpu_options=self._gpu_options))
          # Get the tensors we're interested in.
          self._image_tensor = self._detection_graph.get_tensor_by_name(
              'image_tensor:0')
          self._detection_boxes = self._detection_graph.get_tensor_by_name(
              'detection_boxes:0')
          self._detection_scores = self._detection_graph.get_tensor_by_name(
              'detection_scores:0')
          self._detection_classes = self._detection_graph.get_tensor_by_name(
              'detection_classes:0')
          self._num_detections = self._detection_graph.get_tensor_by_name(
              'num_detections:0')
          self._coco_labels = load_coco_labels(self._flags.path_coco_labels)
          self._bbox_colors = load_coco_bbox_colors(self._coco_labels)
          # Unique bounding box id. Incremented for each bounding box.
          self._unique_id = 0
          # Serve some junk image to load up the model.
          self.__run_model(np.zeros((108, 192, 3)))
  
      @staticmethod
      def connect(camera_stream, time_to_decision_stream):
          """Connects the operator to other streams.
  
          Args:
              camera_stream (:py:class:`erdos.ReadStream`): The stream on which
                  camera frames are received.
  
          Returns:
              :py:class:`erdos.WriteStream`: Stream on which the operator sends
              :py:class:`~pylot.perception.messages.ObstaclesMessage` messages.
          """
          obstacles_stream = erdos.WriteStream()
          return [obstacles_stream]
  
      def on_time_to_decision_update(self, msg):
          self._logger.debug('@{}: {} received ttd update {}'.format(
              msg.timestamp, self.config.name, msg))
  
      @erdos.profile_method()
      def on_msg_camera_stream(self, msg, obstacles_stream):
          """Invoked whenever a frame message is received on the stream.
  
          Args:
              msg (:py:class:`~pylot.perception.messages.FrameMessage`): Message
                  received.
              obstacles_stream (:py:class:`erdos.WriteStream`): Stream on which
                  the operator sends
                  :py:class:`~pylot.perception.messages.ObstaclesMessage`
                  messages.
          """
          self._logger.debug('@{}: {} received message'.format(
              msg.timestamp, self.config.name))
          start_time = time.time()
          # The models expect BGR images.
          assert msg.frame.encoding == 'BGR', 'Expects BGR frames'
          num_detections, res_boxes, res_scores, res_classes = self.__run_model(
              msg.frame.frame)
          obstacles = []
          for i in range(0, num_detections):
              if res_classes[i] in self._coco_labels:
                  if (res_scores[i] >=
                          self._flags.obstacle_detection_min_score_threshold):
                      if (self._coco_labels[res_classes[i]] in OBSTACLE_LABELS):
                          obstacles.append(
                              Obstacle(BoundingBox2D(
                                  int(res_boxes[i][1] *
                                      msg.frame.camera_setup.width),
                                  int(res_boxes[i][3] *
                                      msg.frame.camera_setup.width),
                                  int(res_boxes[i][0] *
                                      msg.frame.camera_setup.height),
                                  int(res_boxes[i][2] *
                                      msg.frame.camera_setup.height)),
                                       res_scores[i],
                                       self._coco_labels[res_classes[i]],
                                       id=self._unique_id))
                          self._unique_id += 1
                      else:
                          self._logger.warning(
                              'Ignoring non essential detection {}'.format(
                                  self._coco_labels[res_classes[i]]))
              else:
                  self._logger.warning('Filtering unknown class: {}'.format(
                      res_classes[i]))
  
          self._logger.debug('@{}: {} obstacles: {}'.format(
              msg.timestamp, self.config.name, obstacles))
  
          # Get runtime in ms.
          runtime = (time.time() - start_time) * 1000
          # Send out obstacles.
          obstacles_stream.send(
              ObstaclesMessage(msg.timestamp, obstacles, runtime))
          obstacles_stream.send(erdos.WatermarkMessage(msg.timestamp))
  
          if self._flags.log_detector_output:
              msg.frame.annotate_with_bounding_boxes(msg.timestamp, obstacles,
                                                     None, self._bbox_colors)
              msg.frame.save(msg.timestamp.coordinates[0], self._flags.data_path,
                             'detector-{}'.format(self.config.name))
  
      def __run_model(self, image_np):
          # Expand dimensions since the model expects images to have
          # shape: [1, None, None, 3]
          image_np_expanded = np.expand_dims(image_np, axis=0)
          (boxes, scores, classes, num_detections) = self._tf_session.run(
              [
                  self._detection_boxes, self._detection_scores,
                  self._detection_classes, self._num_detections
              ],
              feed_dict={self._image_tensor: image_np_expanded})
  
          num_detections = int(num_detections[0])
          res_classes = [int(cls) for cls in classes[0][:num_detections]]
          res_boxes = boxes[0][:num_detections]
          res_scores = scores[0][:num_detections]
          return num_detections, res_boxes, res_scores, res_classes
  
apis: numpy.zeros tensorflow.gfile.GFile tensorflow.Graph numpy.expand_dims tensorflow.import_graph_def tensorflow.GPUOptions tensorflow.ConfigProto tensorflow.GraphDef