I’m trying to display save and stream rgb/ir data to an external server via network, but i want to also be able to display and save in case there is no network so that i don’t lose data.
iam using flir hadron camera with nvidia jetson nano , python with gstreamer to process the data .
what i’m trying to do in my code is to execute the gstreamer that includes streaming if internet is detcted if not switch to the pipeline that only saves and displays and it continues to check for internet to add the stream .
The problem is that when i execute the code with internet connected it start with the pipeline that streams but once it’s disconnected it doesn’t stop the pipelines properly it prints ‘”Internet connection lost. Switching to saving/displaying pipeline.’ and it doesn’t execute the line after which is stop_recording() because it is stuck in the pipeline trying to stream . it only execute the stop_recording() when internet is back again which means that it shuts the streaming pipeline starts the displaying/saving pipeline and shuts them almost imidiatly because internet is back and starts the streaming pipeline again .
do you guys have any idea of how to solve this problem ,i’d appreciat any advise Thank you
Here is my code :
import RPi.GPIO as GPIO
import time
import datetime
import gi
import urllib.request
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
# Initialize GStreamer
Gst.init(None)
# Pin Definitions
input_pin = 24 # Pin 24 as input
# Variables to manage the video capture
is_recording = False
streaming_active = False
rgb_stream_pipeline = None
ir_stream_pipeline = None
rgb_save_pipeline = None
ir_save_pipeline = None
loop = GLib.MainLoop()
internet_check_interval = 3000 # Interval in milliseconds to check internet connectivity
def check_internet(host='http://google.com'):
try:
urllib.request.urlopen(host, timeout=3)
print("connected")
return True
except urllib.request.URLError:
print("not connected")
return False
def create_stream_pipeline(device, width, height, stream_url, file_prefix, timestamp, bitrate):
pipeline_description = (
f"v4l2src io-mode=4 device={device} do-timestamp=true ! "
f"video/x-raw, width={width}, height={height}, framerate=30/1 ! clockoverlay halignment=left valignment=bottom time-format= "%Y-%m-%d %H:%M:%S" font-desc='Sans, 36'! timeoverlay halignment=right valignment=bottom text= "Stream time:" font-desc='Sans, 24' ! "
"tee name=t ! "
f"queue ! nvvidconv ! nvv4l2h264enc bitrate={bitrate} ! h264parse ! "
"tee name=l "
f"! queue ! flvmux ! rtmpsink location='{stream_url} live=1' "
"l. ! "
f"queue ! qtmux ! filesink location=/home/nvidia/Desktop/vid/{file_prefix}_{timestamp}.mp4 "
"t. ! "
"queue leaky=1 ! xvimagesink sync=false"
)
return Gst.parse_launch(pipeline_description)
def create_save_pipeline(device, width, height, file_prefix, timestamp, bitrate):
pipeline_description = (
f"v4l2src io-mode=4 device={device} do-timestamp=true ! "
f"video/x-raw, width={width}, height={height}, framerate=30/1 ! clockoverlay halignment=left valignment=bottom time-format= "%Y-%m-%d %H:%M:%S" font-desc='Sans, 36'! timeoverlay halignment=right valignment=bottom text= "Stream time:" font-desc='Sans, 24' ! "
"tee name=t ! "
f"queue ! nvvidconv ! nvv4l2h264enc bitrate={bitrate} ! h264parse ! "
"tee name=l "
"l. ! "
f"queue ! qtmux ! filesink location=/home/nvidia/Desktop/vid/{file_prefix}_{timestamp}.mp4 "
"t. ! "
"queue leaky=1 ! xvimagesink sync=false"
)
return Gst.parse_launch(pipeline_description)
def on_message_rgb(bus, message, loop):
global streaming_active, rgb_stream_pipeline, rgb_save_pipeline
msg_type = message.type
if msg_type == Gst.MessageType.ERROR:
err, debug_info = message.parse_error()
print(f"Error received from element {message.src.get_name()}: {err.message}")
print(f"Debugging information: {debug_info if debug_info else 'none'}")
if streaming_active:
rgb_stream_pipeline.set_state(Gst.State.NULL)
else:
rgb_save_pipeline.set_state(Gst.State.NULL)
elif msg_type == Gst.MessageType.EOS:
print("End-Of-Stream reached")
if streaming_active:
rgb_stream_pipeline.set_state(Gst.State.NULL)
else:
rgb_save_pipeline.set_state(Gst.State.NULL)
def on_message_ir(bus, message, loop):
global streaming_active, ir_stream_pipeline, ir_save_pipeline
msg_type = message.type
if msg_type == Gst.MessageType.ERROR:
err, debug_info = message.parse_error()
print(f"Error received from element {message.src.get_name()}: {err.message}")
print(f"Debugging information: {debug_info if debug_info else 'none'}")
if streaming_active:
ir_stream_pipeline.set_state(Gst.State.NULL)
else:
ir_save_pipeline.set_state(Gst.State.NULL)
elif msg_type == Gst.MessageType.EOS:
print("End-Of-Stream reached")
if streaming_active:
ir_stream_pipeline.set_state(Gst.State.NULL)
else:
ir_save_pipeline.set_state(Gst.State.NULL)
def start_recording(streaming=True):
global is_recording, streaming_active, rgb_stream_pipeline, ir_stream_pipeline, rgb_save_pipeline, ir_save_pipeline, loop
current_timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
if streaming:
rgb_stream_pipeline = create_stream_pipeline('/dev/video0', 1920, 1080, 'google.com/rgb', 'RGB',
current_timestamp, 4000000)
ir_stream_pipeline = create_stream_pipeline('/dev/video1', 640, 512, 'google.com/ir', 'IR',
current_timestamp, 500000)
bus_rgb = rgb_stream_pipeline.get_bus()
bus_ir = ir_stream_pipeline.get_bus()
bus_rgb.add_signal_watch()
bus_ir.add_signal_watch()
bus_rgb.connect("message", on_message_rgb, loop)
bus_ir.connect("message", on_message_ir, loop)
rgb_stream_pipeline.set_state(Gst.State.PLAYING)
ir_stream_pipeline.set_state(Gst.State.PLAYING)
print("Starting RGB and IR streaming pipelines...")
else:
rgb_save_pipeline = create_save_pipeline('/dev/video0', 1920, 1080, 'RGB',
current_timestamp, 4000000)
ir_save_pipeline = create_save_pipeline('/dev/video1', 640, 512, 'IR',
current_timestamp, 500000)
bus_rgb = rgb_save_pipeline.get_bus()
bus_ir = ir_save_pipeline.get_bus()
bus_rgb.add_signal_watch()
bus_ir.add_signal_watch()
bus_rgb.connect("message", on_message_rgb, loop)
bus_ir.connect("message", on_message_ir, loop)
rgb_save_pipeline.set_state(Gst.State.PLAYING)
ir_save_pipeline.set_state(Gst.State.PLAYING)
print("Starting RGB and IR saving/displaying pipelines...")
streaming_active = streaming
is_recording = True
def stop_recording():
global is_recording, streaming_active, rgb_stream_pipeline, ir_stream_pipeline, rgb_save_pipeline, ir_save_pipeline
if streaming_active:
if rgb_stream_pipeline:
rgb_stream_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
rgb_stream_pipeline.set_state(Gst.State.NULL)
print("RGB streaming stopped")
if ir_stream_pipeline:
ir_stream_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
ir_stream_pipeline.set_state(Gst.State.NULL)
print("IR streaming stopped")
else:
if rgb_save_pipeline:
rgb_save_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
rgb_save_pipeline.set_state(Gst.State.NULL)
print("RGB saving/displaying stopped")
if ir_save_pipeline:
ir_save_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
ir_save_pipeline.set_state(Gst.State.NULL)
print("IR saving/displaying stopped")
is_recording = False
def pin_callback(channel):
global is_recording
if GPIO.input(channel):
print("Rising edge detected on pin", channel)
if not is_recording:
if check_internet():
start_recording(streaming=True)
else:
start_recording(streaming=False)
is_recording = True
else:
print("Falling edge detected on pin", channel)
if is_recording:
stop_recording()
is_recording = False
def internet_check():
global is_recording, streaming_active
if is_recording:
if check_internet():
if not streaming_active:
print("Internet connection detected. Switching to streaming pipeline.")
stop_recording()
start_recording(streaming=True)
else:
if streaming_active:
print("Internet connection lost. Switching to saving/displaying pipeline.")
stop_recording()
start_recording(streaming=False)
return True # Returning True keeps the timeout active
def main():
GPIO.setmode(GPIO.BCM) # Use BCM pin numbering
GPIO.setup(input_pin, GPIO.IN) # Set pin 24 as input
GPIO.add_event_detect(input_pin, GPIO.BOTH, callback=pin_callback, bouncetime=300)
print("Starting demo now! Press CTRL+C to exit")
GLib.timeout_add(internet_check_interval, internet_check)
try:
loop.run()
except KeyboardInterrupt:
print("Stopping pipelines...")
loop.quit()
print("Pipelines stopped.")
if __name__ == '__main__':
main()
i also tried to isolate the streaming pipeline using this gstreamer command (proxysink/proxysrc) that i found on github so that it doesn’t affect the saving/displaying , only adding the stream when internet is detected but it didn’t work it also has the same behaviar of freezing the displaying and streaming intill it is reconnected
here is the code of the proxysink/proxysrc:
import RPi.GPIO as GPIO
import time
import datetime
import gi
import urllib.request
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
# Initialize GStreamer
Gst.init(None)
# Pin Definitions
input_pin = 24 # Pin 24 as input
# Variables to manage the video capture
is_recording = False
rgb_pipeline = None
ir_pipeline = None
loop = GLib.MainLoop()
internet_check_interval = 3000 # Interval in milliseconds to check internet connectivity
internet_check_timeout_id = None # Store the timeout ID
def check_internet(host='http://google.com'):
try:
urllib.request.urlopen(host, timeout=3)
print("connected")
return True
except urllib.request.URLError:
print("not connected")
#stop_recording()
return False
def create_pipeline(device, width, height, file_prefix, timestamp, bitrate):
pipeline_description = (
f"v4l2src io-mode=4 device={device} do-timestamp=true ! "
f"video/x-raw, width={width}, height={height}, framerate=30/1 ! clockoverlay halignment=left valignment=bottom time-format= "%Y-%m-%d %H:%M:%S" font-desc='Sans, 36'! timeoverlay halignment=right valignment=bottom text= "Stream time:" font-desc='Sans, 24' ! "
"tee name=t ! "
f"queue ! nvvidconv ! nvv4l2h264enc bitrate={bitrate} control-rate=0 preset-level=1 maxperf-enable=true ! h264parse ! "
"tee name=l ! "
"queue ! proxysink name=psink "
"l. ! "
f"queue ! qtmux ! filesink location=/home/nvidia/Desktop/vid/{file_prefix}_{timestamp}.mp4 "
"t. ! "
"queue leaky=1 ! xvimagesink sync=false"
)
return Gst.parse_launch(pipeline_description)
def create_streaming_pipeline(stream_url):
pipeline_description = (
"proxysrc name=psrc ! "
f"queue ! flvmux ! rtmpsink location='{stream_url} live=1'"
)
return Gst.parse_launch(pipeline_description)
def on_message_rgb(bus, message, loop):
msg_type = message.type
if msg_type == Gst.MessageType.ERROR:
err, debug_info = message.parse_error()
print(f"Error received from element {message.src.get_name()}: {err.message}")
print(f"Debugging information: {debug_info if debug_info else 'none'}")
rgb_pipeline.set_state(Gst.State.NULL)
rgb_streaming_pipeline.set_state(Gst.State.NULL)
elif msg_type == Gst.MessageType.EOS:
print("End-Of-Stream reached")
rgb_pipeline.set_state(Gst.State.NULL)
rgb_streaming_pipeline.set_state(Gst.State.NULL)
def on_message_ir(bus, message, loop):
msg_type = message.type
if msg_type == Gst.MessageType.ERROR:
err, debug_info = message.parse_error()
print(f"Error received from element {message.src.get_name()}: {err.message}")
print(f"Debugging information: {debug_info if debug_info else 'none'}")
ir_pipeline.set_state(Gst.State.NULL)
ir_streaming_pipeline.set_state(Gst.State.NULL)
elif msg_type == Gst.MessageType.EOS:
print("End-Of-Stream reached")
ir_pipeline.set_state(Gst.State.NULL)
ir_streaming_pipeline.set_state(Gst.State.NULL)
def start_recording():
global is_recording, rgb_pipeline, ir_pipeline, rgb_streaming_pipeline, ir_streaming_pipeline, loop
current_timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
rgb_pipeline = create_pipeline('/dev/video0', 1920, 1080, 'RGB', current_timestamp, 4000000)
ir_pipeline = create_pipeline('/dev/video1', 640, 512, 'IR', current_timestamp, 500000)
rgb_streaming_pipeline = create_streaming_pipeline('google.com/rgb')
ir_streaming_pipeline = create_streaming_pipeline('google.com/ir')
psink_rgb = rgb_pipeline.get_by_name('psink')
psink_ir = ir_pipeline.get_by_name('psink')
psrc_rgb = rgb_streaming_pipeline.get_by_name('psrc')
psrc_ir = ir_streaming_pipeline.get_by_name('psrc')
psrc_rgb.set_property('proxysink', psink_rgb)
psrc_ir.set_property('proxysink', psink_ir)
bus_rgb = rgb_pipeline.get_bus()
bus_ir = ir_pipeline.get_bus()
bus_rgb_streaming = rgb_streaming_pipeline.get_bus()
bus_ir_streaming = ir_streaming_pipeline.get_bus()
bus_rgb.add_signal_watch()
bus_ir.add_signal_watch()
bus_rgb_streaming.add_signal_watch()
bus_ir_streaming.add_signal_watch()
bus_rgb.connect("message", on_message_rgb, loop)
bus_ir.connect("message", on_message_ir, loop)
bus_rgb_streaming.connect("message", on_message_rgb, loop)
bus_ir_streaming.connect("message", on_message_ir, loop)
rgb_pipeline.set_state(Gst.State.PLAYING)
ir_pipeline.set_state(Gst.State.PLAYING)
rgb_streaming_pipeline.set_state(Gst.State.PLAYING)
ir_streaming_pipeline.set_state(Gst.State.PLAYING)
print("Starting RGB and IR pipelines...")
def stop_recording():
global is_recording, rgb_pipeline, ir_pipeline, rgb_streaming_pipeline, ir_streaming_pipeline
if rgb_pipeline:
rgb_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
rgb_pipeline.set_state(Gst.State.NULL)
print("RGB stopped")
if ir_pipeline:
ir_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
ir_pipeline.set_state(Gst.State.NULL)
print("IR stopped")
if rgb_streaming_pipeline:
rgb_streaming_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
rgb_streaming_pipeline.set_state(Gst.State.NULL)
print("RGB streaming pipeline stopped")
if ir_streaming_pipeline:
ir_streaming_pipeline.send_event(Gst.Event.new_eos())
time.sleep(1)
ir_streaming_pipeline.set_state(Gst.State.NULL)
print("IR streaming pipeline stopped")
is_recording = False
def pin_callback(channel):
global is_recording
if GPIO.input(channel):
print("Rising edge detected on pin", channel)
if not is_recording and check_internet():
start_recording()
is_recording = True
else:
print("Falling edge detected on pin", channel)
if is_recording:
stop_recording()
is_recording = False
def internet_check():
global is_recording
if is_recording and not check_internet():
print("Internet connection lost. Stopping recording.")
#stop_recording()
return False # No need to continue checking if internet is lost
return True # Returning True keeps the timeout active
def main():
GPIO.setmode(GPIO.BCM) # Use BCM pin numbering
GPIO.setup(input_pin, GPIO.IN) # Set pin 24 as input
GPIO.add_event_detect(input_pin, GPIO.BOTH, callback=pin_callback, bouncetime=300)
print("Starting demo now! Press CTRL+C to exit")
GLib.timeout_add(internet_check_interval, internet_check)
try:
loop.run()
except KeyboardInterrupt:
print("Stopping pipelines...")
loop.quit()
print("Pipelines stopped.")
if __name__ == '__main__':
main()