Annotations: unified drawing api for opencv and panorama media

resources/kvs_log_configuration

@@ -0,0 +1,6 @@
+log4cplus.rootLogger=INFO, KvsConsoleAppender
+
+# KvsConsoleAppender:
+log4cplus.appender.KvsConsoleAppender=log4cplus::ConsoleAppender
+log4cplus.appender.KvsConsoleAppender.layout=log4cplus::PatternLayout
+log4cplus.appender.KvsConsoleAppender.layout.ConversionPattern=%p:KVSProducer:%m%n

src/annotation.py

@@ -0,0 +1,131 @@
+''' Utility functions for AWS Panorama development. '''
+
+from typing import List, Tuple, Optional, Any, Callable
+import datetime
+import logging
+from abc import ABC, abstractmethod
+from typing import NamedTuple
+import cv2
+
+class Point(NamedTuple):
+    ''' A point with both coordinates normalized to the [0; 1) range. '''
+    x: float
+    y: float
+
+    def scale(self, width, height):
+        ''' Scales this point to image coordinates. '''
+        return (int(self.x * width), int(self.y * height))
+
+    def in_image(self, img: 'np.array'):
+        return self.scale(width=img.shape[1], height=img.shape[0])
+
+
+
+class LabelAnnotation(NamedTuple):
+    ''' A label annotation. '''
+    point: Point
+    text: str
+
+
+class RectAnnotation(NamedTuple):
+    ''' A rectangle annotation. '''
+    point1: Point
+    point2: Point
+
+
+class TimestampAnnotation(LabelAnnotation):
+    ''' A timestamp annotation. '''
+    def __new__(cls, timestamp: Optional[datetime.datetime]=None, point: Point=Point(0.02, 0.04)):
+        timestamp = timestamp or datetime.datetime.now()
+        time_str = timestamp.strftime('%Y-%m-%d %H:%M:%S')
+        return LabelAnnotation.__new__(cls, point=point, text=time_str)
+
+
+
+class AnnotationDriver(ABC):
+    ''' Base class for annotating drawing drivers. 
+    
+    :param context: The context object to draw on. The type of this object is implementation
+        specific.
+    '''
+    def __init__(self, parent_logger: Optional[logging.Logger] = None) -> None:
+        self.logger = (
+            logging.getLogger(self.__class__.__name__) if parent_logger is None else
+            parent_logger.getChild(self.__class__.__name__)
+        )
+
+    def render(self, annotations: List, context: Any) -> Any:
+        for anno in annotations:
+            if isinstance(anno, LabelAnnotation):
+                self.add_label(anno, context)
+            elif isinstance(anno, RectAnnotation):
+                self.add_rect(anno, context)
+            else:
+                assert True, 'Unknown annotation type'
+        return context
+
+    @abstractmethod
+    def add_rect(self, rect: RectAnnotation, context: Any) -> None:
+        ''' Add a text label to the frame. '''
+        raise NotImplementedError
+
+    @abstractmethod
+    def add_label(self, label: LabelAnnotation, context: Any) -> None:
+        ''' Add a label to the frame. '''
+        raise NotImplementedError
+
+
+
+class PanoramaMediaAnnotationDriver(AnnotationDriver):
+    ''' AnnotationDriver implementation for panoramasdk.media type images. '''
+
+    def add_rect(self, rect: RectAnnotation, context: 'panoramasdk.media') -> None:
+        context.add_rect(rect.point1.x, rect.point1.y, rect.point2.x, rect.point2.y)
+
+    def add_label(self, label: LabelAnnotation, context: 'panoramasdk.media') -> None:
+        context.add_label(label.text, label.point.x, label.point.y)
+
+
+
+class OpenCVImageAnnotationDriver(AnnotationDriver):
+    ''' AnnotationDriver implementation for OpenCV images. 
+
+    :param context: The OpenCV image.
+    '''
+
+    DEFAULT_OPENCV_COLOR = (255, 255, 255)
+    DEFAULT_OPENCV_LINEWIDTH = 1
+    DEFAULT_OPENCV_FONT = cv2.FONT_HERSHEY_PLAIN
+    DEFAULT_OPENCV_FONT_SCALE = 1.0
+
+    def add_rect(self, rect: RectAnnotation, context: 'np.array') -> None:
+        cv2.rectangle(
+            context,
+            rect.point1.in_image(context),
+            rect.point2.in_image(context),
+            self.DEFAULT_OPENCV_COLOR,
+            self.DEFAULT_OPENCV_LINEWIDTH
+        )
+
+    def add_label(self, label: LabelAnnotation, context: 'np.array') -> None:
+        cv2.putText(
+            context,
+            label.text,
+            label.point.in_image(context),
+            self.DEFAULT_OPENCV_FONT,
+            self.DEFAULT_OPENCV_FONT_SCALE,
+            self.DEFAULT_OPENCV_COLOR
+        )
+
+
+if __name__ == '__main__':
+    import numpy as np
+    img = np.zeros((500, 500, 3), np.uint8)
+    annos = [
+        RectAnnotation(Point(0.1, 0.1), Point(0.9, 0.9)),
+        LabelAnnotation(Point(0.5, 0.5), 'Hello World'),
+        TimestampAnnotation()
+    ]
+    cv2driver = OpenCVImageAnnotationDriver()
+    cv2driver.render(annos, img)
+    cv2.imwrite('aw.png', img)

