ION606
92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
from roku import Roku
|
|
import os
|
|
import json
|
|
import threading
|
|
|
|
ROKU_IP = os.environ.get("ROKU_IP")
|
|
if not ROKU_IP:
|
|
raise Exception("Roku IP not found in env!")
|
|
|
|
COMMANDS = {
|
|
# Standard Keys
|
|
"home": "Home",
|
|
"reverse": "Rev",
|
|
"forward": "Fwd",
|
|
"play": "Play",
|
|
"select": "Select",
|
|
"left": "Left",
|
|
"right": "Right",
|
|
"down": "Down",
|
|
"up": "Up",
|
|
"back": "Back",
|
|
"replay": "InstantReplay",
|
|
"info": "Info",
|
|
"backspace": "Backspace",
|
|
"search": "Search",
|
|
"enter": "Enter",
|
|
"literal": "Lit",
|
|
# For devices that support "Find Remote"
|
|
"find_remote": "FindRemote",
|
|
# For Roku TV
|
|
"volume_down": "VolumeDown",
|
|
"volume_up": "VolumeUp",
|
|
"volume_mute": "VolumeMute",
|
|
# For Roku TV while on TV tuner channel
|
|
"channel_up": "ChannelUp",
|
|
"channel_down": "ChannelDown",
|
|
# For Roku TV current input
|
|
"input_tuner": "InputTuner",
|
|
"input_hdmi1": "InputHDMI1",
|
|
"input_hdmi2": "InputHDMI2",
|
|
"input_hdmi3": "InputHDMI3",
|
|
"input_hdmi4": "InputHDMI4",
|
|
"input_av1": "InputAV1",
|
|
# For devices that support being turned on/off
|
|
"power": "Power",
|
|
"poweroff": "PowerOff",
|
|
"poweron": "PowerOn",
|
|
}
|
|
|
|
class RokuWrapper():
|
|
def __init__(self, parent):
|
|
self.parent = parent
|
|
|
|
# eh maybe close the connection? Not really needed
|
|
self.timer: threading.Timer | None = None
|
|
|
|
try:
|
|
self.rapp = Roku(ROKU_IP)
|
|
print(f"Roku client connected to {ROKU_IP}", flush=True)
|
|
except Exception as e:
|
|
self.parent.log_error("Failed to create Roku client: %s", e)
|
|
return self.parent._send(500, json.dumps({"error": "roku connection failed"}).encode())
|
|
|
|
def run_command(self, command: str):
|
|
if not command:
|
|
return self.parent._send(404, json.dumps({"error": "command not given"}).encode())
|
|
command = command.split('/')[0]
|
|
|
|
# If command is not known, nothing to do
|
|
if command not in COMMANDS:
|
|
return self.parent._send(404, json.dumps({"error": "unknown command"}).encode())
|
|
|
|
if hasattr(self.rapp, command):
|
|
attr = getattr(self.rapp, command)
|
|
if callable(attr):
|
|
attr()
|
|
return {"result": f"{command} executed"}
|
|
|
|
# Fallback: send the Roku key name using common API methods
|
|
key_name = COMMANDS[command]
|
|
# common method names in Roku libs
|
|
for send_fn in ("press", "keypress", "send_key", "send_keypress", "button"):
|
|
if hasattr(self.rapp, send_fn):
|
|
fn = getattr(self.rapp, send_fn)
|
|
try:
|
|
fn(key_name)
|
|
return {"result": f"{command} sent as {key_name} via {send_fn}"}
|
|
except TypeError:
|
|
continue
|
|
|
|
self.parent._send(404, json.dumps({"error": "unknown endpoint"}).encode())
|