Skip to content
This repository was archived by the owner on Jul 10, 2024. It is now read-only.
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
21 changes: 12 additions & 9 deletions samsungctl/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@
from . import Remote


_LOGGER = logging.getLogger(__package__)


def _read_config():
config = collections.defaultdict(lambda: None, {
"name": "samsungctl",
Expand Down Expand Up @@ -53,7 +56,7 @@ def _read_config():
config_json = json.load(config_file)
except ValueError as e:
message = "Warning: Could not parse the configuration file.\n %s"
logging.warning(message, e)
_LOGGER.warning(message, e)
return config

config.update(config_json)
Expand Down Expand Up @@ -103,7 +106,7 @@ def main():
config.update({k: v for k, v in vars(args).items() if v is not None})

if not config["host"]:
logging.error("Error: --host must be set")
_LOGGER.error("Error: --host must be set")
return

try:
Expand All @@ -112,21 +115,21 @@ def main():
remote.control(key)

if args.interactive:
logging.getLogger().setLevel(logging.ERROR)
_LOGGER.setLevel(logging.ERROR)
from . import interactive
interactive.run(remote)
elif len(args.key) == 0:
logging.warning("Warning: No keys specified.")
_LOGGER.warning("Warning: No keys specified.")
except exceptions.ConnectionClosed:
logging.error("Error: Connection closed!")
_LOGGER.error("Error: Connection closed!")
except exceptions.AccessDenied:
logging.error("Error: Access denied!")
_LOGGER.error("Error: Access denied!")
except exceptions.UnknownMethod:
logging.error("Error: Unknown method '{}'".format(config["method"]))
_LOGGER.error("Error: Unknown method '{}'".format(config["method"]))
except socket.timeout:
logging.error("Error: Timed out!")
_LOGGER.error("Error: Timed out!")
except OSError as e:
logging.error("Error: %s", e.strerror)
_LOGGER.error("Error: %s", e.strerror)


if __name__ == "__main__":
Expand Down
17 changes: 9 additions & 8 deletions samsungctl/remote_legacy.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import time

from . import exceptions
_LOGGER = logging.getLogger(__package__)


class RemoteLegacy():
Expand All @@ -27,7 +28,7 @@ def __init__(self, config):
+ self._serialize_string(config["name"])
packet = b"\x00\x00\x00" + self._serialize_string(payload, True)

logging.info("Sending handshake.")
_LOGGER.info("Sending handshake.")
self.connection.send(packet)
self._read_response(True)

Expand All @@ -42,7 +43,7 @@ def close(self):
if self.connection:
self.connection.close()
self.connection = None
logging.debug("Connection closed.")
_LOGGER.debug("Connection closed.")

def control(self, key):
"""Send a control command."""
Expand All @@ -52,7 +53,7 @@ def control(self, key):
payload = b"\x00\x00\x00" + self._serialize_string(key)
packet = b"\x00\x00\x00" + self._serialize_string(payload, True)

logging.info("Sending control command: %s", key)
_LOGGER.info("Sending control command: %s", key)
self.connection.send(packet)
self._read_response()
time.sleep(self._key_interval)
Expand All @@ -66,7 +67,7 @@ def _read_response(self, first_time=False):
tv_name = self.connection.recv(tv_name_len)

if first_time:
logging.debug("Connected to '%s'.", tv_name.decode())
_LOGGER.debug("Connected to '%s'.", tv_name.decode())

response_len = int.from_bytes(self.connection.recv(2),
byteorder="little")
Expand All @@ -77,19 +78,19 @@ def _read_response(self, first_time=False):
raise exceptions.ConnectionClosed()

if response == b"\x64\x00\x01\x00":
logging.debug("Access granted.")
_LOGGER.debug("Access granted.")
return
elif response == b"\x64\x00\x00\x00":
raise exceptions.AccessDenied()
elif response[0:1] == b"\x0a":
if first_time:
logging.warning("Waiting for authorization...")
_LOGGER.warning("Waiting for authorization...")
return self._read_response()
elif response[0:1] == b"\x65":
logging.warning("Authorization cancelled.")
_LOGGER.warning("Authorization cancelled.")
raise exceptions.AccessDenied()
elif response == b"\x00\x00\x00\x00":
logging.debug("Control accepted.")
_LOGGER.debug("Control accepted.")
return

raise exceptions.UnhandledResponse(response)
Expand Down
7 changes: 4 additions & 3 deletions samsungctl/remote_websocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@


URL_FORMAT = "ws://{}:{}/api/v2/channels/samsung.remote.control?name={}"
_LOGGER = logging.getLogger(__package__)


class RemoteWebsocket():
Expand Down Expand Up @@ -40,7 +41,7 @@ def close(self):
if self.connection:
self.connection.close()
self.connection = None
logging.debug("Connection closed.")
_LOGGER.debug("Connection closed.")

def control(self, key):
"""Send a control command."""
Expand All @@ -57,7 +58,7 @@ def control(self, key):
}
})

logging.info("Sending control command: %s", key)
_LOGGER.info("Sending control command: %s", key)
self.connection.send(payload)
time.sleep(self._key_interval)

Expand All @@ -71,7 +72,7 @@ def _read_response(self):
self.close()
raise exceptions.UnhandledResponse(response)

logging.debug("Access granted.")
_LOGGER.debug("Access granted.")

@staticmethod
def _serialize_string(string):
Expand Down