streamlined the consumer set up

Author: mderner-ca

amazon_kinesis_video_consumer_library/kinesis_video_streams_parser.py

@@ -46,6 +46,8 @@
 import logging
 from threading import Thread
 import ebmlite
+from boto3 import Session
+from botocore.response import StreamingBody
 
 # Init the logger.
 log = logging.getLogger(__name__)
@@ -54,28 +56,25 @@
 class KvsConsumerLibrary(Thread):
     def __init__(
         self,
-        stream_name,
-        get_media_response_object,
+        kvs_stream_name: str,
         on_fragment_arrived,
         on_read_stream_complete,
         on_read_stream_exception,
     ):
-        """
-        Initialize the KVS media consumer library
-        """
         # Call the Thread class's init function
-        Thread.__init__(self)
+        super().__init__()
 
         # Used to trigger graceful exit of this thread
         self._stop_get_media = False
 
         # Init the local vars.
+        self.kvs_stream_name = kvs_stream_name
         log.info("Initilizing KvsConsumerLibrary...")
-        self.stream_name = stream_name
-        self.get_media_response_object = get_media_response_object
+
         self.on_fragment_arrived_callback = on_fragment_arrived
         self.on_read_stream_complete_callback = on_read_stream_complete
         self.on_read_stream_exception = on_read_stream_exception
+        self.__acquire_stream_buffer(kvs_stream_name)
 
         log.info("Loading EBMLlite MKV Schema....")
         self.schema = ebmlite.loadSchema("matroska.xml")
@@ -84,6 +83,23 @@ def __init__(
             raise KeyError("Could not find master element in Matroska schema")
         self.matroska_master_element_type = master
 
+    def __acquire_stream_buffer(self, kvs_stream_name: str):
+        session = Session(region_name='eu-west-2')
+        kvs_client = session.client("kinesisvideo")
+        response = kvs_client.get_data_endpoint(
+            StreamName=kvs_stream_name,
+            APIName="GET_MEDIA"
+        )
+        get_media_endpoint = response["DataEndpoint"]
+        kvs_media_client = session.client(
+            "kinesis-video-media", endpoint_url=get_media_endpoint
+        )
+        get_media_response_object = kvs_media_client.get_media(
+            StreamName=kvs_stream_name,
+            StartSelector={"StartSelectorType": "EARLIEST"}
+        )
+        self.kvs_streaming_buffer: StreamingBody = get_media_response_object["Payload"]
+
     def _get_ebml_header_elements(
         self, fragement_dom: ebmlite.Document
     ) -> list[ebmlite.Element]:
@@ -138,9 +154,6 @@ def run(self):
         """
 
         try:
-            # Get the steam botocore.response.StreamingBody object from
-            # the provided GetMedia response
-            kvs_streaming_buffer = self.get_media_response_object["Payload"]
 
             #########################################
             # Iterate through reading and parsing streaming body
@@ -153,7 +166,7 @@ def run(self):
 
             # Uses the StreamingBody object iterator to read in (default
             # 1024 byte) chunks from the streaming buffer.
-            for chunk in kvs_streaming_buffer:
+            for chunk in self.kvs_streaming_buffer:
                 if self._stop_get_media:
                     break
 
@@ -202,7 +215,7 @@ def run(self):
 
                     # Forward fragment to the on_fragment_arrived callback.
                     self.on_fragment_arrived_callback(
-                        self.stream_name,
+                        self.kvs_stream_name,
                         fragment_bytes,
                         fragment_dom,
                         fragment_receive_duration,
@@ -228,8 +241,8 @@ def run(self):
             #############################################
             # call the on_stream_read_complete() callback and exit the
             # thread.
-            self.on_read_stream_complete_callback(self.stream_name)
+            self.on_read_stream_complete_callback(self.kvs_stream_name)
 
         except Exception as err:
             # Pass any exceptions to exception callback.
-            self.on_read_stream_exception(self.stream_name, err)
+            self.on_read_stream_exception(self.kvs_stream_name, err)

kvs_consumer_library_example.py

@@ -78,27 +78,10 @@ def service_loop(self):
         ####################################################
         # Start an instance of the KvsConsumerLibrary reading in a Kinesis Video Stream
 
-        # Get the KVS Endpoint for the GetMedia Call for this stream
-        log.info(f"Getting KVS GetMedia Endpoint for stream: {KVS_STREAM01_NAME} ........")
-        get_media_endpoint = self._get_data_endpoint(KVS_STREAM01_NAME, "GET_MEDIA")
-
-        # Get the KVS Media client for the GetMedia API call
-        log.info(f"Initializing KVS Media client for stream: {KVS_STREAM01_NAME}........")
-        kvs_media_client = self.session.client(
-            "kinesis-video-media", endpoint_url=get_media_endpoint
-        )
-
-        # Make a KVS GetMedia API call with the desired KVS stream and StartSelector type and time bounding.
-        log.info(f"Requesting KVS GetMedia Response for stream: {KVS_STREAM01_NAME}........")
-        get_media_response = kvs_media_client.get_media(
-            StreamName=KVS_STREAM01_NAME, StartSelector={"StartSelectorType": "EARLIEST"}
-        )
-
         # Initialize an instance of the KvsConsumerLibrary, provide the GetMedia response and the required call-backs
         log.info(f"Starting KvsConsumerLibrary for stream: {KVS_STREAM01_NAME}........")
         my_stream01_consumer = KvsConsumerLibrary(
             KVS_STREAM01_NAME,
-            get_media_response,
             self.on_fragment_arrived,
             self.on_stream_read_complete,
             self.on_stream_read_exception,
@@ -301,15 +284,6 @@ def on_stream_read_exception(self, stream_name, error):
             f"####### ERROR: Exception on read stream: {stream_name}\n####### Fragment Tags:\n{self.last_good_fragment_tags}\nError Message:{error}"
         )
 
-    ####################################################
-    # KVS Helpers
-    def _get_data_endpoint(self, stream_name, api_name):
-        """
-        Convenience method to get the KVS client endpoint for specific API calls.
-        """
-        response = self.kvs_client.get_data_endpoint(StreamName=stream_name, APIName=api_name)
-        return response["DataEndpoint"]
-
 
 if __name__ == "__main__":
     """

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.13"
 dependencies = [
     "av>=17.1.0",
     "boto3>=1.43.28",
+    "botocore>=1.43.28",
     "ebmlite>=3.4.1",
     "librosa>=0.11.0",
     "numpy>=2.4.6",