Initial import: sources, staging packages, build scripts and CI workflow
This commit is contained in:
Executable
+31
@@ -0,0 +1,31 @@
|
||||
#!/bin/sh
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
CONTROL_FILE=$(ls -1 *.control)
|
||||
|
||||
echo "Using control file $CONTROL_FILE"
|
||||
PACKAGE_NAME=$(awk '/^Package: /{print $NF}' $CONTROL_FILE)
|
||||
VERSION=$(awk '/^Version: /{print $NF}' $CONTROL_FILE)
|
||||
|
||||
|
||||
|
||||
|
||||
# tag git repo
|
||||
# git tag -f release-${VERSION}
|
||||
|
||||
echo "Building package $PACKAGE_NAME ($VERSION): $PACKAGE_NAME-${VERSION}_all.deb"
|
||||
# VERSION
|
||||
sed "s/{VERSION}/${VERSION}/" $PACKAGE_NAME.control > packaging/$PACKAGE_NAME/DEBIAN/control
|
||||
|
||||
#create debian package
|
||||
dpkg --build packaging/$PACKAGE_NAME $PACKAGE_NAME-${VERSION}_all.deb
|
||||
|
||||
# dpkg-sig -k 14AF7F05 --sign builder icosi-${VERSION}_all.deb
|
||||
|
||||
echo "clean up"
|
||||
rm packaging/$PACKAGE_NAME/DEBIAN/control
|
||||
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
Package: eendoe-hyperdeck
|
||||
Version: {VERSION}
|
||||
Section: utils
|
||||
Maintainer: nl.eendoe.bart
|
||||
Architecture: all
|
||||
Depends: python3, python3-pip, ffmpeg, eendoe-srs
|
||||
Description: Eendoe hyperdeck client module
|
||||
@@ -0,0 +1,41 @@
|
||||
#!/bin/bash
|
||||
|
||||
#set -e
|
||||
case "$1" in
|
||||
configure|abort-upgrade)
|
||||
|
||||
id -u vid &>/dev/null || useradd --system -m vid
|
||||
|
||||
runuser -l vid -c "pip3 install -r /usr/share/hyperdeck/requirements.txt"
|
||||
|
||||
|
||||
|
||||
FILE="/etc/hyperdeck/config.json"
|
||||
SOURCEFILE="/usr/share/hyperdeck/config.json.orig"
|
||||
if [ -f "$FILE" ]; then
|
||||
echo "$FILE exists."
|
||||
else
|
||||
mkdir -p /etc/hyperdeck/
|
||||
echo "Copy default config"
|
||||
cp "$SOURCEFILE" "$FILE"
|
||||
chmod 744 "$FILE"
|
||||
fi
|
||||
|
||||
|
||||
|
||||
systemd-notify --booted
|
||||
if [ "$?" -eq 0 ]
|
||||
then
|
||||
echo "Systemd running, registering service"
|
||||
systemctl daemon-reload && systemctl enable hyperdeck
|
||||
else
|
||||
echo "Systemd not running, skipping"
|
||||
fi
|
||||
|
||||
mkdir -p /var/log/hyperdeck/
|
||||
chown vid:vid "/var/log/hyperdeck/"
|
||||
|
||||
echo "Installation done"
|
||||
;;
|
||||
esac
|
||||
exit 0
|
||||
@@ -0,0 +1,7 @@
|
||||
#!/bin/sh
|
||||
set -e
|
||||
case "$1" in
|
||||
install|upgrade)
|
||||
;;
|
||||
esac
|
||||
exit 0
|
||||
@@ -0,0 +1,4 @@
|
||||
#!/usr/bin/make -f
|
||||
|
||||
%:
|
||||
dh $@
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
#
|
||||
# Systemd unit file for Hyperdeck recorder
|
||||
#
|
||||
|
||||
[Unit]
|
||||
Description=Eendoe Hyperdeck recorder manager
|
||||
After=network.target
|
||||
RequiresMountsFor=/mnt/video
|
||||
|
||||
[Service]
|
||||
|
||||
|
||||
# Lifecycle
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python3 /usr/share/hyperdeck/main.py -c /etc/hyperdeck/config.json
|
||||
SuccessExitStatus=0
|
||||
Restart=on-abort
|
||||
WorkingDirectory=/usr/share/hyperdeck/
|
||||
|
||||
# Logging
|
||||
SyslogIdentifier=hyperdeck
|
||||
|
||||
# Security
|
||||
User=vid
|
||||
Group=vid
|
||||
PrivateTmp=yes
|
||||
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"videopath": "/mnt/video/",
|
||||
"hyperdeck": {"ip": "10.31.35.221", "port": 9993},
|
||||
"rtsp_url" : "rtsp://admin:admin@10.31.35.205:554/1",
|
||||
"log_path": "/var/log/hyperdeck/"
|
||||
}
|
||||
+264
@@ -0,0 +1,264 @@
|
||||
import os
|
||||
from typing import BinaryIO
|
||||
|
||||
import logging
|
||||
from fastapi import Request,FastAPI, HTTPException, status
|
||||
from fastapi.responses import JSONResponse, FileResponse,StreamingResponse
|
||||
from hyperdeck.application import HyperDeck
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
class ScheduleItemRequest(BaseModel):
|
||||
start: datetime| None
|
||||
end: datetime| None
|
||||
id: str| None
|
||||
title: str| None
|
||||
|
||||
|
||||
def send_bytes_range_requests(
|
||||
file_obj: BinaryIO, start: int, end: int, chunk_size: int = 10_000
|
||||
):
|
||||
"""Send a file in chunks using Range Requests specification RFC7233
|
||||
|
||||
`start` and `end` parameters are inclusive due to specification
|
||||
"""
|
||||
with file_obj as f:
|
||||
f.seek(start)
|
||||
while (pos := f.tell()) <= end:
|
||||
read_size = min(chunk_size, end + 1 - pos)
|
||||
yield f.read(read_size)
|
||||
|
||||
|
||||
def _get_range_header(range_header: str, file_size: int) -> tuple[int, int]:
|
||||
def _invalid_range():
|
||||
return HTTPException(
|
||||
status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
|
||||
detail=f"Invalid request range (Range:{range_header!r})",
|
||||
)
|
||||
|
||||
try:
|
||||
h = range_header.replace("bytes=", "").split("-")
|
||||
start = int(h[0]) if h[0] != "" else 0
|
||||
end = int(h[1]) if h[1] != "" else file_size - 1
|
||||
except ValueError:
|
||||
raise _invalid_range()
|
||||
|
||||
if start > end or start < 0 or end > file_size - 1:
|
||||
raise _invalid_range()
|
||||
return start, end
|
||||
|
||||
|
||||
def range_requests_response(
|
||||
request: Request, file_path: str, content_type: str
|
||||
):
|
||||
"""Returns StreamingResponse using Range Requests of a given file"""
|
||||
|
||||
file_size = os.stat(file_path).st_size
|
||||
range_header = request.headers.get("range")
|
||||
|
||||
headers = {
|
||||
"content-type": content_type,
|
||||
"accept-ranges": "bytes",
|
||||
"content-encoding": "identity",
|
||||
"content-length": str(file_size),
|
||||
"access-control-expose-headers": (
|
||||
"content-type, accept-ranges, content-length, "
|
||||
"content-range, content-encoding"
|
||||
),
|
||||
}
|
||||
start = 0
|
||||
end = file_size - 1
|
||||
status_code = status.HTTP_200_OK
|
||||
|
||||
if range_header is not None:
|
||||
start, end = _get_range_header(range_header, file_size)
|
||||
size = end - start + 1
|
||||
headers["content-length"] = str(size)
|
||||
headers["content-range"] = f"bytes {start}-{end}/{file_size}"
|
||||
status_code = status.HTTP_206_PARTIAL_CONTENT
|
||||
|
||||
return StreamingResponse(
|
||||
send_bytes_range_requests(open(file_path, mode="rb"), start, end),
|
||||
headers=headers,
|
||||
status_code=status_code,
|
||||
)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
CHUNK_SIZE = 1024*1024
|
||||
|
||||
|
||||
|
||||
@app.exception_handler(RuntimeError)
|
||||
async def validation_exception_handler(request: Request, exc: RuntimeError) -> JSONResponse:
|
||||
return JSONResponse({
|
||||
"status":"error",
|
||||
"message": str(exc)
|
||||
},500)
|
||||
|
||||
|
||||
@app.get("/videos/", response_class=JSONResponse)
|
||||
def read_fonts():
|
||||
app = HyperDeck()
|
||||
|
||||
|
||||
files = []
|
||||
for f in app.getVideos():
|
||||
files.append({
|
||||
"name": f["name"],
|
||||
"filesize": f["filesize"],
|
||||
"start": f["start"],
|
||||
"end": f["end"],
|
||||
"title": f["title"],
|
||||
|
||||
})
|
||||
|
||||
|
||||
return files
|
||||
|
||||
|
||||
@app.get("/status/", response_class=JSONResponse)
|
||||
def getStatus():
|
||||
app = HyperDeck()
|
||||
|
||||
date = datetime.now()
|
||||
next = app.getNextScheduled()
|
||||
|
||||
if next:
|
||||
nexttime = next.start.isoformat()
|
||||
else:
|
||||
nexttime = "none"
|
||||
|
||||
|
||||
return JSONResponse({
|
||||
"status": app.getStatus().value,
|
||||
"errormessage" : app.getLastError(),
|
||||
"date": date.isoformat(), #strftime("%Y%m%d%H%M%S"),
|
||||
"nextrecording": nexttime
|
||||
})
|
||||
|
||||
@app.post("/status/record", response_class=JSONResponse)
|
||||
async def startManualRecording(request: Request):
|
||||
logging.info("Manually set status to recording" )
|
||||
app = HyperDeck()
|
||||
|
||||
if app.startRecording():
|
||||
return {"status":"ok"}
|
||||
else:
|
||||
return {"status":"error"}
|
||||
|
||||
|
||||
@app.post("/status/stop", response_class=JSONResponse)
|
||||
async def stopRecording(request: Request):
|
||||
logging.info("Manually set status to idle" )
|
||||
app = HyperDeck()
|
||||
if app.stopRecording():
|
||||
return {"status":"ok"}
|
||||
else:
|
||||
return {"status":"error"}
|
||||
|
||||
|
||||
@app.get("/schedule/active", response_class=JSONResponse)
|
||||
async def getScheduleActive(request: Request):
|
||||
app = HyperDeck()
|
||||
return app.schedule.getActiveItems()
|
||||
|
||||
@app.get("/schedule", response_class=JSONResponse)
|
||||
async def getSchedule(request: Request):
|
||||
app = HyperDeck()
|
||||
return app.getSchedule()
|
||||
|
||||
|
||||
|
||||
|
||||
@app.get("/schedule/lastday", response_class=JSONResponse)
|
||||
async def getItemsLastDay(request: Request):
|
||||
app = HyperDeck()
|
||||
return app.schedule.getItemsLastDay()
|
||||
|
||||
|
||||
@app.post("/schedule", response_class=JSONResponse)
|
||||
async def createScheduledItem(item: ScheduleItemRequest ):
|
||||
app = HyperDeck()
|
||||
|
||||
item.start = item.start.replace(tzinfo=None)
|
||||
item.end = item.end.replace(tzinfo=None)
|
||||
|
||||
if item.end<= item.start:
|
||||
raise HTTPException(status_code=422, detail="End time must be later then start time")
|
||||
|
||||
if not app.isFree(item.start, item.end,""):
|
||||
raise HTTPException(status_code=422, detail="Time slot already allocated")
|
||||
|
||||
app.addScheduledRecording(item.start, item.end, item.title)
|
||||
return app.getSchedule()
|
||||
|
||||
@app.put("/schedule", response_class=JSONResponse)
|
||||
async def editScheduledItem(item: ScheduleItemRequest ):
|
||||
app = HyperDeck()
|
||||
|
||||
item.start = item.start.replace(tzinfo=None)
|
||||
item.end = item.end.replace(tzinfo=None)
|
||||
|
||||
if item.end<= item.start:
|
||||
raise HTTPException(status_code=422, detail="End time must be later then start time")
|
||||
|
||||
|
||||
if not app.isFree(item.start, item.end,item.id):
|
||||
raise HTTPException(status_code=422, detail="Time slot already allocated")
|
||||
|
||||
|
||||
i = app.schedule.getScheduleItem(item.id)
|
||||
if i.status != "waiting":
|
||||
raise HTTPException(status_code=422, detail="Unable to update recorded item")
|
||||
|
||||
i.start = item.start
|
||||
i.end = item.end
|
||||
i.title = item.title
|
||||
app.saveSchedule()
|
||||
return app.getSchedule()
|
||||
|
||||
@app.delete("/schedule", response_class=JSONResponse)
|
||||
async def deleteScheduledItem(item: ScheduleItemRequest ):
|
||||
app = HyperDeck()
|
||||
i = app.schedule.getScheduleItem(item.id)
|
||||
if i.status != "waiting":
|
||||
raise HTTPException(status_code=422, detail="Unable to delete recorded item")
|
||||
|
||||
app.removeScheduledItem(i)
|
||||
|
||||
return app.getSchedule()
|
||||
|
||||
|
||||
@app.get("/videos/get/{video}", response_class=FileResponse)
|
||||
async def read_item(video: str):
|
||||
app = HyperDeck()
|
||||
|
||||
f = app.getVideo(video)
|
||||
|
||||
logging.info("Downloading video " + f["path"])
|
||||
|
||||
return FileResponse(f["path"],headers={'Content-Disposition': 'attachement; filename="'+f["filename"]+'"'})
|
||||
|
||||
|
||||
@app.get("/videos/view/{video}", response_class=StreamingResponse)
|
||||
async def stream_item(video: str, request: Request):
|
||||
app = HyperDeck()
|
||||
f = app.getVideo(video)
|
||||
return range_requests_response(
|
||||
request, file_path=f["path"], content_type="video/mp4"
|
||||
)
|
||||
|
||||
|
||||
@app.get("/videos/{video}", response_class=JSONResponse)
|
||||
async def stream_item_alt(video: str, request: Request):
|
||||
app = HyperDeck()
|
||||
f = app.getVideo(video)
|
||||
return f
|
||||
|
||||
@app.delete("/videos/{video}", response_class=JSONResponse)
|
||||
async def deleteVideo(video: str ):
|
||||
app = HyperDeck()
|
||||
app.removeVideo(video)
|
||||
return app.getSchedule()
|
||||
+436
@@ -0,0 +1,436 @@
|
||||
from enum import Enum
|
||||
import json
|
||||
import socket
|
||||
from hyperdeck.db import Schedule, ScheduleItem, ScheduleItemStatus
|
||||
from hyperdeck.utils import Singleton
|
||||
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
import datetime
|
||||
|
||||
|
||||
class Config(metaclass=Singleton):
|
||||
def open(self, filename):
|
||||
f = open(filename, "r")
|
||||
j = json.load(f)
|
||||
self.fromJson(j)
|
||||
|
||||
def save(self, filename):
|
||||
with open(filename, "w") as f:
|
||||
f.write(json.dumps(self.toJson()))
|
||||
|
||||
def toJson(self):
|
||||
return {}
|
||||
|
||||
def fromJson(self, j):
|
||||
self.videopath = j["videopath"]
|
||||
self.hyperdeck_ip = j["hyperdeck"]["ip"]
|
||||
self.hyperdeck_port = j["hyperdeck"]["port"]
|
||||
|
||||
if "max_age" in j:
|
||||
self.max_age =int( j["max_age"])
|
||||
else:
|
||||
self.max_age = 90
|
||||
|
||||
|
||||
if "max_originals_age" in j:
|
||||
self.maxmax_originals_age__originals_age =int( j["max_originals_age"])
|
||||
else:
|
||||
self.max_originals_age = 5
|
||||
|
||||
if "rtsp_url" in j:
|
||||
self.rtsp_url = j["rtsp_url"]
|
||||
|
||||
self.videopath_originals = os.path.join(self.videopath, "originals")
|
||||
self.videopath_convert = os.path.join(self.videopath, "convert")
|
||||
|
||||
if "log_path" in j:
|
||||
self.log_path = j["log_path"]
|
||||
else:
|
||||
self.log_path = "/var/log/hyperdeck/"
|
||||
|
||||
pass
|
||||
|
||||
def __init__(self, configfile) -> None:
|
||||
self.open(configfile)
|
||||
pass
|
||||
|
||||
|
||||
class Release(metaclass=Singleton):
|
||||
def __init__(self) -> None:
|
||||
f = open("release.json")
|
||||
release = json.load(f)
|
||||
self.version = release["version"]
|
||||
|
||||
class HyperDeckStatus(Enum):
|
||||
IDLE = "idle"
|
||||
RECORDING = "recording"
|
||||
UNKNOWN = "unknown"
|
||||
ERROR ="error"
|
||||
|
||||
class HyperDeck(metaclass=Singleton):
|
||||
|
||||
def load_config(self, configfile:str ):
|
||||
|
||||
|
||||
self.config = Config(configfile)
|
||||
if not os.path.isdir(self.config.videopath):
|
||||
raise Exception("Directory {self.config.videopath} does not exist")
|
||||
|
||||
if not os.path.isdir(self.config.videopath_originals):
|
||||
logging.info(
|
||||
"Creating directory {}".format(self.config.videopath_originals)
|
||||
)
|
||||
os.mkdir(self.config.videopath_originals)
|
||||
|
||||
if not os.path.isdir(self.config.videopath_originals):
|
||||
raise Exception(
|
||||
"Directory {} does not exist".format(
|
||||
self.config.videopath_originals
|
||||
)
|
||||
)
|
||||
|
||||
if not os.path.isdir(self.config.videopath_convert):
|
||||
logging.info("Creating directory {}".format(self.config.videopath_convert))
|
||||
os.mkdir(self.config.videopath_convert)
|
||||
|
||||
if not os.path.isdir(self.config.videopath_convert):
|
||||
raise Exception(
|
||||
"Directory {} does not exist".format(self.config.videopath_convert)
|
||||
)
|
||||
|
||||
self.status = HyperDeckStatus.UNKNOWN
|
||||
|
||||
self.videofiles = []
|
||||
self.current = None
|
||||
self.lasterror = ""
|
||||
self.schedule = Schedule()
|
||||
|
||||
self.processSchedule()
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.config = None
|
||||
os.environ["TZ"] = "Europe/Amsterdam"
|
||||
time.tzset()
|
||||
self.release = Release()
|
||||
|
||||
def getConvertedFilename(self, filename:str):
|
||||
return os.path.join(self.config.videopath_convert, 'c_' + filename)
|
||||
|
||||
def getTargetFilename(self, filename:str):
|
||||
return os.path.join(self.config.videopath, 'c_' + filename)
|
||||
|
||||
|
||||
def getStatus(self):
|
||||
return self.status
|
||||
|
||||
def cleanup(self):
|
||||
|
||||
|
||||
logging.info(f"Ckeaning up originals older than {self.config.max_originals_age} days from {self.config.videopath_originals}")
|
||||
maxdate = datetime.datetime.now() - datetime.timedelta(days=self.config.max_originals_age)
|
||||
for f in os.listdir(self.config.videopath_originals):
|
||||
filename = os.path.join(self.config.videopath_originals,f)
|
||||
if os.path.getmtime(filename) < maxdate.timestamp():
|
||||
logging.info(f"Deleting file {filename}")
|
||||
os.remove(filename)
|
||||
|
||||
|
||||
logging.info(f"Doing cleanup for items older than {self.config.max_age} days")
|
||||
items = self.schedule.getItemsOlderThan(self.config.max_age)
|
||||
for item in items:
|
||||
logging.info(f"Cleaning {item.id}")
|
||||
f = self.getConvertedFilename(item.file)
|
||||
if os.path.exists(f):
|
||||
logging.info(f"Removing file {f}")
|
||||
os.remove(f)
|
||||
|
||||
f = self.getTargetFilename(item.file)
|
||||
if os.path.exists(f):
|
||||
logging.info(f"Removing file {f}")
|
||||
os.remove(f)
|
||||
|
||||
self.schedule.removeItem(item)
|
||||
|
||||
|
||||
logging.info("Cleanup done")
|
||||
|
||||
|
||||
def sendData(self, req):
|
||||
try:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.settimeout(5)
|
||||
s.connect((self.config.hyperdeck_ip, self.config.hyperdeck_port))
|
||||
logging.info(f"Connected to {self.config.hyperdeck_ip}")
|
||||
|
||||
data = s.recv(1024).decode()
|
||||
|
||||
logging.info("Got data from Hyperdeck: " + data)
|
||||
|
||||
s.sendall(req.encode())
|
||||
|
||||
data = s.recv(1024).decode()
|
||||
|
||||
logging.info("Got data from Hyperdeck: " + data)
|
||||
|
||||
s.shutdown(socket.SHUT_RDWR)
|
||||
s.close()
|
||||
|
||||
if data[:3] == "200":
|
||||
self.lasterror = ""
|
||||
return True
|
||||
else:
|
||||
self.status = HyperDeckStatus.ERROR
|
||||
self.lasterror = data[3:]
|
||||
return False
|
||||
except TimeoutError:
|
||||
logging.error("Kon geen verbinding maken met hyperdeck")
|
||||
return False
|
||||
except Exception as e:
|
||||
logging.error("Got data from Hyperdeck: " + str(e))
|
||||
return False
|
||||
|
||||
def getLastError(self):
|
||||
return self.lasterror
|
||||
|
||||
def startRecording(self):
|
||||
if self.status == ScheduleItemStatus.RECORDING:
|
||||
logging.error("Already recording")
|
||||
return False
|
||||
|
||||
n: ScheduleItem = ScheduleItem.create(datetime.datetime.now().replace(microsecond=0), None, "recording")
|
||||
|
||||
n.status = ScheduleItemStatus.RECORDING
|
||||
filename = "vrm_" + n.start.strftime("%Y%m%d%H%M%S")
|
||||
n.file = filename
|
||||
|
||||
|
||||
item = n.to_schema()
|
||||
self.schedule.saveItem(n)
|
||||
|
||||
if self.startRecordingFile(item.file):
|
||||
|
||||
self.current = item
|
||||
return True
|
||||
else:
|
||||
n = self.schedule.getScheduleItem(item.id)
|
||||
n.status = ScheduleItemStatus.FAILED
|
||||
|
||||
self.schedule.saveItem(n)
|
||||
return False
|
||||
|
||||
|
||||
def startRecordingFile(self, filename: str):
|
||||
if self.status == HyperDeckStatus.RECORDING:
|
||||
logging.error("Already recording")
|
||||
return False
|
||||
|
||||
logging.info("Start recording")
|
||||
|
||||
if self.sendData("record: name: " + filename + "\n"):
|
||||
self.status = HyperDeckStatus.RECORDING
|
||||
return True
|
||||
else:
|
||||
self.status = HyperDeckStatus.ERROR
|
||||
return False
|
||||
|
||||
|
||||
|
||||
def stopRecording(self):
|
||||
if not self.current:
|
||||
logging.error("No current recording")
|
||||
return False
|
||||
|
||||
if self.status == ScheduleItemStatus.IDLE:
|
||||
logging.warning("Unable to stop recording: nothing recording")
|
||||
self.current.status = ScheduleItemStatus.FAILED
|
||||
return False
|
||||
|
||||
logging.info("Stop recording")
|
||||
|
||||
if self.sendData("stop\n"):
|
||||
item = self.schedule.getScheduleItem(self.current.id)
|
||||
|
||||
item.status = ScheduleItemStatus.RECORDED
|
||||
self.schedule.saveItem(item)
|
||||
self.current = None
|
||||
self.status = HyperDeckStatus.IDLE
|
||||
return True
|
||||
else:
|
||||
self.status = HyperDeckStatus.ERROR
|
||||
return False
|
||||
|
||||
|
||||
|
||||
def getMachines(self):
|
||||
return self.config.machines
|
||||
|
||||
def loadVideos(self):
|
||||
# if len(self.videofiles)==0 :
|
||||
newlist = []
|
||||
|
||||
for file in os.listdir(self.config.videopath):
|
||||
if file.startswith("c_") and file.endswith(".mp4"):
|
||||
filename = os.path.join(self.config.videopath, file)
|
||||
file_stats = os.stat(filename)
|
||||
|
||||
item = {
|
||||
"name": file,
|
||||
"path": filename,
|
||||
"type": "mp4",
|
||||
"filename": file,
|
||||
"filesize": file_stats.st_size,
|
||||
"start": None,
|
||||
"end": None,
|
||||
"title": None,
|
||||
}
|
||||
|
||||
rec = self.schedule.getScheduleItemByFile(item["name"])
|
||||
|
||||
if rec:
|
||||
item["start"] = rec.start
|
||||
item["end"] = rec.end
|
||||
item["title"] = rec.title
|
||||
|
||||
newlist.append(item)
|
||||
|
||||
self.videofiles = sorted(newlist, key=lambda q: q["name"].lower())
|
||||
|
||||
def getVideo(self, name):
|
||||
self.loadVideos()
|
||||
|
||||
limited_list = [
|
||||
element for element in self.videofiles if element["name"] == name
|
||||
]
|
||||
return limited_list[0]
|
||||
|
||||
def getVideos(self):
|
||||
self.loadVideos()
|
||||
|
||||
return self.videofiles
|
||||
|
||||
"""
|
||||
|
||||
"""
|
||||
|
||||
def removeScheduledItem(self, itm: ScheduleItem):
|
||||
self.schedule.remove(itm)
|
||||
self.saveSchedule()
|
||||
|
||||
def removeVideo(self, file: str):
|
||||
filename = os.path.join(self.config.videopath, file)
|
||||
logging.info("Removing file {}".format(filename))
|
||||
os.remove(filename)
|
||||
|
||||
itm = self.schedule.getScheduleItemByFile(file)
|
||||
if itm:
|
||||
self.removeScheduledItem(itm)
|
||||
|
||||
def isFree(self, start: datetime, end: datetime, exclude_id: str) -> bool:
|
||||
return self.schedule.getFree(start, end, exclude_id)
|
||||
|
||||
|
||||
def addScheduledRecording(self, start: datetime, end: datetime, title: str):
|
||||
n = ScheduleItem.create(start, end, title)
|
||||
s = self.schedule.getScheduleItem(n.id)
|
||||
if s:
|
||||
raise Exception("Schedule item already exists")
|
||||
|
||||
self.schedule.saveItem(n)
|
||||
|
||||
|
||||
|
||||
|
||||
def getSchedule(self):
|
||||
retval = []
|
||||
for s in self.schedule.getAll():
|
||||
j = {
|
||||
"id": s.id,
|
||||
"start": s.start,
|
||||
"status": s.status,
|
||||
"end": s.end,
|
||||
"file": None,
|
||||
"start_recording": s.start_recording,
|
||||
"end_recording": s.end_recording,
|
||||
"start_copying": s.start_copying,
|
||||
"end_copying": s.end_copying,
|
||||
"start_processing": s.start_processing,
|
||||
"end_processing": s.end_processing,
|
||||
"title": s.title,
|
||||
}
|
||||
|
||||
if not j["end"]:
|
||||
j["end"] = j["start"]
|
||||
|
||||
if j["title"]:
|
||||
j["display"] = (
|
||||
j["title"]
|
||||
+ ": "
|
||||
+ s.start.strftime("%H:%M")
|
||||
+ " - "
|
||||
+ j["end"].strftime("%H:%M")
|
||||
)
|
||||
else:
|
||||
j["display"] = (
|
||||
s.start.strftime("%H:%M") + " - " + j["end"].strftime("%H:%M")
|
||||
)
|
||||
|
||||
if s.status == ScheduleItemStatus.DONE:
|
||||
j["file"] = "c_" + s.file + ".mp4"
|
||||
retval.append(j)
|
||||
return retval
|
||||
|
||||
def getNextScheduled(self):
|
||||
|
||||
return self.schedule.getNextScheduled()
|
||||
|
||||
def hasProcessing(self) -> bool:
|
||||
return self.getNextProcessing is None
|
||||
|
||||
"""
|
||||
def markCopied(self, filename):
|
||||
s = self.getByFile(filename)
|
||||
if s and s.status == ScheduleItemStatus.COPYING:
|
||||
s.status = ScheduleItemStatus.COPIED
|
||||
self.saveSchedule()
|
||||
"""
|
||||
|
||||
def processSchedule(self):
|
||||
for s in self.schedule.getItemsByStatus([ScheduleItemStatus.WAITING, ScheduleItemStatus.RECORDING]):
|
||||
if s.status == ScheduleItemStatus.WAITING:
|
||||
if s.end < datetime.datetime.now():
|
||||
id = s.id
|
||||
s.status = ScheduleItemStatus.FAILED
|
||||
self.schedule.saveItem(s)
|
||||
s = self.schedule.getScheduleItem(id)
|
||||
|
||||
if s.start < datetime.datetime.now():
|
||||
if self.startRecordingFile(s.file):
|
||||
s.status = ScheduleItemStatus.RECORDING
|
||||
|
||||
logging.info("Start recording " + s.file)
|
||||
s.start_recording = datetime.datetime.now().replace(
|
||||
microsecond=0
|
||||
)
|
||||
else:
|
||||
s.status = ScheduleItemStatus.FAILED
|
||||
logging.error("Unable to start recording")
|
||||
self.schedule.saveItem(s)
|
||||
|
||||
elif s.status ==ScheduleItemStatus.RECORDING:
|
||||
self.current = s
|
||||
logging.info(s.end)
|
||||
if s.end and (s.end < datetime.datetime.now()):
|
||||
logging.info("End of schedule for " + s.file)
|
||||
if not self.stopRecording():
|
||||
logging.error("Unable to stop recording")
|
||||
else:
|
||||
s.end_recording = datetime.datetime.now().replace(microsecond=0)
|
||||
self.schedule.saveItem(s)
|
||||
|
||||
elif s.end is None:
|
||||
self.status = HyperDeckStatus.RECORDING
|
||||
|
||||
def getWorksheet(self):
|
||||
return self.worksheet
|
||||
+294
@@ -0,0 +1,294 @@
|
||||
from datetime import datetime, timedelta
|
||||
import enum
|
||||
import logging
|
||||
import os
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
from sqlmodel import Enum,Field, Column, SQLModel, Session, create_engine, not_, or_, select
|
||||
|
||||
|
||||
def enum_values(enum_class: type[enum.Enum]) -> list:
|
||||
"""Get values for enum."""
|
||||
return [status.value for status in enum_class]
|
||||
|
||||
|
||||
class ScheduleItemStatus(str, enum.Enum):
|
||||
NEW="new"
|
||||
WAITING="waiting"
|
||||
IDLE="idle"
|
||||
RECORDING="recording"
|
||||
RECORDED="recorded"
|
||||
COPYING="copying"
|
||||
COPIED="copied"
|
||||
PROCESSING="processing"
|
||||
FAILED="failed"
|
||||
DONE="done"
|
||||
|
||||
class ScheduleItemSchema(BaseModel):
|
||||
id: Optional[str] = None
|
||||
status: ScheduleItemStatus
|
||||
file: str = ""
|
||||
title: str = ""
|
||||
|
||||
# Tijden
|
||||
start: Optional[datetime] = None
|
||||
end: Optional[datetime] = None
|
||||
|
||||
# Proces tijden
|
||||
start_recording: Optional[datetime] = None
|
||||
end_recording: Optional[datetime] = None
|
||||
start_copying: Optional[datetime] = None
|
||||
end_copying: Optional[datetime] = None
|
||||
start_processing: Optional[datetime] = None
|
||||
end_processing: Optional[datetime] = None
|
||||
|
||||
last_update: Optional[datetime] = None
|
||||
|
||||
class Config:
|
||||
# In v1 moet dit 'orm_mode' zijn in plaats van 'from_attributes'
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class ScheduleItem(SQLModel, table=True):
|
||||
status:ScheduleItemStatus = Field(
|
||||
sa_column=Column(
|
||||
name="status",
|
||||
nullable=False,
|
||||
type_= Enum(ScheduleItemStatus, values_callable=enum_values),
|
||||
)
|
||||
)
|
||||
file:str = ""
|
||||
start:Optional[datetime] = None
|
||||
end:Optional[datetime] = None
|
||||
id: str | None = Field(default=None, primary_key=True)
|
||||
start_recording:datetime = None
|
||||
end_recording:datetime = None
|
||||
start_copying:datetime = None
|
||||
end_copying:datetime = None
|
||||
start_processing:datetime = None
|
||||
end_processing:datetime = None
|
||||
title:str = ""
|
||||
last_update:datetime = None
|
||||
|
||||
|
||||
def to_schema(self) -> ScheduleItemSchema:
|
||||
"""
|
||||
Converteert het database-object naar een sessie-loos Pydantic model.
|
||||
Dit voorkomt DetachedInstanceErrors buiten de database scope.
|
||||
"""
|
||||
return ScheduleItemSchema.from_orm(self)
|
||||
|
||||
# Je kunt dit ook integreren in je assign of create logica
|
||||
@classmethod
|
||||
def from_schema(cls, schema: ScheduleItemSchema):
|
||||
"""Maakt een database-object aan op basis van een schema."""
|
||||
return cls(**schema.dict())
|
||||
|
||||
def assign(self, item):
|
||||
self.status = item.status
|
||||
self.title = item.title
|
||||
self.end_processing = item.end_processing
|
||||
self.end_copying = item.end_copying
|
||||
self.end_recording = item.end_recording
|
||||
self.last_update = datetime.now()
|
||||
|
||||
def create(start: datetime, end: datetime, title: str) :
|
||||
retval = ScheduleItem()
|
||||
retval.start = start.replace(tzinfo=None)
|
||||
if end:
|
||||
retval.end = end.replace(tzinfo=None)
|
||||
if retval.end <= retval.start:
|
||||
raise Exception("End time must be later then start time")
|
||||
retval.last_update = datetime.now()
|
||||
|
||||
|
||||
retval.id = start.strftime("R%Y%m%d%H%M%S")
|
||||
retval.title = title
|
||||
|
||||
retval.status = ScheduleItemStatus.WAITING
|
||||
retval.file = "vr_" + start.strftime("%Y%m%d%H%M%S")
|
||||
return retval
|
||||
|
||||
def setCopying(self):
|
||||
self.status = ScheduleItemStatus.COPYING
|
||||
self.start_copying = datetime.now().replace(microsecond=0)
|
||||
|
||||
def setProcessing(self):
|
||||
self.status = ScheduleItemStatus.PROCESSING
|
||||
self.start_processing = datetime.now().replace(microsecond=0)
|
||||
|
||||
def fromJson(j):
|
||||
retval = ScheduleItem()
|
||||
retval.start = datetime.fromisoformat(j["start"])
|
||||
|
||||
if "title" in j:
|
||||
retval.title = j["title"]
|
||||
else:
|
||||
retval.title = None
|
||||
|
||||
if "end" in j:
|
||||
retval.end = datetime.fromisoformat(j["end"])
|
||||
else:
|
||||
retval.end = None
|
||||
|
||||
if "start_recording" in j and j["start_recording"]:
|
||||
retval.start_recording = datetime.fromisoformat(
|
||||
j["start_recording"]
|
||||
)
|
||||
if "end_recording" in j and j["end_recording"]:
|
||||
retval.end_recording = datetime.fromisoformat(j["end_recording"])
|
||||
|
||||
if "start_copying" in j and j["start_copying"]:
|
||||
retval.start_copying = datetime.fromisoformat(j["start_copying"])
|
||||
if "end_copying" in j and j["end_copying"]:
|
||||
retval.end_copying = datetime.fromisoformat(j["end_copying"])
|
||||
|
||||
if "start_processing" in j and j["start_processing"]:
|
||||
retval.start_processing = datetime.fromisoformat(
|
||||
j["start_processing"]
|
||||
)
|
||||
if "end_processing" in j and j["end_processing"]:
|
||||
retval.end_processing = datetime.fromisoformat(j["end_processing"])
|
||||
|
||||
retval.id = retval.start.strftime("R%Y%m%d%H%M%S")
|
||||
retval.status = j["status"]
|
||||
retval.file = j["file"]
|
||||
return retval
|
||||
|
||||
class Schedule:
|
||||
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def saveItems(self, items: List[ScheduleItem]):
|
||||
with Session(engine) as session:
|
||||
for item in items:
|
||||
item.last_update = datetime.now()
|
||||
session.add(item)
|
||||
session.commit()
|
||||
|
||||
|
||||
|
||||
def saveItem(self, item: ScheduleItem):
|
||||
with Session(engine) as session:
|
||||
item.last_update = datetime.now()
|
||||
session.add(item)
|
||||
session.commit()
|
||||
|
||||
def removeItem(self, item: ScheduleItem):
|
||||
with Session(engine) as session:
|
||||
session.delete(item)
|
||||
session.commit()
|
||||
|
||||
def getItemsOlderThan(self, age:int) -> List[ScheduleItem]:
|
||||
periode = timedelta(days=age)
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.start < datetime.now()-periode)
|
||||
return session.exec(statement).all()
|
||||
|
||||
def getFree(self, start: datetime, end: datetime, exclude_id: str) -> bool:
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(
|
||||
not_(or_(
|
||||
ScheduleItem.id == exclude_id,
|
||||
ScheduleItem.status != ScheduleItemStatus.WAITING,
|
||||
ScheduleItem.start >= end,
|
||||
not ScheduleItem.end,
|
||||
ScheduleItem.end <= start
|
||||
|
||||
)))
|
||||
return session.exec(statement).first() is None
|
||||
|
||||
def getNextScheduled(self):
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.start > datetime.now()).order_by(ScheduleItem.start)
|
||||
|
||||
session.exec(statement).first()
|
||||
|
||||
|
||||
def getItemsLastDay(self):
|
||||
time = datetime.now()-timedelta(days=1)
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(
|
||||
or_(
|
||||
ScheduleItem.start_copying > time,
|
||||
ScheduleItem.start_processing > time,
|
||||
ScheduleItem.start_recording > time,
|
||||
ScheduleItem.last_update > time
|
||||
)).order_by(ScheduleItem.start
|
||||
)
|
||||
return session.exec(statement).all()
|
||||
|
||||
|
||||
def getAll(self):
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).order_by(ScheduleItem.start)
|
||||
return session.exec(statement).all()
|
||||
|
||||
def getActiveItems(self):
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(
|
||||
ScheduleItem.status.in_([ScheduleItemStatus.PROCESSING,ScheduleItemStatus.COPYING])
|
||||
)
|
||||
return session.exec(statement).all()
|
||||
|
||||
def getItemsByStatus(self, statusses: List[ScheduleItem]):
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.status.in_( statusses))
|
||||
return session.exec(statement).all()
|
||||
|
||||
def returnFirstByStatus(self, status: ScheduleItemStatus) -> ScheduleItem|None:
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.status == status).order_by(ScheduleItem.start)
|
||||
return session.exec(statement).first()
|
||||
|
||||
|
||||
def getNextCopying(self) -> ScheduleItem:
|
||||
|
||||
item = self.returnFirstByStatus(ScheduleItemStatus.RECORDED)
|
||||
if item:
|
||||
id = item.id
|
||||
item.setCopying()
|
||||
self.saveItem(item)
|
||||
item = self.getScheduleItem(id)
|
||||
return item
|
||||
|
||||
|
||||
def getNextProcessing(self) -> ScheduleItem:
|
||||
item = self.returnFirstByStatus(ScheduleItemStatus.COPIED)
|
||||
if item:
|
||||
id = item.id
|
||||
item.setProcessing()
|
||||
self.saveItem(item)
|
||||
item = self.getScheduleItem(id)
|
||||
return item
|
||||
|
||||
def getByFile(self, filename):
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.file == filename)
|
||||
return session.exec(statement).first()
|
||||
|
||||
|
||||
def getScheduleItemByFile(self, filename) -> ScheduleItem:
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where("c_" +ScheduleItem.file+ ".mp4" == filename)
|
||||
return session.exec(statement).first()
|
||||
|
||||
def getScheduleItem(self, id:str) -> ScheduleItem:
|
||||
with Session(engine) as session:
|
||||
statement = select(ScheduleItem).where(ScheduleItem.id == id)
|
||||
return session.exec(statement).first()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
sqlite_file_name =os.path.join(os.getcwd(), "hyperdeck.db")
|
||||
sqlite_url = f"sqlite:///{sqlite_file_name}"
|
||||
|
||||
engine = create_engine(sqlite_url, echo=True)
|
||||
|
||||
def create_db_and_tables():
|
||||
SQLModel.metadata.create_all(engine)
|
||||
+75
@@ -0,0 +1,75 @@
|
||||
from ftplib import FTP
|
||||
from hyperdeck.application import HyperDeck
|
||||
import logging,os ,re
|
||||
|
||||
|
||||
logger = logging.getLogger('tasklogger')
|
||||
|
||||
def downloadFiles(name) -> bool:
|
||||
|
||||
app = HyperDeck()
|
||||
|
||||
logger.info("Download file " + name)
|
||||
|
||||
with FTP(host=app.config.hyperdeck_ip, user='anonymous', passwd='secret') as ftp:
|
||||
logger.info("Connected to FTP")
|
||||
|
||||
logger.debug(ftp.getwelcome())
|
||||
files = ftp.nlst('/1')
|
||||
logger.info("{} files found".format(len(files)))
|
||||
|
||||
cnt = 0
|
||||
found = False
|
||||
for f in files:
|
||||
if f.startswith( name) and f.endswith('.mp4'):
|
||||
if re.search("_[0-9]{4}\.mp4$",f):
|
||||
nr = f[-8:-4]
|
||||
else:
|
||||
nr = str(cnt).zfill(4)
|
||||
|
||||
targetfile = os.path.join(app.config.videopath_originals, nr + '_' + name + ".mp4")
|
||||
logger.info("Retrieving file " + f + ' to '+targetfile)
|
||||
|
||||
if os.path.isfile(targetfile):
|
||||
logger.error("Target file already exists: " + targetfile)
|
||||
return False
|
||||
|
||||
with open(targetfile, 'wb') as fp:
|
||||
ftp.retrbinary('RETR /1/' + f, fp.write)
|
||||
ftp.delete('/1/' + f)
|
||||
found = True
|
||||
|
||||
cnt += 1
|
||||
|
||||
files = ftp.nlst('/2')
|
||||
for f in files:
|
||||
if f.startswith( name) and f.endswith('.mp4'):
|
||||
if re.search("_[0-9]{4}\.mp4$",f):
|
||||
nr = f[-8:-4]
|
||||
else:
|
||||
nr = str(cnt).zfill(4)
|
||||
|
||||
targetfile = os.path.join(app.config.videopath_originals, nr + '_' + name + ".mp4")
|
||||
logger.info("Retrieving file " + f + ' to '+targetfile)
|
||||
|
||||
if os.path.isfile(targetfile):
|
||||
logger.error("Target file already exists: " + targetfile)
|
||||
return False
|
||||
|
||||
with open(targetfile, 'wb') as fp:
|
||||
ftp.retrbinary('RETR /2/' + f, fp.write)
|
||||
ftp.delete('/2/' + f)
|
||||
found = True
|
||||
|
||||
cnt += 1
|
||||
|
||||
|
||||
if found:
|
||||
return True
|
||||
|
||||
logger.error("File not found: " + name)
|
||||
return False
|
||||
|
||||
logger.error("Unable to connect to FTP")
|
||||
return False
|
||||
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
|
||||
|
||||
from json import JSONEncoder
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
||||
from hyperdeck.db import ScheduleItem
|
||||
|
||||
|
||||
|
||||
|
||||
class ScheduleItemEncoder(JSONEncoder):
|
||||
def default(self, o):
|
||||
if isinstance(o, ScheduleItem):
|
||||
return o.toJson()
|
||||
|
||||
return json.JSONEncoder.default(self, o)
|
||||
|
||||
|
||||
def ScheduleItemDecoder(j):
|
||||
print(j)
|
||||
return ScheduleItem.fromJson(j)
|
||||
|
||||
class jsonconverter:
|
||||
def loadschedule(self):
|
||||
try:
|
||||
with open(os.path.join(os.getcwd(), "schedule.json"), "r") as f:
|
||||
self.schedule = json.load(f, object_hook=ScheduleItemDecoder)
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logging.error("Unable to load schedule" + repr(e))
|
||||
self.schedule = []
|
||||
raise e
|
||||
|
||||
|
||||
def saveSchedule(self):
|
||||
with open(os.path.join(os.getcwd(), "schedule.json"), "w") as f:
|
||||
f.write(json.dumps(self.schedule, cls=ScheduleItemEncoder))
|
||||
+148
@@ -0,0 +1,148 @@
|
||||
import logging
|
||||
|
||||
from rocketry import Rocketry
|
||||
from rocketry.conds import every
|
||||
|
||||
from hyperdeck.application import HyperDeck
|
||||
import datetime
|
||||
from hyperdeck.db import ScheduleItemStatus
|
||||
from hyperdeck.download import downloadFiles
|
||||
from hyperdeck.transcode import transcodeFile
|
||||
|
||||
import os
|
||||
|
||||
import multiprocessing
|
||||
|
||||
|
||||
from rocketry.conds import (
|
||||
minutely, daily
|
||||
)
|
||||
|
||||
manager = multiprocessing.Manager()
|
||||
|
||||
copyqueue = manager.Queue()
|
||||
copyqueue_done = manager.Queue()
|
||||
processqueue= manager.Queue()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
application = HyperDeck()
|
||||
# Creating the Rocketry app
|
||||
app = Rocketry(config={"task_execution": "async"})
|
||||
|
||||
formatter = logging.Formatter('[%(asctime)s] %(message)s')
|
||||
|
||||
handler =logging.FileHandler(os.path.join(application.config.log_path, "scheduler.log"))
|
||||
handler.setFormatter(formatter)
|
||||
tasklogger = logging.getLogger("rocketry.task")
|
||||
tasklogger.setLevel(logging.INFO)
|
||||
tasklogger.addHandler(handler)
|
||||
|
||||
|
||||
logger = logging.getLogger('tasklogger')
|
||||
|
||||
handler =logging.FileHandler(os.path.join(application.config.log_path, "task.log") )
|
||||
|
||||
handler.setFormatter(formatter)
|
||||
logger.setLevel(logging.INFO)
|
||||
logger.addHandler(handler)
|
||||
|
||||
# Creating some tasks
|
||||
|
||||
logger.info("Startup, checking for in progress items")
|
||||
for s in application.schedule.getActiveItems():
|
||||
if s.status == ScheduleItemStatus.COPYING:
|
||||
s.start_copying = datetime.datetime.now().replace(microsecond=0)
|
||||
logger.info("File " + s.file + " marked for copying, queuing")
|
||||
copyqueue.put(s.to_schema())
|
||||
|
||||
if s.status == ScheduleItemStatus.PROCESSING:
|
||||
s.start_processing = datetime.datetime.now().replace(microsecond=0)
|
||||
logger.info("File " + s.file + " marked for processing, queuing")
|
||||
processqueue.put(s.to_schema())
|
||||
|
||||
|
||||
@app.task(minutely)
|
||||
def do_checkschedule():
|
||||
application.processSchedule()
|
||||
|
||||
|
||||
@app.task(every('10 seconds'))
|
||||
def get_NewFileName():
|
||||
|
||||
|
||||
if not copyqueue_done.empty():
|
||||
item = copyqueue_done.get()
|
||||
|
||||
logger.info("Process queue output for " + item.file)
|
||||
itemdb=application.schedule.getScheduleItem(item.id)
|
||||
itemdb.assign(item)
|
||||
application.schedule.saveItem(itemdb)
|
||||
|
||||
if copyqueue.empty():
|
||||
currentcopy = application.schedule.getNextCopying()
|
||||
if currentcopy:
|
||||
logger.info("Add to copy queue " + currentcopy.file)
|
||||
copyqueue.put(currentcopy.to_schema())
|
||||
|
||||
if processqueue.empty():
|
||||
currentcopy = application.schedule.getNextProcessing()
|
||||
if currentcopy:
|
||||
logger.info("Add to process queue " + currentcopy.file)
|
||||
processqueue.put(currentcopy.to_schema())
|
||||
|
||||
@app.task(minutely,execution="process",timeout=60*60*24)
|
||||
def do_copyfile(q=copyqueue,qd=copyqueue_done ):
|
||||
|
||||
if q.empty():
|
||||
return
|
||||
|
||||
item = copyqueue.get()
|
||||
logger.info("Start copying " + item.file)
|
||||
|
||||
if not downloadFiles(item.file):
|
||||
item.status = ScheduleItemStatus.FAILED
|
||||
logger.info("Copying failed for " + item.file)
|
||||
|
||||
else:
|
||||
item.status = ScheduleItemStatus.COPIED
|
||||
item.end_copying = datetime.datetime.now().replace(microsecond=0)
|
||||
logger.info("Copying success for " + item.file)
|
||||
|
||||
qd.put(item)
|
||||
|
||||
return
|
||||
|
||||
@app.task(daily,execution="process",timeout=60*60*24)
|
||||
def do_cleanup():
|
||||
application.cleanup()
|
||||
|
||||
@app.task(minutely,execution="process",timeout=60*60*24)
|
||||
def do_processfile(q=processqueue,qd=copyqueue_done ):
|
||||
|
||||
if q.empty():
|
||||
return
|
||||
|
||||
item = processqueue.get()
|
||||
logger.info("Start processing " + item.file)
|
||||
|
||||
|
||||
|
||||
if not transcodeFile(item.file):
|
||||
item.status = ScheduleItemStatus.FAILED
|
||||
else:
|
||||
item.status = ScheduleItemStatus.DONE
|
||||
item.end_processing = datetime.datetime.now().replace(microsecond=0)
|
||||
|
||||
qd.put(item)
|
||||
|
||||
return
|
||||
|
||||
#time.sleep(5)
|
||||
#raise TaskTerminationException()
|
||||
|
||||
if __name__ == "__main__":
|
||||
# If this script is run, only Rocketry is run
|
||||
app.run()
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
import os
|
||||
from hyperdeck.application import HyperDeck
|
||||
import logging
|
||||
|
||||
|
||||
|
||||
logger = logging.getLogger('tasklogger')
|
||||
|
||||
def transcodeFile(name) -> bool:
|
||||
|
||||
app = HyperDeck()
|
||||
|
||||
filename= name + ".mp4"
|
||||
|
||||
tofind = os.path.join(app.config.videopath_originals, '*' + name + ".mp4")
|
||||
|
||||
convertedfile = os.path.join(app.config.videopath_convert, 'c_' + filename)
|
||||
targetfile = os.path.join(app.config.videopath, 'c_' + filename)
|
||||
|
||||
r= os.system('for f in ' + tofind + '; do echo "file \'$f\'" ; done > /tmp/processlist.txt')
|
||||
|
||||
logger.info("Transcode file " + filename)
|
||||
|
||||
logfile = app.config.log_path + name +'.log'
|
||||
r= os.system('/usr/bin/ffmpeg -hide_banner -loglevel warning -nostats -f concat -safe 0 -i /tmp/processlist.txt -y -c:v libx264 -preset medium -b:v 3000k -maxrate 3000k -bufsize 6000k -vf "scale=1280:-1,format=yuv420p" -g 50 -c:a aac -b:a 128k -ac 2 -ar 44100 -threads 2 ' + convertedfile + ' > ' + logfile)
|
||||
logger.info("Converting done for file " + filename)
|
||||
|
||||
|
||||
if not os.path.isfile(convertedfile):
|
||||
logger.error("Converted file {} not found, transcode failed".format(convertedfile))
|
||||
return False
|
||||
os.rename(convertedfile , targetfile)
|
||||
|
||||
return r==0
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
|
||||
class Singleton(type):
|
||||
_instances = {}
|
||||
def __call__(cls, *args, **kwargs):
|
||||
if cls not in cls._instances:
|
||||
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
|
||||
return cls._instances[cls]
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import getopt
|
||||
from hyperdeck.application import HyperDeck
|
||||
import asyncio
|
||||
import uvicorn
|
||||
|
||||
|
||||
from hyperdeck.db import create_db_and_tables
|
||||
from hyperdeck.jsondb import jsonconverter
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
opts, args = getopt.getopt(sys.argv[1:], "c:o:")
|
||||
configfile = "config.json"
|
||||
for opt, arg in opts:
|
||||
if opt == "-h":
|
||||
print("test.py -c <config>")
|
||||
sys.exit()
|
||||
elif opt in ("-c"):
|
||||
configfile = arg
|
||||
|
||||
create_db_and_tables()
|
||||
|
||||
hApp = HyperDeck()
|
||||
hApp.load_config(configfile)
|
||||
|
||||
if os.path.exists("schedule.json"):
|
||||
logging.info("Found schedule.json, loading")
|
||||
conv = jsonconverter()
|
||||
conv.loadschedule()
|
||||
logging.info("Loading done, creating database")
|
||||
|
||||
hApp.schedule.saveItems(conv.schedule)
|
||||
|
||||
logging.info("Database ready, moving file")
|
||||
os.rename("schedule.json","schedule.json.bak")
|
||||
|
||||
|
||||
from hyperdeck.api import app as app_fastapi
|
||||
from hyperdeck.scheduler import app as app_rocketry
|
||||
|
||||
|
||||
|
||||
session = app_rocketry.session
|
||||
|
||||
|
||||
class Server(uvicorn.Server):
|
||||
def handle_exit(self, sig: int, frame) -> None:
|
||||
app_rocketry.session.shut_down()
|
||||
return super().handle_exit(sig, frame)
|
||||
|
||||
|
||||
async def app():
|
||||
server = Server(
|
||||
config=uvicorn.Config(
|
||||
app_fastapi, workers=1, host=["0.0.0.0", "::"], port=8180, loop="asyncio"
|
||||
)
|
||||
)
|
||||
|
||||
api = asyncio.create_task(server.serve())
|
||||
sched = asyncio.create_task(app_rocketry.serve())
|
||||
|
||||
await asyncio.wait([sched, api])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
asyncio.run(app())
|
||||
@@ -0,0 +1 @@
|
||||
{ "version" : "0.15.2" }
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
anyio==4.10.0
|
||||
click==8.1.3
|
||||
fastapi==0.116.1
|
||||
h11==0.16.0
|
||||
idna==3.10
|
||||
sniffio==1.3.0
|
||||
starlette==0.47.3
|
||||
typing_extensions==4.15.0
|
||||
pydantic==1.10.10
|
||||
uvicorn
|
||||
jinja2
|
||||
python-multipart
|
||||
requests
|
||||
rocketry==2.4.1
|
||||
opencv-python
|
||||
ffmpeg-python
|
||||
sqlmodel
|
||||
@@ -0,0 +1,24 @@
|
||||
#!/bin/bash
|
||||
|
||||
import sys, getopt,os
|
||||
from hyperdeck.application import Config, HyperDeck
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
|
||||
opts, args = getopt.getopt(sys.argv[1:],"c:o:")
|
||||
configfile="config.json"
|
||||
for opt, arg in opts:
|
||||
if opt == '-h':
|
||||
print ('test.py -c <config>')
|
||||
sys.exit()
|
||||
elif opt in ("-c"):
|
||||
configfile = arg
|
||||
|
||||
config = Config(configfile)
|
||||
|
||||
logging.info("Setting up connection to: " +config.rtsp_url )
|
||||
|
||||
|
||||
os.system('/usr/bin/ffmpeg -hide_banner -loglevel warning -nostats -re -rtsp_transport tcp -i '+ config.rtsp_url +' -c copy -s hd720 -f flv -y rtmp://localhost/live/livestream > /tmp/ffmpeglive.log')
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
from rocketry import Rocketry
|
||||
|
||||
app = Rocketry()
|
||||
|
||||
@app.task('daily')
|
||||
def do_things():
|
||||
print("dd")
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run()
|
||||
Reference in New Issue
Block a user