pavel.muhortov/utils
pavel.muhortov/utils
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

deleted camsutil, ffmpeger, procutil

Browse Source
This commit is contained in:
pavel.muhortov 2023-03-11 23:18:58 +03:00
parent 124076dc87
commit 2b79247a5b
23 changed files with 20 additions and 693 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
README.md
View File

@ -1,57 +1,16 @@
# utils
Small tools needed to solve immediate tasks independently or as part of a project
* [`camsutil`](https://git.hmp.today/pavel.muhortov/utils#camsutil)
* [`cronutil`](https://git.hmp.today/pavel.muhortov/utils#cronutil)
* [`confutil`.py](https://git.hmp.today/pavel.muhortov/utils#confutil-py)
* [`ffmpeger`.py](https://git.hmp.today/pavel.muhortov/utils#ffmpeger-py)
* [`procutil`.py](https://git.hmp.today/pavel.muhortov/utils#procutil-py)
* [`sendmail`.py](https://git.hmp.today/pavel.muhortov/utils#sendmail-py)
* [`simplewc`.py](https://git.hmp.today/pavel.muhortov/utils#simplewc-py)
____
## `camsutil`
**Description:** Creation of a request to the camera API based on the prepared template
**Dependencies:** Python 3 (tested version 3.9.5)
| PARAMETERS | DESCRIPTION | DEFAULT|
|-------------|-------------|--------|
|**--host**|hostname or ip address|**REQUIRED**|
|**--user**|valid user|**REQUIRED**|
|**--password**|valid password|**REQUIRED**|
|**--template**|the name of an existing template|**REQUIRED**|
|**[-h]**|print help and exit||
|**[--protocol]**|http, https, etc.|http|
|**[--port]**|port number|80|
|**[--channel]**|ptz channel number|101|
|**[--vid]**|video channel id|1|
|**[--x]**|horizontal positioning: azimuth, pan|0|
|**[--y]**|vertical positioning: elevation, tilt|0|
|**[--z]**|zoom direction, absolute zoom|0|
|**[--speed]**|positioning speed: from 1 to 7|1|
|**[--time]**|momentary duration, max 100000ms|100000|
|**[--text]**|overlay text content|`None`|
|**[--enabled]**|enabled (true) or disabled (false) overlay text|true|
Example usage in terminal with Python:
```shell
python3 ./ptz.py --host HOST --user USER --password PASS --template Hikvision_GetCapabilities.html
```
Example usage in terminal with make the script executable:
```shell
chmod u+x ./ptz.py
ptz.py --host HOST --user USER --password PASS --template Hikvision_PtzPreset.html --vid 2 --speed 5
```
Example usage in Python:
```Python
from camsutil import ptz
data = ptz.API(host='HOST', user='USER', password='PASS', template='Hikvision_GetJPEG.html', x=1920, y=1080).call()
with open('img.jpg', 'wb') as output:
output.write(data)
```
____
## `cronutil`
**Description:** Control wrapper for the [schedule](https://github.com/dbader/schedule) package
**Dependencies:** Python 3 (tested version 3.9.5)
@ -74,6 +33,7 @@ Examples:
`'1:07:00:00'` - every 00 seconds 00 minutes 07 hours Monday
Example usage in Python:
```Python
from time import strftime
from cronutil import Scheduler
@ -90,11 +50,14 @@ cron.add('2,4:**:59:59', cron.stop)
```
____
## `confutil`.py
**Description:** Parser of configs, arguments, parameters
**Dependencies:** Python 3 (tested version 3.9.5)
Example config to parse:
```text
[main]
# This block contains basic parameters
@ -112,6 +75,7 @@ directory=www;
```
Example usage in Python:
```Python
from os import path
from confutil import Parse
@ -122,74 +86,9 @@ if path.exists(conf):
```
____
## `ffmpeger`.py
**Description:** FFmpeg management from Python
**Dependencies:** Python 3 (tested version 3.9.5), installed or downloaded [ffmpeg](https://ffmpeg.org/download.html), [procutil.py](https://git.hmp.today/pavel.muhortov/utils#procutil-py) in the same directory
| PARAMETERS | DESCRIPTION | DEFAULT|
|-------------|-------------|--------|
|**-s**, **--src**|sources urls|**REQUIRED**|
|**[-h]**|print help and exit||
|**[--preset]**|240p, 360p, 480p, 720p, 1080p, 1440p, 2160p|`None`|
|**[--fps]**|frame per second encoding output|`None`|
|**[--dst]**|destination url|`None`|
|**[--ffpath]**|alternative path to bin|`None`|
|**[--watchdog]**|detect ffmpeg freeze and terminate||
|**[--sec]**|seconds to wait before the watchdog terminates|15|
|**[--mono]**|detect ffmpeg running copy and terminate||
Example usage in cron with Python:
```shell
# at every minute
* * * * * /usr/bin/python3 ~/ffmpeger.py -s rtsp://user:pass@host:554/Streaming/Channels/video,http://Streaming/Channels/audio --dst rtmp://a.rtmp.youtube.com/live2/YOUKEY --mono --watchdog --sec 30 >> /dev/null 2>&1
* * * * * /usr/bin/python3 ~/ffmpeger.py -s ~/media.mp4 --dst rtmp://b.rtmp.youtube.com/live2?backup=1/YOUKEY --mono >> /dev/null 2>&1
```
Example usage in terminal with make the script executable:
```shell
chmod u+x ./ffmpeger.py
ffmpeger.py -s rtsp://user:pass@host:554/Streaming/Channels/101 --dst rtp://239.0.0.1:5554
```
Example usage in Python:
```Python
from ffmpeger import FFmpeg
FFmpeg.run(src='null, anull', preset='240p', fps=10)
```
____
## `procutil`.py
**Description:** Find a running process from Python
**Dependencies:** Python 3 (tested version 3.9.5)
| PARAMETERS | DESCRIPTION | DEFAULT|
|-------------|-------------|--------|
|**[-h]**|print help and exit||
|**[--find]**|find process pid, name or arguments||
|**[--exclude]**|exclude process pid, name or arguments|`None`|
|**[--self]**|find a clones of self|`True`|
|**[--kill]**|kill the process with pid||
Example usage in terminal with Python for find all running processes:
```shell
python3 ./procutil.py
```
Example usage in terminal with make the script executable for find all specified processes:
```shell
chmod u+x ./procutil.py
./procutil.py --find ssh --exclude sftp
```
Example usage in Python for find a clones of self:
```Python
from os import getpid
from sys import argv
from procutil import Proc
processes = Proc.search(' '.join(argv), str(getpid()))
if processes:
for process in processes:
print(process)
```
## `sendmail`.py
**Description:** Sending email from Python
**Dependencies:** Python 3 (tested version 3.9.5)
@ -210,15 +109,20 @@ if processes:
|**[--time]**|minutes of attempts to send|3|
Example usage in terminal with Python:
```shell
python3 ./sendmail.py -u user@gmail.com -p pass -d addr1@gmail.com,addr2@gmail.com
```
Example usage in terminal with make the script executable:
```shell
chmod u+x ./sendmail.py
./sendmail.py -u user@gmail.com -p pass -d addr1@gmail.com,addr2@gmail.com
```
Example usage in Python:
```Python
from sendmail import Mail
@ -228,7 +132,9 @@ print(log)
```
____
## `simplewc`.py
**Description:** Update Let's Encrypt wildcard certificate with DNS-01 challenge
**Dependencies:** Python 3 (tested version 3.9.5),
installed or downloaded [acme.sh](https://github.com/Neilpang/acme.sh),
@ -247,13 +153,17 @@ dns is supported to [dynamic update](https://en.wikipedia.org/wiki/Dynamic_DNS)
|**[--test]**|"test" argument for the acme.sh|`False`|
Example usage in cron with Python:
```shell
# at 00:00 on Monday
0 0 * * 1 /usr/bin/python3 ~/simplewc.py --domain EXAMPLE.COM --server 8.8.8.8 --keyname KEY --keydata YOU_KEY_CONTENT > /dev/null
# 00:00 on day-of-month 1 and 15
0 0 1,15 * * /usr/bin/python3 ~/simplewc.py --domain EXAMPLE.COM --server dyn.dns.he.net --keyname - --keydata YOU_DDNSKEY > /dev/null
```
Example usage in terminal with make the script executable:
```shell
chmod u+x ./simplewc.py
./simplewc.py --domain EXAMPLE.COM --server 8.8.8.8 --keyname KEY --keydata YOU_KEY_CONTENT --test --force

0
camsutil/__init__.py
View File

144
camsutil/ptz.py
View File

@ -1,144 +0,0 @@
#!/usr/bin/env python3
import urllib.request
from os import path, sep
class API:
"""
Creation of a request to the camera API based on the prepared template.
"""
def __init__(self, host: str, user: str, password: str, template: str,
protocol: str = 'http', port: int = 80, channel: int = 101, vid: int = 1,
x: int = 0, y: int = 0, z: int = 0, speed: int = 1, time: int = 100000,
text: str = None, enabled: str = 'true'):
"""
Object constructor
:param host: hostname or ip address
:param user: valid user
:param password: valid password
:param template: the name of an existing template
:param protocol: http, https, etc.
:param port: port number
:param channel: ptz channel number
:param vid: video channel id
:param x: horizontal positioning: azimuth, pan
:param y: vertical positioning: elevation, tilt
:param z: zoom direction, absolute zoom
:param speed: positioning speed: from 1 to 7
:param time: momentary duration, max 100000ms
:param text: overlay text content
:param enabled: enabled (true) or disabled (false) overlay text
"""
self._host = host
self._user = user
self._pswd = password
self._temp = path.dirname(path.abspath(__file__)) + sep + 'templates' + sep + template
self._protocol = protocol
self._port = port
self._channel = channel
self._id = vid
self._x = x
self._y = y
self._z = z
self._speed = speed
self._time = time
self._message = text
self._enabledMessage = enabled
self._data = ''
self._type = ''
self._url = self._protocol + '://' + self._host + ':' + str(self._port)
self._method = ''
with open(self._temp) as file:
content = file.read() \
.replace('@CHANNEL@', str(self._channel)) \
.replace('@ID@', str(self._id)) \
.replace('@XXXX@', str(self._x)) \
.replace('@YYYY@', str(self._y)) \
.replace('@ZZZZ@', str(self._z)) \
.replace('@TTTT@', str(self._time)) \
.replace('@SSSS@', str(self._speed)) \
.replace('@MSG@', str(self._message)) \
.replace('@ENABLED@', str(self._enabledMessage))
for line in content.splitlines():
if not ('<!--' in line):
self._data += line + '\n'
elif 'PATH:' in line:
self._url += line.split(' ')[2]
elif 'METHOD:' in line:
self._method += line.split(' ')[2]
elif 'TYPE:' in line:
self._type += line.split(' ')[2]
def call(self):
"""
Call a request to the camera api
:return: response from the camera api
"""
# Preparing authorization
pswd = urllib.request.HTTPPasswordMgrWithDefaultRealm()
pswd.add_password(None, self._url, self._user, self._pswd)
auth = urllib.request.HTTPDigestAuthHandler(pswd)
urllib.request.install_opener(urllib.request.build_opener(auth))
# Request
request = urllib.request.Request(url=self._url, data=bytes(self._data.encode('utf-8')), method=self._method)
request.add_header('Content-Type', self._type)
# Response
response = urllib.request.urlopen(request).read()
if response.startswith(b'\xff\xd8'):
return response
else:
return str(response.decode('utf-8'))
if __name__ == "__main__":
from argparse import ArgumentParser
args = ArgumentParser(
prog='PTZ',
description='Creation of a request to the camera API based on the prepared template',
epilog='Dependencies: Python 3 (tested version 3.9.5)'
)
args.add_argument('--host', type=str, required=True,
help='hostname or ip address')
args.add_argument('--user', type=str, required=True,
help='valid user')
args.add_argument('--password', type=str, required=True,
help='valid password')
args.add_argument('--template', type=str, required=True,
help='the name of an existing template')
args.add_argument('--protocol', type=str, default='http', required=False,
help='http, https, etc.')
args.add_argument('--port', type=int, default=80, required=False,
help='port number')
args.add_argument('--channel', type=int, default=101, required=False,
help='ptz channel number')
args.add_argument('--vid', type=int, default=1, required=False,
help='video channel id')
args.add_argument('--x', type=int, default=0, required=False,
help='horizontal positioning: azimuth, pan')
args.add_argument('--y', type=int, default=0, required=False,
help='vertical positioning: elevation, tilt')
args.add_argument('--z', type=int, default=0, required=False,
help='zoom direction, absolute zoom')
args.add_argument('--speed', type=int, default=1, required=False,
help='positioning speed: from 1 to 7')
args.add_argument('--time', type=int, default=100000, required=False,
help='momentary duration, max 100000ms')
args.add_argument('--text', type=str, default=None, required=False,
help='overlay text content')
args.add_argument('--enabled', type=str, default='true', required=False,
help='enabled (true) or disabled (false) overlay text')
args = args.parse_args()
action = API(
host=args.host, user=args.user, password=args.password, template=args.template,
protocol=args.protocol, port=args.port, channel=args.channel, vid=args.vid,
x=args.x, y=args.y, z=args.z, speed=args.speed, time=args.time,
text=args.text, enabled=args.enabled
)
print(action.call())

3
camsutil/templates/Hikvision_GetCapabilities.html
View File

@ -1,3 +0,0 @@
<!-- PATH: /ISAPI/PTZCtrl/channels/1/capabilities -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->

3
camsutil/templates/Hikvision_GetJPEG.html
View File

@ -1,3 +0,0 @@
<!-- PATH: /Streaming/channels/@ID@/picture?snapShotImageType=JPEG&videoResolutionWidth=@XXXX@&videoResolutionHeight=@YYYY@ -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->

3
camsutil/templates/Hikvision_GetPosition.html
View File

@ -1,3 +0,0 @@
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/status -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->

3
camsutil/templates/Hikvision_GetReboot.html
View File

@ -1,3 +0,0 @@
<!-- PATH: /System/reboot -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->

12
camsutil/templates/Hikvision_GoAbsolute.xml
View File

@ -1,12 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<PTZData>
<AbsoluteHigh>
<elevation>@YYYY@</elevation>
<azimuth>@XXXX@</azimuth>
<absoluteZoom>@ZZZZ@</absoluteZoom>
</AbsoluteHigh>
</PTZData>
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/absolute -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->
<!-- XXXX: 0..3600 YYYY: -900..2700 ZZZZ: 0..1000 -->

12
camsutil/templates/Hikvision_GoContinuous.xml
View File

@ -1,12 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<PTZData>
<pan>@XXXX@</pan>
<tilt>@YYYY@</tilt>
<zoom>@ZZZZ@</zoom>
</PTZData>
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/continuous -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->
<!-- XXXX: -100..100 YYYY: -100..100 ZZZZ: -100..100 -->
<!-- x360* = 3300..10 sec x1* = 9,167..0,028 sec -->
<!-- Max action time: 180sec -->

5
camsutil/templates/Hikvision_GoHomeposition.xml
View File

@ -1,5 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<PTZData></PTZData>
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/homeposition/goto -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->

13
camsutil/templates/Hikvision_GoMomentary.xml
View File

@ -1,13 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<PTZData>
<pan>@XXXX@</pan>
<tilt>@YYYY@</tilt>
<zoom>@ZZZZ@</zoom>
<Momentary>
<duration>@TTTT@</duration>
</Momentary>
</PTZData>
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/momentary -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->
<!-- XXXX: -100..100 YYYY: -100..100 ZZZZ: -100..100 TTTT: -->

4
camsutil/templates/Hikvision_PtzDown.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=TILT_DOWN&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

4
camsutil/templates/Hikvision_PtzLeft.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=PAN_LEFT&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

4
camsutil/templates/Hikvision_PtzPreset.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=GOTO_PRESET&presetNo=@ID@&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

4
camsutil/templates/Hikvision_PtzRight.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=PAN_RIGHT&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

3
camsutil/templates/Hikvision_PtzStop.html
View File

@ -1,3 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=GOTO_PRESET&mode=stop -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->

4
camsutil/templates/Hikvision_PtzUp.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=TILT_UP&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

4
camsutil/templates/Hikvision_PtzZoomIn.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=ZOOM_OUT&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

4
camsutil/templates/Hikvision_PtzZoomOut.html
View File

@ -1,4 +0,0 @@
<!-- PATH: /PTZ/channels/1/PTZControl?command=ZOOM_IN&speed=@SSSS@&mode=start -->
<!-- METHOD: GET -->
<!-- TYPE: application/x-www-form-urlencoded -->
<!-- SSSS: 1..7 -->

5
camsutil/templates/Hikvision_SetHomeposition.xml
View File

@ -1,5 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<PTZData></PTZData>
<!-- PATH: /ISAPI/PTZCtrl/channels/@CHANNEL@/homeposition -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->

13
camsutil/templates/Hikvision_SetTextOverlay.xml
View File

@ -1,13 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<TextOverlayList version="1.0" xmlns="http://www.hikvision.com/ver10/XMLSchema">
<TextOverlay>
<id>@ID@</id>
<enabled>@ENABLED@</enabled>
<posX>@XXXX@</posX>
<posY>@YYYY@</posY>
<message>@MSG@</message>
</TextOverlay>
</TextOverlayList>
<!-- PATH: /Video/inputs/channels/@CHANNEL@/overlays/text -->
<!-- METHOD: PUT -->
<!-- TYPE: text/xml -->

202
ffmpeger.py
View File

@ -1,202 +0,0 @@
#!/usr/bin/env python3
from multiprocessing import Process, Queue
from os import path, environ
from subprocess import Popen, PIPE, STDOUT
from sys import platform
from time import sleep
from procutil import Proc # or copy class Proc from file procutil.py here for autonomy of this file
class FFmpeg:
"""
FFmpeg management from Python
"""
@classmethod
def run(cls, src: str, preset: str = None, fps: int = None, dst: str = None,
ffpath: str = None, watchdog: bool = False, sec: int = 5, mono: bool = False):
"""
Running the installed ffmpeg
:param src: sources urls (example: "rtsp://user:pass@host:554/Streaming/Channels/101, anull")
:param preset: 240p, 360p, 480p, 720p, 1080p, 1440p, 2160p
:param fps: frame per second encoding output
:param dst: destination url (example: rtp://239.0.0.1:5554)
:param ffpath: alternative path to bin (example: /usr/bin/ffmpeg)
:param watchdog: detect ffmpeg freeze and terminate
:param sec: seconds to wait before the watchdog terminates
:param mono: detect ffmpeg running copy and terminate
:return: None
"""
process = cls._bin(ffpath).split()+cls._src(src).split()+cls._preset(preset, fps).split()+cls._dst(dst).split()
if mono and Proc.search(' '.join(process)):
print('Process already exist, exit...')
else:
with Popen(process, stdout=PIPE, stderr=STDOUT) as proc:
que = None
if watchdog:
que = Queue()
Process(target=cls._watchdog, args=(proc.pid, sec, que,), daemon=True).start()
for line in proc.stdout:
if not que:
print(line, flush=True)
else:
que.put(line)
exit()
@classmethod
def _bin(cls, path_ffmpeg: str):
"""
Returns the path to the ffmpeg depending on the OS
:param path_ffmpeg: alternative path to bin
:return: path to ffmpeg
"""
faq = ('\n'
'Main download page: https://ffmpeg.org/download.html\n'
'\n'
'Install on Linux (Debian):\n'
'\tsudo apt install -y ffmpeg\n'
'\tTarget: /usr/bin/ffmpeg\n'
'\n'
'Install on Windows:\n'
'\tDownload and extract archive from: https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-full.7z\n'
'\tTarget: "%PROGRAMFILES%\\ffmpeg\\bin\\ffmpeg.exe"\n'
'\n'
'Install on MacOS:\n'
'\tDownload and extract archive from: https://evermeet.cx/ffmpeg/\n'
'\tTarget: /usr/bin/ffmpeg\n')
if not path_ffmpeg:
if platform.startswith('linux') or platform.startswith('darwin'):
path_ffmpeg = '/usr/bin/ffmpeg'
elif platform.startswith('win32'):
path_ffmpeg = environ['PROGRAMFILES'] + "\\ffmpeg\\bin\\ffmpeg.exe"
if path.exists(path_ffmpeg):
return path_ffmpeg
else:
print('ON', platform, 'PLATFORM', 'not found ffmpeg', faq)
return None
@classmethod
def _src(cls, sources: str):
"""
Parsing sources into ffmpeg format
:param sources: comma-separated list of sources in string format
:return: ffmpeg format list of sources
"""
list_sources = []
for src in sources.split(','):
src = src.strip()
if 'null' in src:
src = ' '.join(['-f lavfi -i', src])
elif 'rtsp' in src:
src = ' '.join(['-rtsp_transport tcp -i', src])
else:
src = ' '.join(['-stream_loop -1 -re -i', src])
list_sources.append(src)
return ' '.join(list_sources)
@classmethod
def _preset(cls, choice: str, fps: int):
"""
Parsing preset into ffmpeg format
:param choice: preset selection
:param fps: frame per second encoding output
:return: ffmpeg format encoding parameters
"""
tune = '-tune zerolatency'
video = '-c:v copy'
audio = '-c:a aac -b:a 128k'
width, height, kbps = None, None, None
if choice:
if choice == '240p':
width, height, kbps = 426, 240, 480
if choice == '360p':
width, height, kbps = 640, 360, 720
if choice == '480p':
width, height, kbps = 854, 480, 1920
if choice == '720p':
width, height, kbps = 1280, 720, 3960
if choice == '1080p':
width, height, kbps = 1920, 1080, 5940
if choice == '1440p':
width, height, kbps = 2560, 1440, 12960
if choice == '2160p':
width, height, kbps = 3840, 2160, 32400
if width and height and kbps:
video = ''.join(['-vf scale=', str(width), ':', str(height), ',setsar=1:1'])
video = ' '.join([video, '-c:v libx264 -pix_fmt yuv420p -preset ultrafast'])
if fps:
video = ' '.join([video, '-r', str(fps), '-g', str(fps * 2)])
video = ' '.join([video, '-b:v', str(kbps) + 'k'])
return ' '.join([tune, video, audio])
@classmethod
def _dst(cls, destination: str):
"""
Parsing destination into ffmpeg format
:param destination:
:return: ffmpeg format destination
"""
container = '-f null'
stdout = '-v debug' # '-nostdin -nostats' # '-report'
if destination:
if 'rtmp' in destination:
container = '-f flv'
elif "rtp" in destination:
container = '-f rtp_mpegts'
else:
destination = '-'
return ' '.join([container, destination, stdout])
@classmethod
def _watchdog(cls, pid: int, sec: int = 5, que: Queue = None):
"""
If no data arrives in the queue, kill the process
:param pid: process ID
:param sec: seconds to wait for data
:param que: queue pointer
:return: None
"""
if que:
while True:
while not que.empty():
print(que.get())
sleep(sec)
if que.empty():
Proc.kill(pid)
print('exit by watchdog')
break
exit()
if __name__ == "__main__":
from argparse import ArgumentParser
args = ArgumentParser(
prog='FFmpeger',
description='FFmpeg management from Python',
epilog='Dependencies: '
'Python 3 (tested version 3.9.5), '
'installed or downloaded ffmpeg, '
'procutil.py in the same directory'
)
args.add_argument('-s', '--src', type=str, required=True,
help='sources urls (example: "rtsp://user:pass@host:554/Streaming/Channels/101, anull")')
args.add_argument('--preset', type=str, default=None, required=False,
help='240p, 360p, 480p, 720p, 1080p, 1440p, 2160p')
args.add_argument('--fps', type=int, default=None, required=False,
help='frame per second encoding output')
args.add_argument('--dst', type=str, default=None, required=False,
help='destination url (example: rtp://239.0.0.1:5554)')
args.add_argument('--ffpath', type=str, default=None, required=False,
help='alternative path to bin (example: /usr/bin/ffmpeg)')
args.add_argument('--watchdog', action='store_true', required=False,
help='detect ffmpeg freeze and terminate')
args.add_argument('--sec', type=int, default=15, required=False,
help='seconds to wait before the watchdog terminates')
args.add_argument('--mono', action='store_true', required=False,
help='detect ffmpeg running copy and terminate')
args = vars(args.parse_args())
FFmpeg.run(src=args['src'], preset=args['preset'], fps=args['fps'], dst=args['dst'],
ffpath=args['ffpath'], watchdog=args['watchdog'], sec=args['sec'], mono=args['mono'])

134
procutil.py
View File

@ -1,134 +0,0 @@
#!/usr/bin/env python3
from os import path, getpid
from subprocess import Popen, PIPE
from sys import argv, platform
class Proc:
"""
Find a running process from Python
"""
@classmethod
def _list_windows(cls):
"""
Find all running process with wmi
:return: list of dictionaries with descriptions of found processes
"""
execlist = []
separate = b'\r\r\n'
out, err = Popen(['wmic', 'process', 'get', 'CommandLine,ExecutablePath,Name,ProcessId', '/format:list'],
stdout=PIPE, stderr=PIPE).communicate()
for line in out.split(separate + separate):
execpid, exename, exepath, cmdline = None, None, None, None
for subline in line.split(separate):
if b'ProcessId=' in subline:
execpid = subline.split(b'=')[1].decode('utf-8')
if b'Name=' in subline:
exename = subline.split(b'=')[1].decode('utf-8')
if b'ExecutablePath=' in subline:
exepath = subline.split(b'=')[1].decode('utf-8')
if b'CommandLine=' in subline:
cmdline = subline.split(b'=')[1].decode('utf-8')
if execpid and exename:
execlist.append({'execpid': execpid, 'exename': exename, 'exepath': exepath, 'cmdline': cmdline})
return execlist
@classmethod
def _list_linux(cls):
"""
Find all running process with ps
:return: list of dictionaries with descriptions of found processes
"""
execlist = []
out, err = Popen(['/bin/ps', '-eo', 'pid,args'], stdout=PIPE, stderr=PIPE).communicate()
for line in out.splitlines():
execpid = line.split()[0].decode('utf-8')
exepath = line.split()[1].decode('utf-8')
exename = path.basename(exepath)
cmdline = line.split(None, 1)[1].decode('utf-8')
if execpid and exename:
execlist.append({'execpid': execpid, 'exename': exename, 'exepath': exepath, 'cmdline': cmdline})
return execlist
@classmethod
def list(cls):
"""
Find all running process
:return: list of dictionaries with descriptions of found processes
"""
if platform.startswith('linux') or platform.startswith('darwin'):
return cls._list_linux()
elif platform.startswith('win32'):
return cls._list_windows()
else:
return None
@classmethod
def search(cls, find: str, exclude: str = None):
"""
Find specified processes
:param find: find process pid, name or arguments
:param exclude: exclude process pid, name or arguments
:return: list of dictionaries with descriptions of found processes
"""
proc_found = []
try:
for proc in cls.list():
if exclude and (exclude in proc['execpid'] or exclude in proc['exename'] or
exclude in proc['exepath'] or exclude in proc['cmdline']):
pass
elif find in proc['execpid'] or find in proc['exename'] or \
find in proc['exepath'] or find in proc['cmdline']:
proc_found.append(proc)
except TypeError as ex:
print('ON', platform, 'PLATFORM', 'search ERROR:', ex)
finally:
if len(proc_found) == 0:
return None
else:
return proc_found
@classmethod
def kill(cls, pid: int):
"""
Kill the process by means of the OS
:param pid: process ID
:return: None
"""
if platform.startswith('linux') or platform.startswith('darwin'):
Popen(['kill', '-s', 'SIGKILL', str(pid)])
elif platform.startswith('win32'):
Popen(['taskkill', '/PID', str(pid), '/F'])
if __name__ == "__main__":
from argparse import ArgumentParser
args = ArgumentParser(
prog='Proc',
description='Find a running process from Python',
epilog='Dependencies: Python 3 (tested version 3.9.5)'
)
args.add_argument('--find', type=str, required=False,
help='find process pid, name or arguments')
args.add_argument('--exclude', type=str, default=None, required=False,
help='exclude process pid, name or arguments')
args.add_argument('--self', action='store_true', required=False,
help='find a clones of self')
args.add_argument('--kill', type=int, required=False,
help='kill the process with pid')
args = vars(args.parse_args())
processes = None
if args['kill']:
Proc.kill(args['kill'])
elif args['find']:
processes = Proc.search(args['find'], args['exclude'])
elif args['self']:
processes = Proc.search(' '.join(argv), str(getpid()))
else:
processes = Proc.list()
if processes:
for process in processes:
print(process)