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:

mrtj

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:

mrtj

"interfaces": [
    {
        "name": "interface",
        "category": "business_logic",
        "asset": "code_asset",
        "inputs": [
            {
                "name": "video_in",
                "type": "media"
            },

mrtj

{
    "name": "code_node",
    "interface": "${AWS_ACCOUNT_ID}::application_logic.interface",
    "overridable": false,
    "launch": "onAppStart"
}

mrtj

panorama-cli build-container \
    --container-asset-name code \
    --package-path packages/${AWS_ACCOUNT_ID}-application_logic-1.0

mrtj

{
    "runtimeDescriptor": {
        "envelopeVersion": "2021-01-01",
        "entry": {
            "path": "python3",
            "name": "/panorama/application.py"
        }
    }
}

mrtj

def main():
    logger = get_logger(level=logging.DEBUG)
    try:
        logger.info('INITIALIZING APPLICATION')
        app = Application(logger)
        logger.info('PROCESSING STREAMS')
        # video processing loop:
        while True:
            app.process_streams()
    except:

mrtj

import panoramasdk
import numpy as np

class Application(panoramasdk.node):
    def __init__(self, logger):
        super().__init__()
        self.logger = logger
        self.threshold = 0.
        self.MODEL_NODE = 'model_node'
        self.MODEL_INPUT_NAME = 'images'

mrtj

import logging
from logging.handlers import RotatingFileHandler

def get_logger(name=__name__, level=logging.INFO):
    logger = logging.getLogger(name)
    logger.setLevel(level)
    handler = RotatingFileHandler(
        "/opt/aws/panorama/logs/app.log", 
        maxBytes=10000000, 
        backupCount=2