src/panorama.py → src/idcard.py

@@ -3,14 +3,14 @@
 import os
 import logging
 import datetime
-from typing import Dict
+from typing import Dict, Tuple
 
 import boto3
 
 class AutoIdentity:
     ''' AutoIdentity instance queries metadata of the current application instance.
 
-    The IAM policy associated with the Panorama Appplication Role of this app should grant
+    The IAM policy associated with the Panorama Application Role of this app should grant
     the execution of `panorama:ListApplicationInstances` operation.
 
     :param device_region: The AWS region where this Panorama appliance is registered.

src/kvs.py

@@ -93,8 +93,8 @@ def _get_pipeline(self, fps, width, height):
         self.logger.info(f'GStreamer pipeline definition:\n{pipeline_safe}')
         return pipeline
 
-    def _put_frame(self, frame, timestamp, show_timestamp):
-        result = super()._put_frame(frame, timestamp, show_timestamp)
+    def _put_frame(self, *args, **kwargs):
+        result = super()._put_frame(*args, **kwargs)
         self.credentials_handler.check_refresh()
         return result
 

src/spyglass.py

@@ -1,4 +1,7 @@
-''' SpyGlass can be used to send OpenCV frames to a GStreamer pipeline. '''
+''' SpyGlass can be used to send OpenCV frames to a GStreamer pipeline, and 
+annotation drivers unify the drawing API of different backends (for example, 
+OpenCV or panoramasdk.media).
+'''
 
 import os
 import time
@@ -7,16 +10,16 @@
 from collections import OrderedDict
 import datetime
 from enum import Enum
-from typing import Optional
+from typing import List, Tuple, Optional, Any, Callable
+import datetime
+from abc import ABC, abstractmethod
 
 import cv2
 from dotenv import find_dotenv, dotenv_values
 
-from .utils import add_timestamp
-
-class SpyGlass:
+class SpyGlass(ABC):
 
-    ''' Base class for sending OpenCV frames to a remote service using GStreamer.
+    ''' Abstract base class for sending OpenCV frames to a remote service using GStreamer.
 
     SpyGlass can be used to create programatically a video stream and send it to
     an external video ingestion service supported by GStreamer. Once the SpyGlass
@@ -124,6 +127,7 @@ def _check_gst_plugin(self, plugin_name: str) -> bool:
             )
             return False
 
+    @abstractmethod
     def _get_pipeline(self, fps: float, width: int, height: int) -> str:
         ''' Returns to GStreamer pipeline definition.
 
@@ -133,17 +137,14 @@ def _get_pipeline(self, fps: float, width: int, height: int) -> str:
             'SpyGlass._get_pipeline() should be implemented in subclasses'
         )
 
-    def _put_frame(self, frame, timestamp, show_timestamp):
+    def _put_frame(self, frame):
         size = (self._int_width, self._int_height)
         resized = cv2.resize(frame, size, interpolation=cv2.INTER_LINEAR)
         if not self._video_writer or not self._video_writer.isOpened():
             self._frame_log(lambda: self.logger.warning(
                 'Tried to write to cv2.VideoWriter but it is not opened'
             ))
             return False
-        if show_timestamp:
-            timestamp = timestamp or datetime.datetime.now()
-            add_timestamp(resized, timestamp=timestamp)
         self._video_writer.write(resized)
         return True
 
@@ -218,12 +219,12 @@ def _finish_warmup(self, frame):
         return fps, width, height
 
     @property
