From e2fb5196372bcd26cd54b9782fdd79207dea4c9c Mon Sep 17 00:00:00 2001 From: Andrea Brancaleoni Date: Tue, 2 May 2023 14:30:02 +0200 Subject: [PATCH 1/2] art: get matte list on fw14xx --- samsungtvws/art.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/samsungtvws/art.py b/samsungtvws/art.py index 62a3fad..d93c9ed 100644 --- a/samsungtvws/art.py +++ b/samsungtvws/art.py @@ -352,7 +352,10 @@ def get_matte_list(self): assert response data = json.loads(response["data"]) - return json.loads(data["matte_type_list"]) + if "matte_type_list" in data: + return json.loads(data["matte_type_list"]) + else "matte_list" in data: + return json.loads(data["matte_list"]) def change_matte(self, content_id, matte_id): self._send_art_request( From c2e38e6daf6a32aa80d4c9ceab4f032a2e90818b Mon Sep 17 00:00:00 2001 From: Andrea Brancaleoni Date: Tue, 2 May 2023 15:20:21 +0200 Subject: [PATCH 2/2] art: fix for fw14xx --- samsungtvws/art.py | 176 ++++++++++++++++++++++++++++++--------------- 1 file changed, 119 insertions(+), 57 deletions(-) diff --git a/samsungtvws/art.py b/samsungtvws/art.py index d93c9ed..dac3ee8 100644 --- a/samsungtvws/art.py +++ b/samsungtvws/art.py @@ -27,6 +27,7 @@ import socket from typing import Any, Dict, List, Optional, Union import uuid +import time import websocket @@ -131,6 +132,52 @@ def _send_art_request( return response + def _send_art_legacy( + self, + request_data: Dict[str, Any], + file, + wait_for_event: Optional[str] = None, + wait_for_sub_event: Optional[str] = None, + ) -> Optional[Dict[str, Any]]: + + + payload = bytearray() + header = bytes(ArtChannelEmitCommand.art_app_request(request_data).get_payload(), 'utf-8') + payload.extend(len(header).to_bytes(2, "big")) + payload.extend(header) + payload.extend(file) + _LOGGING.debug("SamsungTVWS websocket command: %s", payload) + + self.open() + self.connection.send(payload, websocket.ABNF.OPCODE_BINARY) + + if not wait_for_event: + return None + + assert self.connection + event: Optional[str] = None + sub_event: Optional[str] = None + + while event != wait_for_event: + data = self.connection.recv() + response = helper.process_api_response(data) + event = response.get("event", "*") + assert event + self._websocket_event(event, response) + if event == wait_for_event and wait_for_sub_event: + # Check sub event, reset event if it doesn't match + data = json.loads(response["data"]) + sub_event = data.get("event", "*") + if sub_event == "error": + raise exceptions.ResponseError( + f"`{request_data['request']}` request failed " + f"with error number {data['error_code']}" + ) + if sub_event != wait_for_sub_event: + event = None + + return response + def _get_rest_api(self) -> SamsungTVRest: if self._rest_api is None: self._rest_api = SamsungTVRest(self.host, self.port, self.timeout) @@ -221,63 +268,78 @@ def upload(self, file, matte="shadowbox_polar", file_type="png", date=None): if date is None: date = datetime.now().strftime("%Y:%m:%d %H:%M:%S") - response = self._send_art_request( - { - "request": "send_image", - "file_type": file_type, - "conn_info": { - "d2d_mode": "socket", - "connection_id": random.randrange(4 * 1024 * 1024 * 1024), - "id": self.art_uuid, + try: + response = self._send_art_request( + { + "request": "send_image", + "file_type": file_type, + "conn_info": { + "d2d_mode": "socket", + "connection_id": random.randrange(4 * 1024 * 1024 * 1024), + "id": self.art_uuid, + }, + "image_date": date, + "matte_id": matte, + "file_size": file_size, }, - "image_date": date, - "matte_id": matte, - "file_size": file_size, - }, - wait_for_event=D2D_SERVICE_MESSAGE_EVENT, - wait_for_sub_event="ready_to_use", - ) - assert response - data = json.loads(response["data"]) - conn_info = json.loads(data["conn_info"]) - - header = json.dumps( - { - "num": 0, - "total": 1, - "fileLength": file_size, - "fileName": "dummy", - "fileType": file_type, - "secKey": conn_info["key"], - "version": "0.0.1", - } - ) - - art_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - art_socket.connect((conn_info["ip"], int(conn_info["port"]))) - art_socket.send(len(header).to_bytes(4, "big")) - art_socket.send(header.encode("ascii")) - art_socket.send(file) - - wait_for_sub_event = "image_added" - wait_for_event = "d2d_service_message" - - assert self.connection - event: Optional[str] = None - sub_event: Optional[str] = None - - while event != wait_for_event: - data = self.connection.recv() - response = helper.process_api_response(data) - event = response.get("event", "*") - assert event - self._websocket_event(event, response) - if event == wait_for_event and wait_for_sub_event: - # Check sub event, reset event if it doesn't match - data = json.loads(response["data"]) - sub_event = data.get("event", "*") - if sub_event != wait_for_sub_event: - event = None + wait_for_event=D2D_SERVICE_MESSAGE_EVENT, + wait_for_sub_event="ready_to_use", + ) + + assert response + data = json.loads(response["data"]) + conn_info = json.loads(data["conn_info"]) + + header = json.dumps( + { + "num": 0, + "total": 1, + "fileLength": file_size, + "fileName": "dummy", + "fileType": file_type, + "secKey": conn_info["key"], + "version": "0.0.1", + } + ) + + art_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + art_socket.connect((conn_info["ip"], int(conn_info["port"]))) + art_socket.send(len(header).to_bytes(4, "big")) + art_socket.send(header.encode("ascii")) + art_socket.send(file) + + wait_for_sub_event = "image_added" + wait_for_event = "d2d_service_message" + + assert self.connection + event: Optional[str] = None + sub_event: Optional[str] = None + + while event != wait_for_event: + data = self.connection.recv() + response = helper.process_api_response(data) + event = response.get("event", "*") + assert event + self._websocket_event(event, response) + if event == wait_for_event and wait_for_sub_event: + # Check sub event, reset event if it doesn't match + data = json.loads(response["data"]) + sub_event = data.get("event", "*") + if sub_event != wait_for_sub_event: + event = None + + except exceptions.ResponseError: + response = self._send_art_legacy( + { + "request": "send_image", + "file_type": file_type, + "matte_id": matte, + "id": self.art_uuid + }, + file=file, + wait_for_event=D2D_SERVICE_MESSAGE_EVENT, + wait_for_sub_event="image_added" + ) data = json.loads(response["data"]) @@ -354,7 +416,7 @@ def get_matte_list(self): if "matte_type_list" in data: return json.loads(data["matte_type_list"]) - else "matte_list" in data: + elif "matte_list" in data: return json.loads(data["matte_list"]) def change_matte(self, content_id, matte_id):