Цель состоит в том, чтобы отслеживать, есть ли активный поток на камерах, получите URL-адрес потока RTSP и перенесите его в список воспроизведения HLS для локального использования.
Вот код Python:
Код: Выделить всё
import asyncio
from decouple import config
import logging
from my_pyaarlo import PyArlo
import urllib.parse
from queue import Queue
import signal
# Read config from ENV (unchanged)
ARLO_USER = config('ARLO_USER')
ARLO_PASS = config('ARLO_PASS')
IMAP_HOST = config('IMAP_HOST')
IMAP_USER = config('IMAP_USER')
IMAP_PASS = config('IMAP_PASS')
DEBUG = config('DEBUG', default=False, cast=bool)
PYAARLO_BACKEND = config('PYAARLO_BACKEND', default=None)
PYAARLO_REFRESH_DEVICES = config('PYAARLO_REFRESH_DEVICES', default=0, cast=int)
PYAARLO_STREAM_TIMEOUT = config('PYAARLO_STREAM_TIMEOUT', default=0, cast=int)
PYAARLO_STORAGE_DIR = config('PYAARLO_STORAGE_DIR', default=None)
PYAARLO_ECDH_CURVE = config('PYAARLO_ECDH_CURVE', default=None)
# Initialize logging
logging.basicConfig(
level=logging.DEBUG if DEBUG else logging.INFO,
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
ffmpeg_processes = {}
event_queue = Queue()
shutdown_event = asyncio.Event()
async def handle_idle_event(camera):
logger.info(f"Idle event detected for camera: {camera.name}")
await stop_ffmpeg_stream(camera.name)
async def get_stream_url(camera):
try:
# Attempt to get the stream URL
stream_url = await asyncio.to_thread(camera.get_stream()
if stream_url:
return stream_url
else:
logger.warning(f"Unable to get stream URL for {camera.name}. Stream might not be active.")
return None
except Exception as e:
logger.error(f"Error getting stream URL for {camera.name}: {e}")
return None
async def handle_user_stream_active_event(camera):
logger.info(f"User stream active event detected for camera: {camera.name}")
# Get the stream URL
stream_url = await get_stream_url(camera)
if stream_url:
logger.info(f"Stream URL for {camera.name}: {stream_url}")
await start_ffmpeg_stream(camera.name, stream_url)
else:
logger.warning(f"No stream URL available for {camera.name}")
async def event_handler(device, attr, value):
logger.debug(f"Event: {device.name}, Attribute: {attr}, Value: {value}")
if attr == 'activityState':
if value == 'idle':
await handle_idle_event(device)
elif value in ['userStreamActive']:
await handle_user_stream_active_event(device)
elif attr == 'mediaUploadNotification':
logger.info(f"Media uploaded for camera: {device.name}")
def sync_event_handler(device, attr, value):
# This function will be called by PyArlo's synchronous callbacks
event_queue.put((device, attr, value))
async def process_event_queue():
while not shutdown_event.is_set():
try:
if not event_queue.empty():
device, attr, value = event_queue.get()
await event_handler(device, attr, value)
await asyncio.sleep(0.1) # Small delay to prevent busy-waiting
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error processing event: {e}")
async def display_status(arlo):
while not shutdown_event.is_set():
print("\n--- Camera Statuses ---")
for camera in arlo.cameras:
print(f"{camera.name}: {camera.state}")
print("------------------------")
await asyncio.sleep(5)
async def start_ffmpeg_stream(camera_name, stream_url):
if camera_name not in ffmpeg_processes:
output_hls = f"/tmp/{camera_name}.m3u8"
try:
new_url = urllib.parse.quote(stream_url.encode(), safe=':/?&=')
logger.info(f"NEW_URL: {new_url}")
ffmpeg_cmd = [
"ffmpeg", "-hide_banner", "-loglevel", "quiet", "-nostats", "-nostdin", "-y", "-re",
"-i", new_url,
"-c:v", "libx264", "-preset", "veryfast",
"-an", "-sn",
"-f", "hls", "-hls_time", "4", "-hls_list_size", "10",
"-hls_flags", "delete_segments", output_hls,
]
logger.info(f"Starting FFmpeg command: {ffmpeg_cmd}")
process = await asyncio.create_subprocess_exec(
*ffmpeg_cmd,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL
)
ffmpeg_processes[camera_name] = process
logger.info(f"Started ffmpeg process with PID: {process.pid}")
except Exception as e:
logger.error(f"Error starting FFmpeg for {camera_name}: {e}")
async def stop_ffmpeg_stream(camera_name):
logger.info(f"Stopping ffmpeg process for {camera_name}")
ffmpeg_process = ffmpeg_processes.pop(camera_name, None)
if ffmpeg_process:
ffmpeg_process.terminate()
try:
await ffmpeg_process.wait()
logger.info(f"{camera_name} stopped successfully")
except Exception as e:
print(f"FFMPEG Process didn't stop in time, forcefully terminating: {e}")
ffmpeg_process.kill()
else:
logger.info(f"FFmpeg process for {camera_name} already stopped")
async def shutdown(signal, loop):
logger.info(f"Received exit signal {signal.name}...")
shutdown_event.set()
tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
[task.cancel() for task in tasks]
logger.info(f"Cancelling {len(tasks)} outstanding tasks")
await asyncio.gather(*tasks, return_exceptions=True)
loop.stop()
async def main():
# Initialize PyArlo
arlo_args = {
'username': ARLO_USER,
'password': ARLO_PASS,
'tfa_source': 'imap',
'tfa_type': 'email',
'tfa_host': IMAP_HOST,
'tfa_username': IMAP_USER,
'tfa_password': IMAP_PASS,
'save_session': True,
'verbose_debug': DEBUG
}
# Add optional arguments
for arg, value in [
('refresh_devices_every', PYAARLO_REFRESH_DEVICES),
('stream_timeout', PYAARLO_STREAM_TIMEOUT),
('backend', PYAARLO_BACKEND),
('storage_dir', PYAARLO_STORAGE_DIR),
('ecdh_curve', PYAARLO_ECDH_CURVE)
]:
if value:
arlo_args[arg] = value
try:
arlo = await asyncio.to_thread(PyArlo, **arlo_args)
except Exception as e:
logger.error(f"Failed to initialize PyArlo: {e}")
return
logger.info("Connected to Arlo. Monitoring events...")
# Register event handlers for each camera
for camera in arlo.cameras:
camera.add_attr_callback('*', sync_event_handler)
# Start the status display task
status_task = asyncio.create_task(display_status(arlo))
# Start the event processing task
event_processing_task = asyncio.create_task(process_event_queue())
# Set up signal handlers
loop = asyncio.get_running_loop()
for s in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(
s, lambda s=s: asyncio.create_task(shutdown(s, loop)))
try:
# Keep the main coroutine running
while not shutdown_event.is_set():
try:
await asyncio.sleep(1)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Unexpected error in main loop: {e}")
finally:
logger.info("Shutting down...")
for camera_name in list(ffmpeg_processes.keys()):
await stop_ffmpeg_stream(camera_name)
# Cancel and wait for all tasks
tasks = [status_task, event_processing_task]
for task in tasks:
if not task.done():
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
logger.info("Program terminated.")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Keyboard interrupt received. Exiting.")
except Exception as e:
logger.error(f"Unhandled exception: {e}")
finally:
logger.info("Program exit complete.")
Я пробовал много чего, без асинхронности, с многопроцессорностью, с подпроцессом... поведение всегда одинаковое. В некоторых случаях я получал событие простоя после прерывания по клавише.
Другая информация:
- Когда я останавливаю активный поток, событие не получено, но когда я запускаю поток сразу после этого, это событие получено.
- Когда я запускаю ту же команду ffmpeg, но с локальным длинным видеофайлом, все Хорошо. Вот почему я предполагаю, что команда ffmpeg влияет на основной процесс.
Код: Выделить всё
import asyncio
import signal
import sys
import os
async def run_infinite_command():
# Start a simple HTTP server as our "infinite" command
url = "rstp://localhost:8554/camera1/stream" # it is a fake url
ffmpeg_cmd = [
"ffmpeg", "-re", "-i", url,
"-c:v", "libx264", "-preset", "veryfast",
"-c:a", "copy",
"-f", "hls", "-hls_time", "4", "-hls_list_size", "10",
"-hls_flags", "delete_segments", "/tmp/output.m3u8"
]
process = await asyncio.create_subprocess_exec(
*ffmpeg_cmd,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL
)
print(f"Started HTTP server with PID: {process.pid}")
return process
async def main():
# Start the infinite command
process = await run_infinite_command()
# Run the main loop for a few seconds
for i in range(10):
print(f"Main loop iteration {i+1}")
await asyncio.sleep(1)
# Stop the infinite command
print("Stopping the HTTP server...")
if sys.platform == "win32":
# On Windows, we need to use CTRL_C_EVENT
os.kill(process.pid, signal.CTRL_C_EVENT)
else:
# On Unix-like systems, we can use SIGTERM
process.send_signal(signal.SIGTERM)
# Wait for the process to finish
try:
await asyncio.wait_for(process.wait(), timeout=5.0)
print("HTTP server stopped successfully")
except asyncio.TimeoutError:
print("HTTP server didn't stop in time, forcefully terminating")
process.kill()
print("Program finished")
if __name__ == "__main__":
asyncio.run(main())
Не могли бы вы помочь?
Подробнее здесь: https://stackoverflow.com/questions/790 ... eg-command
