Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

bugfix/frame-fw14xx #125

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+122 −57
Open

bugfix/frame-fw14xx #125

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e2fb519
art: get matte list on fw14xx
thypon May 2, 2023
c2e38e6
art: fix for fw14xx
thypon May 2, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 122 additions & 57 deletions samsungtvws/art.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import socket
from typing import Any, Dict, List, Optional, Union
import uuid
import time

import websocket

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think here it would be better to split it completely into two methods:

    try:
        self._upload_v1(file, matte, file_type, date)
    except exceptions.ResponseError:
        self._upload_v2(file, matte, file_type, date)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

won't have access to the TV 'till next month, will check later.
Thanks for the review!

{
"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"])

Expand Down Expand Up @@ -352,7 +414,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"])
elif "matte_list" in data:
return json.loads(data["matte_list"])
Comment on lines +419 to +420
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's better here to fail hard.

Suggested change
elif "matte_list" in data:
return json.loads(data["matte_list"])
return json.loads(data["matte_list"])

Another option is to raise an exception

        if "matte_type_list" in data:
            return json.loads(data["matte_type_list"])
        if "matte_list" in data:
            return json.loads(data["matte_list"])
        raise ...


def change_matte(self, content_id, matte_id):
self._send_art_request(
Expand Down