Turn anything
into a live stream.

@stream
def pitch(stream_builder: StreamBuilder) -> None:
    camera = stream_builder.remote_output("pitch-cam/sideline/camera/video")
    detector = stream_builder.add(
        HuggingFaceDetector, config={"model": "facebook/detr-resnet-50"}
    )
    overlay = stream_builder.add(Overlay)
    stream_builder.connect(camera, detector.input("video_from_upstream"))
    stream_builder.connect(camera, overlay.input("video_from_upstream"))
    stream_builder.connect(
        detector.output("detections_to_downstream"),
        overlay.input("detections_from_upstream"),
    )
    stream_builder.expose(
        overlay.output("video_to_downstream"), Exposure.PUBLIC
    )

@node
class HuggingFaceDetector:
    @node.input(delivery_profile="newest")
    def video_from_upstream(self) -> VideoFrame: ...

    @node.output()
    def detections_to_downstream(self) -> dict: ...

    def __init__(self, config: HuggingFaceDetectorConfig) -> None:
        self.model = config.model

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.detect = pipeline("object-detection", model=self.model)

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        frame = ctx.inputs.read("video_from_upstream", into=VideoFrame)
        if frame is not None:
            with frame.cpu() as pixels:
                objects = self.detect(Image.fromarray(pixels[:, :, :3]))
            ctx.outputs.write("detections_to_downstream", {"objects": objects})