Skip to content
Snippets Groups Projects
Commit 3a80aa45 authored by Simon Künzel's avatar Simon Künzel
Browse files

Initial commit

parents
Branches main
No related tags found
No related merge requests found
.idea
.venv
config.py
__pycache__
import logging
from pathlib import Path
from time import sleep
from kiloview_e3_api import KiloviewE3Api
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s: %(message)s",
)
logger = logging.getLogger(__name__)
AUTOMATIC_RESTART = True
"""
Create config.py with following contents (Adjust IP, username and password):
BASE_URL = "http://192.168.0.100/api"
USERNAME = "..."
PASSWORD = "..."
"""
config = {}
exec(Path("config.py").read_text(), config)
BASE_URL = config["BASE_URL"]
USERNAME = config["USERNAME"]
PASSWORD = config["PASSWORD"]
def main():
api = KiloviewE3Api(BASE_URL)
api.login(USERNAME, PASSWORD)
logger.info("Logged in")
while True:
logger.info("Checking status...")
files = api.get_device_files("/run/media/sda1")
if len(files) == 0 or files[0]["end_time"] is not None:
logger.warning("NO FILE IS RECORDING!!!!!")
if AUTOMATIC_RESTART:
try:
logger.info("Restarting...")
api.set_do_hdmi_recording(False)
api.set_do_hdmi_recording(True)
logger.info("Restarted")
except Exception as e:
logger.error("Unable to restart recording", e)
else:
logger.info("Still running")
sleep(1)
if __name__ == "__main__":
main()
import json
import requests
JsonTypes = dict[str, "JsonTypes"] | list["JsonTypes"] | str | int | bool | None
class KiloviewE3Api:
def __init__(self, base_url: str):
self._base_url = base_url
self._token: str | None = None
def _do_request(self, method: str, path: str, body: JsonTypes | None = None, query_params: dict | None = None, require_login: bool = True) -> JsonTypes:
if not path.startswith("/"):
raise ValueError("Path must start with /")
headers = {}
if body is not None:
headers["Content-Type"] = "application/json"
if require_login:
if self._token is None:
raise Exception("Not logged in")
headers["Cookie"] = f"token={self._token}"
resp = requests.request(
method,
f"{self._base_url}{path}",
data=None if body is None else json.dumps(body),
headers=headers,
params=query_params,
)
if not resp.ok:
raise Exception(f"Request not ok. Got status: {resp.status_code}: {resp.content}")
resp_body = resp.json()
if not resp_body["result"]:
raise Exception(f"Request not ok: {resp_body['msg']}")
return resp_body["data"]
def login(self, username: str, password: str):
resp = self._do_request("POST", "/systemctrl/users/login", {
"username": username,
"password": password
}, require_login=False)
self._token = resp["token"]
def get_device_files(self, device: str) -> list:
return self._do_request("GET", "/record/record/get_disk_files", query_params={
"device": device
})
def set_do_hdmi_recording(self, do_recording: bool):
self._do_request("POST", "/record/hdmi/recording", body={
"start": do_recording
})
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment