just sample of masking people using Yolact on Kinesis Video Streams

mask_kvs_stream.py

def mask_kvs_stream(net:Yolact, in_arn:str, out_arn:str, cuda:bool):
    out_stream_name = out_arn.split(":")[-1].split("/")[1]

    reader = KVSReader(in_arn)

    while True:
        streaming_url = reader.get_hls_streaming_session_url()

        if not streaming_url:
            time.sleep(1)
            continue

        mask_kvs_hls(net, streaming_url, out_stream_name, cuda)
        

def mask_kvs_hls(net:Yolact, in_stream_url:str, kvs_out_stream_name:str, cuda:bool):
    print(f"{datetime.datetime.now()}: HLS URL: {in_stream_url}, kvs_out_stream_name: {kvs_out_stream_name}, cuda: {cuda}", flush=True)
    target_fps = args.target_fps
    
    print(f"{datetime.datetime.now()}: FastBaseTransform()", flush=True)
    transform = FastBaseTransform()

    cap = cv2.VideoCapture(in_stream_url)

    if not cap.isOpened():
        print("VideoCapture could not be opened")
        sys.exit(-1)

    print(f"{datetime.datetime.now()}: VideoCapture is opened", flush=True)

    input_fps   = round(cap.get(cv2.CAP_PROP_FPS))
    frame_width  = round(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    frame_height = round(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    print(f"input_fps: {input_fps}")
    print(f"input_size: {frame_width}x{frame_height}", flush=True)

    access_key = os.environ["AWS_ACCESS_KEY_ID"]
    secret_key = os.environ["AWS_SECRET_ACCESS_KEY"]
    GSTREAMER_OUT = ' ! '.join([
        'appsrc',
        'clockoverlay halignment=right valignment=top font-desc="Sans bold 60px"',
        'videoconvert',
        'video/x-raw,format=YV12',
        'x264enc byte-stream=true noise-reduction=10000 speed-preset=ultrafast tune=zerolatency ',
        'video/x-h264,stream-format=avc,alignment=au,profile=baseline',
        ' '.join([
            'kvssink',
            f'stream-name={kvs_out_stream_name}',
            'storage-size=512',
            f'access-key={access_key}',
            f'secret-key={secret_key}',
            'aws-region=ap-northeast-1',
            # 'buffer-duration=10',
            # 'connection-staleness=60',
            # "fragment-acks=true",
            "framerate=1",
            # "key-frame-fragmentation=false",
            # "max-latency=10"
        ]),
    ])

    print(f"{datetime.datetime.now()}: GSTREAMER_OUT:", GSTREAMER_OUT, flush=True)
    out = cv2.VideoWriter(GSTREAMER_OUT, cv2.CAP_GSTREAMER, 0, target_fps, (frame_width, frame_height), True)
    
    while True:
        print(f"{datetime.datetime.now()}: read image", flush=True)

        ret, original_frame = cap.read()
        if not ret:
            break

        if target_fps == 0 or frame_width == 0 or frame_height == 0:
            print("Invalid FPS, width or height")
            sys.exit(-1)
        print(f"{datetime.datetime.now()}: predict image start", flush=True)

        frame = torch.from_numpy(original_frame).float()
        if cuda:
            frame = frame.cuda()
        batch = transform(frame.unsqueeze(0))
        preds = net(batch)
        print(f"{datetime.datetime.now()}: predict image end", flush=True)

        print(f"{datetime.datetime.now()}: postprocess start", flush=True)
        processed = prep_display(preds, frame, None, None, undo_transform=False, class_color=True)
        print(f"{datetime.datetime.now()}: postprocess end", flush=True)

        print(f"{datetime.datetime.now()}: write image start", flush=True)

        if args.local_imshow:
            cv2.imshow('processed', processed)
            cv2.waitKey(1)
        out.write(processed)
        print(f"{datetime.datetime.now()}: write image end", flush=True)

    out.release()
    cap.release()

    if args.local_imshow:
        cv2.destroyAllWindows()