177 lines
4.5 KiB
Python
177 lines
4.5 KiB
Python
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()
|