Feature/add the stream manager mocks

rr/stream/stream_manager.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+
+#  Copyright (C) 2021 RidgeRun, LLC (http://www.ridgerun.com)
+#  Authors: Daniel Chaves <[email protected]>
+#           Marisol Zeledon <[email protected]>
+
+from rr.ai.ai_manager import AIManagerOnNewImage
+from rr.gstreamer.media_manager import MediaManager
+
+model = "/opt/edge_ai_apps/models/detection/TFL-OD-200-ssd-mobV1-coco-mlperf-300x300/"
+disp_width = 2040
+disp_height = 1920
+
+
+class OnNewImage():
+    def __init__(self, ai_manager):
+        self.ai_manager = ai_manager
+
+    def __call__(self, image):
+        self.ai_manager.process_image(image, model, disp_width, disp_height)
+
+
+class StreamManagerError(RuntimeError):
+    pass
+
+
+class StreamManager():
+    """
+    Class that orchestrate the stream interoperations
+
+    Attributes
+    ----------
+
+    Methods
+    -------
+    """
+
+    def __init__(self, ai_manager, media_manager):
+        """
+        Constructor for the Stream Manager object
+        """
+
+        self.ai_manager = ai_manager
+        self.media_manager = media_manager
+
+        cb = OnNewImage(ai_manager)
+        self.media_manager.install_callback(cb)
+
+    def start(self):
+        """
+        Start the stream server
+        """
+        self.media_manger.play()

tests/stream/test_stream_manager.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+
+#  Copyright (C) 2021 RidgeRun, LLC (http://www.ridgerun.com)
+#  Authors: Daniel Chaves <[email protected]>
+#           Marisol Zeledon <[email protected]>
+
+import unittest
+from unittest.mock import MagicMock
+
+from rr.stream.stream_manager import StreamManager
+
+
+class MockAiManager:
+    def __init__(self):
+        self.process = MagicMock()
+
+
+class MockImage:
+    pass
+
+
+model = "/opt/edge_ai_apps/models/detection/TFL-OD-200-ssd-mobV1-coco-mlperf-300x300/"
+disp_width = 2040
+disp_height = 1920
+mock_image = MockImage()
+
+
+class MockMediaManager:
+    def install_callback(self, cb):
+        self.cb = cb
+
+    def play(self):
+        self.cb(mock_image, model, disp_width, disp_height)
+
+
+class TestStreamManager(unittest.TestCase):
+    def test_success(self):
+        media_manager = MockMediaManager()
+        ai_manager = MockAiManager()
+
+        stream_manager = StreamManager(ai_manager, media_manager)
+
+        stream_manager.start()
+
+        ai_manager.process.assert_called_with(
+            mock_image, model, disp_width, disp_height)
+
+
+if __name__ == '__main__':
+    unittest.main()