Skip to content

Instantly share code, notes, and snippets.

Avatar

Janos Tolgyesi mrtj

View GitHub Profile
@mrtj
mrtj / stream_to_s3.py
Last active Jul 22, 2022
Download a file and stream it directly to an #aws S3 bucket without saving it locally
View stream_to_s3.py
#!/usr/bin/env python3
import boto3
import requests
import tqdm
from urllib.parse import urlparse
def stream_to_s3(
source_url,
target_url,
View lazy_property.py
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
mrtj / test-launch.py
Last active Apr 5, 2022
Python port of GStreamer RTSP Server test-launch.c
View test-launch.py
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
mrtj / backpack-annotation-example.py
Last active Jun 20, 2022
Example usage of backpack.annotation module
View backpack-annotation-example.py
import panoramasdk
from backpack.annotation import (
Point, LabelAnnotation, RectAnnotation, TimestampAnnotation,
OpenCVImageAnnotationDriver,
PanoramaMediaAnnotationDriver
)
class Application(panoramasdk.node):
def __init__(self):
@mrtj
mrtj / kvsskyline-usage.py
Last active Jun 20, 2022
Example usage of backpack.kvs.KVSSkyLine class
View kvsskyline-usage.py
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'
View panorama-process-media-fix.py
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)
View panorama-yolox-postprocess.py
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:
View panorama-yolox-preprocess.py
import cv2
class Application(panoramasdk.node):
# ...
def preprocess(self, img, input_size, swap=(2, 0, 1)):
if len(img.shape) == 3:
padded_img = np.ones((input_size[0], input_size[1], 3), dtype=np.uint8) * 114
else:
View panorama-application-package-fix.json
"interfaces": [
{
"name": "interface",
"category": "business_logic",
"asset": "code_asset",
"inputs": [
{
"name": "video_in",
"type": "media"
},