All/Python/local/camera/wvp/wvp.py

from .lib import *
from time import time
from threading import Lock
from requests import session

__all__ = ["WvpProxy"]


class WvpProxy:
    def __init__(self):
        self.__sess = session()
        self.__nextExpire, self.__expireLock = time(), Lock()
        self.__login()

    def __isExpired(self) -> "bool":
        with self.__expireLock:
            return time() < self.__nextExpire

    def __login(self):
        resp = self.__sess.get(f"{WVP_API}/user/login?username={USER}&password={PWD_MD5}", headers=HEADER)
        assert resp.status_code == 200, f"Login Fail: status code is <{resp.status_code}>"
        assert (body := resp.json()) and body["code"] == 0, f"Login Fail: body code is <{body['code']}>"
        if resp.cookies:
            self.__sess.cookies = resp.cookies
        with self.__expireLock:
            self.__nextExpire = time() + ALIVE_TIME

    def start(self, device: "str", channel: "str") -> "str":
        if self.__isExpired():
            self.__login()
        resp = self.__sess.get(f"{WVP_API}/play/start/{device}/{channel}", headers=HEADER)
        assert resp.status_code == 200, f"Start Fail: status code is <{resp.status_code}>"
        body = resp.json()
        assert body["code"] == 0, f"Start Fail: body code is <{body['code']}>"
        with self.__expireLock:
            self.__nextExpire = time() + ALIVE_TIME

        return f"http://139.9.167.178:180/rtp/{device}_{channel}.live.mp4"  # body["data"]["fmp4"]

    def stop(self, device: "str", channel: "str") -> "None":
        if self.__isExpired():
            self.__login()
        resp = self.__sess.get(f"{WVP_API}/play/stop/{device}/{channel}", headers=HEADER)
        assert resp.status_code == 200, f"Stop Fail: status code is <{resp.status_code}>"
        with self.__expireLock:
            self.__nextExpire = time() + ALIVE_TIME