1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
#!/usr/bin/env python3
import include_homekit
import sys
import os
import subprocess
import asyncio
import signal
from typing import TextIO
from argparse import ArgumentParser
from socket import gethostname
from asyncio.streams import StreamReader
from homekit.config import config as homekit_config
from homekit.linux import LinuxBoardsConfig
from homekit.camera import IpcamConfig, CaptureType
from homekit.camera.util import get_hls_directory, get_hls_channel_name, get_recordings_path
ipcam_config = IpcamConfig()
lbc_config = LinuxBoardsConfig()
channels = (1, 2)
tasks = []
restart_delay = 3
lock = asyncio.Lock()
worker_type: CaptureType
async def read_output(stream: StreamReader,
thread_name: str,
output: TextIO):
try:
while True:
line = await stream.readline()
if not line:
break
print(f"[{thread_name}] {line.decode().strip()}", file=output)
except asyncio.LimitOverrunError:
print(f"[{thread_name}] Output limit exceeded.", file=output)
except Exception as e:
print(f"[{thread_name}] Error occurred while reading output: {e}", file=sys.stderr)
async def run_ffmpeg(cam: int, channel: int):
prefix = get_hls_channel_name(cam, channel)
if homekit_config.app_config.logging_is_verbose():
debug_args = ['-v', '-info']
else:
debug_args = ['-nostats', '-loglevel', 'error']
# protocol = 'tcp' if ipcam_config.should_use_tcp_for_rtsp(cam) else 'udp'
protocol = 'tcp'
user, pw = ipcam_config.get_rtsp_creds()
ip = ipcam_config.get_camera_ip(cam)
path = ipcam_config.get_camera_type(cam).get_channel_url(channel)
ext = ipcam_config.get_camera_container(cam)
ffmpeg_command = ['ffmpeg', *debug_args,
'-rtsp_transport', protocol,
'-i', f'rtsp://{user}:{pw}@{ip}:554{path}',
'-c', 'copy',]
if worker_type == CaptureType.HLS:
ffmpeg_command.extend(['-bufsize', '1835k',
'-pix_fmt', 'yuv420p',
'-flags', '-global_header',
'-hls_time', '2',
'-hls_list_size', '3',
'-hls_flags', 'delete_segments',
os.path.join(get_hls_directory(cam, channel), 'live.m3u8')])
elif worker_type == CaptureType.RECORD:
ffmpeg_command.extend(['-f', 'segment',
'-strftime', '1',
'-segment_time', '00:10:00',
'-segment_atclocktime', '1',
os.path.join(get_recordings_path(cam), f'record_%Y-%m-%d-%H.%M.%S.{ext.value}')])
else:
raise ValueError(f'invalid worker type: {worker_type}')
while True:
try:
process = await asyncio.create_subprocess_exec(
*ffmpeg_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
stdout_task = asyncio.create_task(read_output(process.stdout, prefix, sys.stdout))
stderr_task = asyncio.create_task(read_output(process.stderr, prefix, sys.stderr))
await asyncio.gather(stdout_task, stderr_task)
# check the return code of the process
if process.returncode != 0:
raise subprocess.CalledProcessError(process.returncode, ffmpeg_command)
except (FileNotFoundError, PermissionError, subprocess.CalledProcessError) as e:
# an error occurred, print the error message
error_message = f"Error occurred in {prefix}: {e}"
print(error_message, file=sys.stderr)
# sleep for 5 seconds before restarting the process
await asyncio.sleep(restart_delay)
async def run():
kwargs = {}
if worker_type == CaptureType.RECORD:
kwargs['filter_by_server'] = gethostname()
for cam in ipcam_config.get_all_cam_names(**kwargs):
for channel in channels:
task = asyncio.create_task(run_ffmpeg(cam, channel))
tasks.append(task)
try:
await asyncio.gather(*tasks)
except KeyboardInterrupt:
print('KeyboardInterrupt: stopping processes...', file=sys.stderr)
for task in tasks:
task.cancel()
# wait for subprocesses to terminate
await asyncio.gather(*tasks, return_exceptions=True)
# send termination signal to all subprocesses
for task in tasks:
process = task.get_stack()
if process:
process.send_signal(signal.SIGTERM)
if __name__ == '__main__':
capture_types = [t.value for t in CaptureType]
parser = ArgumentParser()
parser.add_argument('type', type=str, metavar='CAPTURE_TYPE', choices=tuple(capture_types),
help='capture type (variants: '+', '.join(capture_types)+')')
arg = homekit_config.load_app(no_config=True, parser=parser)
worker_type = CaptureType(arg['type'])
asyncio.run(run())
|
