mrtj

from typing import Iterable
from functools import reduce

def replace_all(text: str, table: Iterable[tuple[str, str]]) -> str:
    """ Replaces multiple strings in a string.

    Usage:
    ```python
        replace_all("Hello World", [("Hello", "Hi"), ("World", "Earth")])
        # 'Hi Earth'

mrtj

from typing import Iterable, Callable, Any
from functools import reduce, partial

def pipeline(functions: Iterable[Callable[[Any], Any]]) -> Callable[[Any], Any]:
  """ Creates a pipeline from a sequence of functions.

  Usage:
  ```python
      processor = pipeline([str.lower, str.split])
      processor("Hello World")

mrtj

#!/usr/bin/env python3

import boto3
import requests
import tqdm
from urllib.parse import urlparse

def stream_to_s3(
    source_url,
    target_url,

mrtj

import functools

def lazy_property(func):
    ''' Caches the return value of a function, and turns it into a property. 
    
    Intended to be used as a function decorator::
    
        >>> class Foo:
        >>>     @lazy_property
        >>>     def bar(self):

mrtj

import gi

gi.require_version('Gst', '1.0')
gi.require_version('GstRtspServer', '1.0')
from gi.repository import GLib, Gst, GstRtspServer

Gst.init(None)

import argparse
parser = argparse.ArgumentParser(description='GStreamer RTSP server test-launch')

mrtj

import panoramasdk
from backpack.annotation import (
    Point, LabelAnnotation, RectAnnotation, TimestampAnnotation,
    OpenCVImageAnnotationDriver, 
    PanoramaMediaAnnotationDriver
)

class Application(panoramasdk.node):

    def __init__(self):

mrtj

import panoramasdk
from backpack.kvs import KVSSkyLine, KVSFileCredentialsHandler

# You might want to read these values from Panorama application parameters
stream_region = 'us-east-1'
stream_name = 'panorama-video'

# The example Dockerfile writes static configuration variables to this file
# If you change the .env file path in the Dockerfile, you should change it also here
DOTENV_PATH = '/panorama/.env'

mrtj

mrtj

class Application(panoramasdk.node):
    
    # ...
    
    def process_media(self, stream):
        image_data, ratio = self.preprocess(stream.image, self.MODEL_INPUT_SIZE)
        inference_results = self.call(
            {self.MODEL_INPUT_NAME: image_data}, self.MODEL_NODE
        )
        self.process_results(inference_results, stream, ratio)

mrtj

from yolox_postprocess import demo_postprocess, multiclass_nms

class Application(panoramasdk.node):

    # ...
    
    def process_results(self, inference_results, stream, ratio):
        media_height, media_width, _ = stream.image.shape
        media_scale = np.asarray([media_width, media_height, media_width, media_height])
        for output in inference_results: