Hello guys,
I am currently using the hailo-infra repository and I am trying to connect my Hailo detection app to a real-time IP camera using RTSP.
Currently, I am using a placeholder video source like this:
self.video_source = “rtsp://username:password@camera_ip:554/profile1
I want to know how to properly get frames from an IP camera in realtime for inference.
Has anyone done this before or can suggest the right way to set the video source / GStreamer pipeline?
Here is a simplified version of my code:
import setproctitle
from hailo_apps.hailo_app_python.core.common.installation_utils import detect_hailo_arch
from hailo_apps.hailo_app_python.core.common.core import get_default_parser, get_resource_path
from hailo_apps.hailo_app_python.core.common.defines import (
DETECTION_APP_TITLE,
DETECTION_PIPELINE,
RESOURCES_MODELS_DIR_NAME,
RESOURCES_SO_DIR_NAME,
DETECTION_POSTPROCESS_SO_FILENAME,
DETECTION_POSTPROCESS_FUNCTION,
)
from hailo_apps.hailo_app_python.core.gstreamer.gstreamer_helper_pipelines import (
SOURCE_PIPELINE,
INFERENCE_PIPELINE,
INFERENCE_PIPELINE_WRAPPER,
TRACKER_PIPELINE,
USER_CALLBACK_PIPELINE,
DISPLAY_PIPELINE,
)
from hailo_apps.hailo_app_python.core.gstreamer.gstreamer_app import (
GStreamerApp,
app_callback_class,
dummy_callback,
)
class GStreamerDetectionApp(GStreamerApp):
def init(self, app_callback, user_data, parser=None):
if parser is None:
parser = get_default_parser()
parser.add_argument(
“–labels-json”,
default=None,
help=“Path to custom labels JSON file”,
)
super().__init__(parser, user_data)
# video
# self.video_source = "/path/to/test_video.mp4"
# camera
# self.video_source = "rtsp://username:password@camera_ip:554/profile1"
self.hef_path = "/path/to/model.hef"
self.labels_json = "/path/to/labels.json"
self.batch_size = 2
self.video_width = 1280
self.video_height = 720
nms_score_threshold = 0.3
nms_iou_threshold = 0.45
if self.options_menu.arch is None:
detected_arch = detect_hailo_arch()
if detected_arch is None:
raise ValueError(
"Could not auto-detect Hailo architecture. Please specify --arch manually."
)
self.arch = detected_arch
print(f"Auto-detected Hailo architecture: {self.arch}")
else:
self.arch = self.options_menu.arch
self.post_process_so = get_resource_path(
DETECTION_PIPELINE,
RESOURCES_SO_DIR_NAME,
DETECTION_POSTPROCESS_SO_FILENAME,
)
self.post_function_name = DETECTION_POSTPROCESS_FUNCTION
self.app_callback = app_callback
self.thresholds_str = (
f"nms-score-threshold={nms_score_threshold} "
f"nms-iou-threshold={nms_iou_threshold} "
f"output-format-type=HAILO_FORMAT_TYPE_FLOAT32"
)
setproctitle.setproctitle(DETECTION_APP_TITLE)
self.create_pipeline()
def get_pipeline_string(self):
source_pipeline = SOURCE_PIPELINE(
video_source=self.video_source,
video_width=self.video_width,
video_height=self.video_height,
frame_rate=self.frame_rate,
sync=self.sync,
)
detection_pipeline = INFERENCE_PIPELINE(
hef_path=self.hef_path,
post_process_so=self.post_process_so,
post_function_name=self.post_function_name,
batch_size=self.batch_size,
config_json=self.labels_json,
additional_params=self.thresholds_str,
)
detection_pipeline_wrapper = INFERENCE_PIPELINE_WRAPPER(detection_pipeline)
tracker_pipeline = TRACKER_PIPELINE(class_id=1)
user_callback_pipeline = USER_CALLBACK_PIPELINE()
display_pipeline = DISPLAY_PIPELINE(
video_sink=self.video_sink, sync=self.sync, show_fps=self.show_fps
)
pipeline_string = (
f"{source_pipeline}! "
f"{detection_pipeline_wrapper}! "
f"{tracker_pipeline}! "
f"{user_callback_pipeline}! "
f"{display_pipeline} "
)
print(pipeline_string)
return pipeline_string
def main():
user_data = dummy_callback() # ใช้ callback ตัวอย่าง
app = GStreamerDetectionApp(dummy_callback, user_data)
app.run()
if name == “main”:
print(“Starting Hailo Detection App…”)
main()