-    def state(self) -> SpyGlass.State:
+    def state(self) -> 'SpyGlass.State':
         ''' State of the SpyGlass. '''
         return self._state
 
     @state.setter
-    def state(self, state: SpyGlass.State) -> None:
+    def state(self, state: 'SpyGlass.State') -> None:
         ''' Set the state of the SpyGlass. '''
         self.logger.info(f'state = {state}')
         self._state = state
@@ -250,16 +251,10 @@ def start_streaming(self) -> None:
     def put(
         self,
         frame: 'np.array',
-        timestamp: Optional[datetime.datetime] = None,
-        show_timestamp: bool = False
     ) -> bool:
         ''' Put a frame to the video stream.
 
         :param frame: A numpy array of (height, width, 3) shape and of `np.uint8` type.
-        :param timestamp: If you have a presentation timestamp of the frame,
-            you can set it here. It will be sent downstream on the pipeline on a
-            best-effort basis.
-        :param show_timestamp: Show a timestamp on the video stream.
         :return: True if the frame was effectively put on the downstream pipeline.
         '''
 
@@ -297,7 +292,7 @@ def put(
             return False
 
         if self.state == SpyGlass.State.STREAMING:
-            return self._put_frame(frame, timestamp, show_timestamp)
+            return self._put_frame(frame)
 
         # Should not arrive here
         assert False, f'Unhandled SpyGlass state {self.state}'

src/timepiece.py

@@ -6,14 +6,23 @@
 import threading
 from collections import deque
 from itertools import islice
-from typing import List, Deque, Optional, Iterator, Dict, Any, Callable
+from typing import List, Deque, Optional, Iterator, Dict, Any, Callable, Tuple
 
 from dateutil.tz import tzlocal
 
 def local_now():
     ''' Returns the current time in local time zone. '''
     return datetime.datetime.now(tzlocal())
 
+
+def panorama_timestamp_to_datetime(panorama_ts: Tuple[int, int]) -> datetime.datetime:
+    ''' Converts panoramasdk.media.time_stamp (seconds, microsececonds)
+    tuple to python datetime.
+    '''
+    sec, microsec = panorama_ts
+    return datetime.datetime.fromtimestamp(sec + microsec / 1000000.0)
+
+
 class BaseTimer:
     ''' Base class for code execution time measuring timers.'''
 
@@ -463,6 +472,8 @@ def _calculate_stats(self):
 
 if __name__ == '__main__':
     import random
+    from concurrent.futures import ThreadPoolExecutor
+
     with StopWatch('root') as root:
         with root.child('task1', max_intervals=5) as task1:
             time.sleep(0.01)
@@ -474,8 +485,9 @@ def _calculate_stats(self):
                 time.sleep(0.09)
             with subtask1_1:
                 time.sleep(0.05)
-        with root.child('task2') as task2:
-            time.sleep(0.17)
+        for i in range(5):
+            with root.child('task2') as task2:
+                time.sleep(random.random() / 10)
     print(root)
 
     ticker = Ticker(max_intervals=20)
@@ -487,14 +499,31 @@ def _calculate_stats(self):
     print('\n')
 
     cb = lambda name: print(f'{name} was called at {datetime.datetime.now()}')
+    executor = ThreadPoolExecutor()
 
-    at_ = local_now() + datetime.timedelta(seconds=3)
-    atschedule = AtSchedule(at=at_, callback=cb, cbkwargs={'name': 'AtSchedule'})
+    at = local_now() + datetime.timedelta(seconds=3)
+    atschedule = AtSchedule(
+        at=at, 
+        callback=cb, 
+        cbkwargs={'name': 'AtSchedule'}, 
+        executor=executor
+    )
 
     iv = datetime.timedelta(seconds=1.35)
-    ivschedule = IntervalSchedule(interval=iv, callback=cb, cbkwargs={'name': 'IntervalSchedule'})
+    ivschedule = IntervalSchedule(
+        interval=iv, 
+        callback=cb, 
+        cbkwargs={'name': 'IntervalSchedule'},
+        executor=executor
+    )
+
+    ordinalschedule = OrdinalSchedule(
+        ordinal=17, 
+        callback=cb, 
+        cbkwargs={'name': 'OrdinalSchedule'},
+        executor=executor
+    )
 
-    ordinalschedule = OrdinalSchedule(ordinal=17, callback=cb, cbkwargs={'name': 'OrdinalSchedule'})
     alarmclock = AlarmClock([atschedule, ivschedule, ordinalschedule])
 
     for i in range(25*5):