This repository has been archived on 2024-03-03. You can view files and clone it, but cannot push or open issues or pull requests.
unifi-cam-proxy/unifi/main.py

177 lines
4.5 KiB
Python
Raw Normal View History

2024-03-02 11:37:04 +00:00
import argparse
import asyncio
import logging
import sys
from shutil import which
import coloredlogs
from pyunifiprotect import ProtectApiClient
from unifi.cams import (
DahuaCam,
FrigateCam,
HikvisionCam,
Reolink,
ReolinkNVRCam,
RTSPCam,
)
from unifi.core import Core
from unifi.version import __version__
CAMS = {
"amcrest": DahuaCam,
"dahua": DahuaCam,
"frigate": FrigateCam,
"hikvision": HikvisionCam,
"lorex": DahuaCam,
"reolink": Reolink,
"reolink_nvr": ReolinkNVRCam,
"rtsp": RTSPCam,
}
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--version", action="version", version=__version__)
parser.add_argument("--host", "-H", required=True, help="NVR ip address and port")
parser.add_argument("--nvr-username", required=False, help="NVR username")
parser.add_argument("--nvr-password", required=False, help="NVR password")
parser.add_argument(
"--cert",
"-c",
required=True,
default="client.pem",
help="Client certificate path",
)
parser.add_argument(
"--token", "-t", required=False, default=None, help="Adoption token"
)
parser.add_argument("--mac", "-m", default="AABBCCDDEEFF", help="MAC address")
parser.add_argument(
"--ip",
"-i",
default="192.168.1.10",
help="IP address of camera (only used to display in UI)",
)
parser.add_argument(
"--name",
"-n",
default="unifi-cam-proxy",
help="Name of camera (only works for UFV)",
)
parser.add_argument(
"--model",
default="UVC G3",
choices=[
"UVC",
"UVC AI 360",
"UVC AI Bullet",
"UVC AI THETA",
"UVC AI DSLR",
"UVC Pro",
"UVC Dome",
"UVC Micro",
"UVC G3",
"UVC G3 Battery",
"UVC G3 Dome",
"UVC G3 Micro",
"UVC G3 Mini",
"UVC G3 Instant",
"UVC G3 Pro",
"UVC G3 Flex",
"UVC G4 Bullet",
"UVC G4 Pro",
"UVC G4 PTZ",
"UVC G4 Doorbell",
"UVC G4 Doorbell Pro",
"UVC G4 Doorbell Pro PoE",
"UVC G4 Dome",
"UVC G4 Instant",
"UVC G5 Bullet",
"UVC G5 Dome",
"UVC G5 Flex",
"UVC G5 Pro",
"AFi VC",
"Vision Pro",
],
help="Hardware model to identify as",
)
parser.add_argument(
"--fw-version",
"-f",
default="UVC.S2L.v4.23.8.67.0eba6e3.200526.1046",
help="Firmware version to initiate connection with",
)
parser.add_argument(
"--verbose", "-v", action="store_true", help="increase output verbosity"
)
sp = parser.add_subparsers(
help="Camera implementations",
dest="impl",
required=True,
)
for name, impl in CAMS.items():
subparser = sp.add_parser(name)
impl.add_parser(subparser)
return parser.parse_args()
async def generate_token(args, logger):
try:
protect = ProtectApiClient(
args.host, 443, args.nvr_username, args.nvr_password, verify_ssl=False
)
await protect.update()
response = await protect.api_request("cameras/manage-payload")
return response["mgmt"]["token"]
except Exception:
logger.exception(
"Could not automatically fetch token, please see docs at"
" https://unifi-cam-proxy.com/"
)
return None
finally:
await protect.close_session()
async def run():
args = parse_args()
klass = CAMS[args.impl]
core_logger = logging.getLogger("Core")
class_logger = logging.getLogger(klass.__name__)
level = logging.INFO
if args.verbose:
level = logging.DEBUG
for logger in [core_logger, class_logger]:
coloredlogs.install(level=level, logger=logger)
# Preflight checks
for binary in ["ffmpeg", "nc"]:
if which(binary) is None:
logger.error(f"{binary} is not installed")
sys.exit(1)
if not args.token:
args.token = await generate_token(args, logger)
if not args.token:
logger.error("A valid token is required")
sys.exit(1)
cam = klass(args, logger)
c = Core(args, cam, core_logger)
await c.run()
def main():
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
if __name__ == "__main__":
main()