diff --git a/app/wyzebridge/wyze_api.py b/app/wyzebridge/wyze_api.py index 74ca385e..e1ff8102 100644 --- a/app/wyzebridge/wyze_api.py +++ b/app/wyzebridge/wyze_api.py @@ -18,6 +18,7 @@ from wyzebridge.bridge_utils import env_bool, env_filter from wyzebridge.config import IMG_PATH, MOTION, TOKEN_PATH from wyzebridge.logging import logger +from wyzecam.api import RateLimitError, WyzeAPIError, post_v2_device from wyzecam.api_models import WyzeAccount, WyzeCamera, WyzeCredential @@ -64,7 +65,7 @@ def wrapper(self, *args: Any, **kwargs: Any): if not self.refresh_token(): return return func(self, *args, **kwargs) - except (wyzecam.api.RateLimitError, wyzecam.api.WyzeAPIError) as ex: + except (RateLimitError, wyzecam.api.WyzeAPIError) as ex: logger.error(f"[API] {ex}") except ConnectionError as ex: logger.warning(f"[API] {ex}") @@ -148,7 +149,7 @@ def login(self, fresh_data: bool = False) -> Optional[WyzeCredential]: sleep(15) except ValueError as ex: logger.error(f"[API] {ex}") - except wyzecam.api.RateLimitError as ex: + except RateLimitError as ex: logger.error(f"[API] {ex}") except RequestException as ex: logger.error(f"[API] ERROR: {ex}") @@ -243,7 +244,7 @@ def get_kvs_signal(self, cam_name: str) -> Optional[dict]: logger.info("☁️ Fetching signaling data from the Wyze API...") wss = wyzecam.api.get_cam_webrtc(self.auth, cam.mac) return wss | {"result": "ok", "cam": cam_name} - except (HTTPError, wyzecam.api.WyzeAPIError) as ex: + except (HTTPError, WyzeAPIError) as ex: if isinstance(ex, HTTPError) and ex.response.status_code == 404: ex = "Camera does not support WebRTC" logger.warning(ex) @@ -300,27 +301,40 @@ def run_action(self, cam: WyzeCamera, action: str): try: resp = wyzecam.api.run_action(self.auth, cam, action.lower()) return {"status": "success", "response": resp["result"]} - except (ValueError, wyzecam.api.WyzeAPIError) as ex: + except (ValueError, WyzeAPIError) as ex: logger.error(f"[CONTROL] ERROR {ex}") return {"status": "error", "response": str(ex)} @authenticated - def get_pid_info(self, cam: WyzeCamera, pid: str = ""): - logger.info(f"[CONTROL] ☁️ Get Device Info for {cam.name_uri} via Wyze API") + def get_device_info(self, cam: WyzeCamera, pid: str = ""): + logger.info(f"[CONTROL] ☁️ get_device_Info for {cam.name_uri} via Wyze API") + params = {"device_mac": cam.mac, "device_model": cam.product_model} try: - property_list = wyzecam.api.get_device(self.auth, "get_device_Info", cam)[ - "property_list" - ] - except (ValueError, wyzecam.api.WyzeAPIError) as ex: + res = post_v2_device(self.auth, "get_device_Info", params)["property_list"] + except (ValueError, WyzeAPIError) as ex: logger.error(f"[CONTROL] ERROR - {ex}") return {"status": "error", "response": str(ex)} if not pid: - return {"status": "success", "response": property_list} + return {"status": "success", "response": res} - resp = next((item for item in property_list if item["pid"] == pid)) + if not (item := next((i for i in res if i["pid"] == pid), None)): + logger.error(f"[CONTROL] ERROR - {pid} not found") + return {"status": "error", "response": f"{pid} not found"} - return {"status": "success", "value": resp.get("value"), "response": resp} + return {"status": "success", "value": item.get("value"), "response": item} + + @authenticated + def set_property(self, cam: WyzeCamera, pid: str = ""): + logger.info(f"[CONTROL] ☁️ set_property for {cam.name_uri} via Wyze API") + params = {"device_mac": cam.mac, "device_model": cam.product_model} + try: + res = post_v2_device(self.auth, "set_property", params)["property_list"] + except (ValueError, WyzeAPIError) as ex: + logger.error(f"[CONTROL] ERROR - {ex}") + return {"status": "error", "response": str(ex)} + + return {"status": "success", "response": res} @authenticated def get_events(self, macs: Optional[list] = None, last_ts: int = 0): @@ -334,21 +348,22 @@ def get_events(self, macs: Optional[list] = None, last_ts: int = 0): } try: - resp = wyzecam.api.get_device(self.auth, "get_event_list", params=params) + resp = post_v2_device(self.auth, "get_event_list", params) return time(), resp["event_list"] - except wyzecam.api.RateLimitError as ex: + except RateLimitError as ex: logger.error(f"[EVENTS] RateLimitError: {ex}, cooling down.") return ex.reset_by, [] @authenticated def set_device_info(self, cam: WyzeCamera, params: dict): if not isinstance(params, dict): - return {"status": "error", "response": f"invalid param type [{params=}]"} + return {"status": "error", "response": f"Invalid params [{params=}]"} + logger.info( + f"[CONTROL] ☁ set_device_Info {params} for {cam.name_uri} via Wyze API" + ) + params |= {"device_mac": cam.mac} try: - logger.info( - f"[CONTROL] ☁ Set Device Info {params} for {cam.name_uri} via Wyze API" - ) - wyzecam.api.set_device_info(self.auth, cam, params) + wyzecam.api.post_device(self.auth, "set_device_Info", params) return {"status": "success", "response": "success"} except ValueError as ex: error = f'{ex.args[0].get("code")}: {ex.args[0].get("msg")}' diff --git a/app/wyzebridge/wyze_commands.py b/app/wyzebridge/wyze_commands.py index eefe96c3..60266f53 100644 --- a/app/wyzebridge/wyze_commands.py +++ b/app/wyzebridge/wyze_commands.py @@ -18,7 +18,6 @@ "motion_tracking": "K11020GetMotionTracking", "motion_tagging": "K10290GetMotionTagging", "camera_info": "K10020CheckCameraInfo", - "battery": "K10050GetPowerLevel", "battery_usage": "K10448GetBatteryUsage", "rtsp": "K10604GetRtspParam", "param_info": "K10020CheckCameraParams", # Requires a Payload diff --git a/app/wyzebridge/wyze_stream.py b/app/wyzebridge/wyze_stream.py index 9389de24..cb66ea31 100644 --- a/app/wyzebridge/wyze_stream.py +++ b/app/wyzebridge/wyze_stream.py @@ -291,7 +291,7 @@ def state_control(self, payload) -> dict: def power_control(self, payload: str) -> dict: if payload not in {"on", "off", "restart"}: - resp = self.api.get_pid_info(self.camera, "P3") + resp = self.api.get_device_info(self.camera, "P3") resp["value"] = "on" if resp["value"] == "1" else "off" return resp run_cmd = payload if payload == "restart" else f"power_{payload}" @@ -319,7 +319,10 @@ def send_cmd(self, cmd: str, payload: str | list | dict = "") -> dict: return self.state_control(payload or cmd) if cmd == "device_info": - return self.api.get_pid_info(self.camera) + return self.api.get_device_info(self.camera) + + if cmd == "battery": + return self.api.get_device_info(self.camera, "P8") if cmd == "power": return self.power_control(str(payload).lower()) diff --git a/app/wyzecam/api.py b/app/wyzecam/api.py index 9837420f..e1cd67a0 100644 --- a/app/wyzecam/api.py +++ b/app/wyzecam/api.py @@ -8,7 +8,7 @@ from os import environ, getenv from typing import Any, Optional -import requests +from requests import Response, get, post from wyzecam.api_models import WyzeAccount, WyzeCamera, WyzeCredential IOS_VERSION = getenv("IOS_VERSION") @@ -92,11 +92,11 @@ def login( [get_camera_list()][wyzecam.api.get_camera_list]. """ phone_id = phone_id or str(uuid.uuid4()) - headers = get_headers(phone_id) + headers = _headers(phone_id) headers["content-type"] = "application/json" payload = sort_dict( - {"email": email.strip(), "password": triplemd5(password), **(mfa or {})} + {"email": email.strip(), "password": hash_password(password), **(mfa or {})} ) api_version = "v2" if getenv("API_ID") and getenv("API_KEY"): @@ -106,9 +106,7 @@ def login( headers["appid"] = "umgm_78ae6013d158c4a5" headers["signature2"] = sign_msg("v3", payload) - resp = requests.post( - f"{AUTH_API}/{api_version}/user/login", data=payload, headers=headers - ) + resp = post(f"{AUTH_API}/{api_version}/user/login", data=payload, headers=headers) resp.raise_for_status() return WyzeCredential.model_validate(dict(resp.json(), phone_id=phone_id)) @@ -124,7 +122,7 @@ def send_sms_code(auth_info: WyzeCredential, phone: str = "Primary") -> str: :param auth_info: the result of a [`login()`][wyzecam.api.login] call. :returns: verification_id required to logging in with SMS verification. """ - resp = requests.post( + resp = post( f"{AUTH_API}/user/login/sendSmsCode", json={}, params={ @@ -132,7 +130,7 @@ def send_sms_code(auth_info: WyzeCredential, phone: str = "Primary") -> str: "sessionId": auth_info.sms_session_id, "userId": auth_info.user_id, }, - headers=get_headers(auth_info.phone_id), + headers=_headers(auth_info.phone_id), ) resp.raise_for_status() @@ -149,14 +147,14 @@ def send_email_code(auth_info: WyzeCredential) -> str: :param auth_info: the result of a [`login()`][wyzecam.api.login] call. :returns: verification_id required to logging in with SMS verification. """ - resp = requests.post( + resp = post( f"{AUTH_API}/v2/user/login/sendEmailCode", json={}, params={ "userId": auth_info.user_id, "sessionId": auth_info.email_session_id, }, - headers=get_headers(auth_info.phone_id), + headers=_headers(auth_info.phone_id), ) resp.raise_for_status() @@ -176,18 +174,17 @@ def refresh_token(auth_info: WyzeCredential) -> WyzeCredential: [get_camera_list()][wyzecam.api.get_camera_list]. """ - payload = _get_payload(auth_info.access_token, auth_info.phone_id) + payload = _payload(auth_info.access_token, auth_info.phone_id) payload["refresh_token"] = auth_info.refresh_token - resp = requests.post( + resp = post( f"{WYZE_API}/user/refresh_token", json=payload, - headers=get_headers(), + headers=_headers(), ) - resp_json = validate_resp(resp) return WyzeCredential.model_validate( dict( - resp_json["data"], + validate_resp(resp)["data"], user_id=auth_info.user_id, phone_id=auth_info.phone_id, ) @@ -206,28 +203,26 @@ def get_user_info(auth_info: WyzeCredential) -> WyzeAccount: for passing to [`WyzeIOTC.connect_and_auth()`][wyzecam.iotc.WyzeIOTC.connect_and_auth]. """ - resp = requests.post( + resp = post( f"{WYZE_API}/user/get_user_info", - json=_get_payload(auth_info.access_token, auth_info.phone_id), - headers=get_headers(), + json=_payload(auth_info.access_token, auth_info.phone_id), + headers=_headers(), ) - resp_json = validate_resp(resp) return WyzeAccount.model_validate( - dict(resp_json["data"], phone_id=auth_info.phone_id) + dict(validate_resp(resp)["data"], phone_id=auth_info.phone_id) ) def get_homepage_object_list(auth_info: WyzeCredential) -> dict[str, Any]: """Get all homepage objects.""" - resp = requests.post( + resp = post( f"{WYZE_API}/v2/home_page/get_object_list", - json=_get_payload(auth_info.access_token, auth_info.phone_id), - headers=get_headers(), + json=_payload(auth_info.access_token, auth_info.phone_id), + headers=_headers(), ) - resp_json = validate_resp(resp) - return resp_json["data"] + return validate_resp(resp)["data"] def get_camera_list(auth_info: WyzeCredential) -> list[WyzeCamera]: @@ -285,59 +280,31 @@ def get_camera_list(auth_info: WyzeCredential) -> list[WyzeCamera]: def run_action(auth_info: WyzeCredential, camera: WyzeCamera, action: str): """Send run_action commands to the camera.""" payload = dict( - _get_payload(auth_info.access_token, auth_info.phone_id, "run_action"), + _payload(auth_info.access_token, auth_info.phone_id, "run_action"), action_params={}, action_key=action, instance_id=camera.mac, provider_key=camera.product_model, ) - resp = requests.post( - f"{WYZE_API}/v2/auto/run_action", json=payload, headers=get_headers() - ) + resp = post(f"{WYZE_API}/v2/auto/run_action", json=payload, headers=_headers()) - resp_json = validate_resp(resp) + return validate_resp(resp)["data"] - return resp_json["data"] +def post_v2_device(auth_info: WyzeCredential, endpoint: str, params: dict) -> dict: + """Post data to the v2 device API.""" + params |= _payload(auth_info.access_token, auth_info.phone_id, endpoint) + resp = post(f"{WYZE_API}/v2/device/{endpoint}", json=params, headers=_headers()) -def get_device( - auth_info: WyzeCredential, - endpoint: str, - camera: Optional[WyzeCamera] = None, - params: Optional[dict] = None, -) -> dict: - """Get data from the v2 device API.""" - payload = _get_payload(auth_info.access_token, auth_info.phone_id, endpoint) + return validate_resp(resp)["data"] - if camera: - payload |= {"device_mac": camera.mac, "device_model": camera.product_model} - if params: - payload |= params - resp = requests.post( - f"{WYZE_API}/v2/device/{endpoint}", json=payload, headers=get_headers() - ) - resp_json = validate_resp(resp) +def post_device(auth_info: WyzeCredential, endpoint: str, params: dict) -> dict: + """Post data to the v1 device API.""" + params |= _payload(auth_info.access_token, auth_info.phone_id, endpoint) + resp = post(f"{WYZE_API}/device/{endpoint}", json=params, headers=_headers()) - return resp_json["data"] - - -def set_device_info( - auth_info: WyzeCredential, camera: WyzeCamera, params: dict -) -> dict: - """Get device info.""" - payload = dict( - _get_payload(auth_info.access_token, auth_info.phone_id, "set_device_Info"), - device_mac=camera.mac, - **params, - ) - resp = requests.post( - f"{WYZE_API}/device/set_device_info", json=payload, headers=get_headers() - ) - - resp_json = validate_resp(resp) - - return resp_json["data"] + return validate_resp(resp)["data"] def get_cam_webrtc(auth_info: WyzeCredential, mac_id: str) -> dict: @@ -345,11 +312,11 @@ def get_cam_webrtc(auth_info: WyzeCredential, mac_id: str) -> dict: if not auth_info.access_token: raise AccessTokenError() - ui_headers = get_headers() + ui_headers = _headers() ui_headers["content-type"] = "application/json" ui_headers["authorization"] = auth_info.access_token - resp = requests.get( + resp = get( f"https://webrtc.api.wyze.com/signaling/device/{mac_id}?use_trickle=true", headers=ui_headers, ) @@ -365,7 +332,7 @@ def get_cam_webrtc(auth_info: WyzeCredential, mac_id: str) -> dict: } -def validate_resp(resp): +def validate_resp(resp: Response) -> dict: resp.raise_for_status() if int(resp.headers.get("X-RateLimit-Remaining", 100)) <= 10: raise RateLimitError(resp) @@ -378,12 +345,12 @@ def validate_resp(resp): return resp_json -def _get_payload( - access_token: Optional[str], phone_id: Optional[str] = "", req_path: str = "default" -): +def _payload( + access_token: Optional[str], phone_id: Optional[str] = "", endpoint: str = "default" +) -> dict: return { - "sc": SC_SV[req_path]["sc"], - "sv": SC_SV[req_path]["sv"], + "sc": SC_SV[endpoint]["sc"], + "sv": SC_SV[endpoint]["sv"], "app_ver": f"com.hualai.WyzeCam___{APP_VERSION}", "app_version": APP_VERSION, "app_name": "com.hualai.WyzeCam", @@ -394,7 +361,7 @@ def _get_payload( } -def get_headers(phone_id: Optional[str] = "") -> dict[str, str]: +def _headers(phone_id: Optional[str] = None) -> dict[str, str]: if not phone_id: return {"user-agent": SCALE_USER_AGENT} id, key = getenv("API_ID"), getenv("API_KEY") @@ -402,7 +369,7 @@ def get_headers(phone_id: Optional[str] = "") -> dict[str, str]: return { "apikey": key, "keyid": id, - "user-agent": f"docker-wyze-bridge-{getenv('VERSION')}", + "user-agent": f"docker-wyze-bridge/{getenv('VERSION')}", } return { @@ -412,7 +379,7 @@ def get_headers(phone_id: Optional[str] = "") -> dict[str, str]: } -def triplemd5(password: str) -> str: +def hash_password(password: str) -> str: """Run hashlib.md5() algorithm 3 times.""" encoded = password.strip() for _ in range(3): diff --git a/app/wyzecam/tutk/tutk_protocol.py b/app/wyzecam/tutk/tutk_protocol.py index 574a05ed..8dec6536 100644 --- a/app/wyzecam/tutk/tutk_protocol.py +++ b/app/wyzecam/tutk/tutk_protocol.py @@ -467,18 +467,6 @@ def parse_response(self, resp_data): } -class K10050GetPowerLevel(TutkWyzeProtocolMessage): - def __init__(self): - super().__init__(10050) - - def parse_response(self, resp_data): - data = json.loads(resp_data) - try: - return data["camerainfo"]["powerlevel"] - except KeyError: - return 0 - - class K10056SetResolvingBit(TutkWyzeProtocolMessage): """ A message used to set the resolution and bitrate of the camera.