[Question]: Using streamgear with webgear

[MubashirWaheed]

Issue guidelines

• I've read the Issue Guidelines and wholeheartedly agree.

Issue Checklist

• I have searched open or closed issues for my problem and found nothing related or helpful.
• I have read the Documentation and found nothing related to my problem.
• I have gone through the Bonus Examples and FAQs and found nothing related or helpful.

Describe your Question

I checked docs and didn't find any example for my use case. I am getting a series of frames after processing and want to show them on the web. So basically a combination of streamgear and webgear.
I am using the inference pipeline from roboflow so I can't yield frame like in opencv. I can'rt just pass the on_prediction to the webgear config as shown in the example

You can read here:https://inference.roboflow.com/using_inference/inference_pipeline/#usage

Terminal log output(Optional)

No response

Python Code(Optional)

sample code 
// initialize like this?
stream = CamGear()
streamer = StreamGear(output=".\dash_out.mpd")

class CustomSink:
    def __init__(self, weights_path: str, zone_configuration_path: str, classes: List[int]):
    // initialization 

    def on_prediction(self, result: dict, frame: VideoFrame) -> None:
        self.fps_monitor.tick()
        fps = self.fps_monitor.fps
        detections = sv.Detections.from_ultralytics(result)
        detections = detections[find_in_list(detections.class_id, self.classes)]
        detections = self.tracker.update_with_detections(detections)

        annotated_frame = frame.image.copy()

        annotated_frame = sv.draw_text(
            scene=annotated_frame,
            text=f"{fps:.1f}",
            text_anchor=sv.Point(40, 30),
            background_color=sv.Color.from_hex("#A351FB"),
            text_color=sv.Color.from_hex("#000000"),
        )
            labels = [
                f"#{tracker_id} {int(time // 60):02d}:{int(time % 60):02d}" if class_id != 2 else f"#{tracker_id}"
                for tracker_id, time, class_id in zip(detections_in_zone.tracker_id, time_in_zone,detections_in_zone.class_id)
            ]

            annotated_frame = LABEL_ANNOTATOR.annotate(
                scene=annotated_frame,
                detections=detections_in_zone,
                labels=labels,
                custom_color_lookup=custom_color_lookup,
            )

        # replace with streamgear and webgear
        streamer.stream(annotated_frame)
        try:
        # Resize the frame to fit the window size
            resized_frame = cv2.resize(annotated_frame, (width, height))
        except cv2.error as e:
            print(f"Error resizing frame: {e}")
            return  # Skip this frame and continue with the next one

        # Display the resized frame in the window
        cv2.imshow('Resizable Window', resized_frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            cv2.destroyAllWindows()
            raise SystemExit("Program terminated by user")

VidGear Version

0.3.3

Python version

3.11.8

Operating System version

ubuntu

Any other Relevant Information?

No response

[welcome]

Thanks for opening this issue, a maintainer will get back to you shortly!

In the meantime:

• Read our Issue Guidelines, and update your issue accordingly. Please note that your issue will be fixed much faster if you spend about half an hour preparing it, including the exact reproduction steps and a demo.
• Go comprehensively through our dedicated FAQ & Troubleshooting section.
• For any quick questions and typos, please refrain from opening an issue, as you can reach us on Gitter community channel.

[abhiTronix]

@MubashirWaheed Where are you using WebGear and StreamGear APIs in your code? Can you provide the context for each of these APIs? I cannot understand how you're using them, and what is the flow (input - processing - output) of your Python script. If you can explain, that would be great.

[MubashirWaheed]

@abhiTronix I want to integrate the vidgear library and specifically the streamgear and webgear apis in my code. The above shared code is the CustomSink class. There are two main methods on the class infer and on_prediction. Inside the infer based on the detection model objects are detected. In the on_prediction callback I am processing the frame eg annotating the frames.
basically I want to display the stream of frames from the on_prediction callback to the web page and to achieve this I want to integrate the webgear and streamgear APIs

Is there an example available for this?

[MubashirWaheed]

I tried doing something like this

def main(
    weight_path: str,
    rtsp_url: str,
    zone_configuration_path: str
    // other params
) -> None:
    sink = CustomSink(weights_path=weight_path, zone_configuration_path=zone_configuration_path, classes=classes)
    // maybe I can use webgear like this?
    // I tried but it does not work
    web = WebGear(source=sink.on_prediction, logging=True)
    pipeline = InferencePipeline.init_with_custom_logic(
        video_reference=rtsp_url,
        on_video_frame=sink.infer,
        on_prediction=sink.on_prediction
    )

here are the docs: https://inference.roboflow.com/using_inference/inference_pipeline/