Add main.py
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
import sys
|
||||
import os
|
||||
import socket
|
||||
import struct
|
||||
import time
|
||||
|
||||
# Standard KeyCodes
|
||||
KEY_MAP = {
|
||||
'w': (19, "DPAD_UP"),
|
||||
's': (20, "DPAD_DOWN"),
|
||||
'a': (21, "DPAD_LEFT"),
|
||||
'd': (22, "DPAD_RIGHT"),
|
||||
' ': (23, "DPAD_CENTER (OK)"),
|
||||
'e': (23, "DPAD_CENTER (OK)"),
|
||||
'\r': (23, "DPAD_CENTER (OK)"),
|
||||
'\n': (23, "DPAD_CENTER (OK)"),
|
||||
'b': (4, "BACK"),
|
||||
'h': (3, "HOME"),
|
||||
'm': (82, "MENU"),
|
||||
'+': (24, "VOLUME_UP"),
|
||||
'=': (24, "VOLUME_UP"),
|
||||
'-': (25, "VOLUME_DOWN"),
|
||||
'_': (25, "VOLUME_DOWN"),
|
||||
'0': (164, "VOLUME_MUTE"),
|
||||
'p': (26, "POWER"),
|
||||
}
|
||||
|
||||
def discover(timeout=2.5):
|
||||
print("Searching for compatible TVs on the local network (UDP 48689)...")
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
||||
sock.settimeout(timeout)
|
||||
|
||||
packet = bytearray(50)
|
||||
magic = b"FindECloudBox"
|
||||
struct.pack_into(">IIIII", packet, 0, 0, 0, 0, len(magic), 0)
|
||||
packet[20:20 + len(magic)] = magic
|
||||
|
||||
try:
|
||||
sock.sendto(packet, ("255.255.255.255", 48689))
|
||||
while True:
|
||||
data, addr = sock.recvfrom(1024)
|
||||
reply = data.decode("utf-8", errors="ignore")
|
||||
if reply.startswith("ECloudBox"):
|
||||
parts = reply.split(":")
|
||||
name = parts[1] if len(parts) > 1 else "Unknown"
|
||||
port = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 8121
|
||||
print(f"Found TV: '{name}' at {addr[0]}:{port}")
|
||||
return addr[0], port, name
|
||||
except socket.timeout:
|
||||
print("No TV responded to UDP broadcast.")
|
||||
finally:
|
||||
sock.close()
|
||||
return None
|
||||
|
||||
def connect(ip, port=8121):
|
||||
print(f"Connecting to {ip}:{port}...")
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.settimeout(5.0)
|
||||
sock.connect((ip, port))
|
||||
|
||||
sock.sendall(b"getServerInfo\r\neshare\r\n\r\n")
|
||||
try:
|
||||
resp = sock.recv(1024).decode("utf-8", errors="ignore")
|
||||
if resp.strip():
|
||||
print(f"Server response: {resp.strip()[:100]}")
|
||||
except socket.timeout:
|
||||
pass
|
||||
|
||||
sock.sendall(b"GETPASSWORDCONFIG\r\neshare\r\n\r\n")
|
||||
try:
|
||||
resp = sock.recv(1024).decode("utf-8", errors="ignore")
|
||||
except socket.timeout:
|
||||
pass
|
||||
|
||||
sock.sendall(b"sayHello\r\nEShareClient\r\n\r\n")
|
||||
print("Handshake complete! Connected.")
|
||||
return sock
|
||||
|
||||
def send_key(sock, keycode):
|
||||
cmd = f"KEYEVENT\r\n{keycode}\r\n\r\n".encode("utf-8")
|
||||
sock.sendall(cmd)
|
||||
|
||||
def send_text(sock, text):
|
||||
escaped = text.replace("\\", "\\\\").replace("\r", "\\r").replace("\n", "\\n").replace("\t", "\\t")
|
||||
cmd = f"CONTENT\r\n1\r\n{escaped}\r\n\r\n".encode("utf-8")
|
||||
sock.sendall(cmd)
|
||||
|
||||
def getch():
|
||||
if not sys.stdin.isatty():
|
||||
line = sys.stdin.readline()
|
||||
return line[0] if line else 'q'
|
||||
|
||||
import tty, termios
|
||||
fd = sys.stdin.fileno()
|
||||
old = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setraw(fd)
|
||||
ch = sys.stdin.read(1)
|
||||
if ch == '\x1b':
|
||||
ch2 = sys.stdin.read(1)
|
||||
if ch2 == '[':
|
||||
ch3 = sys.stdin.read(1)
|
||||
arrows = {'A': 'w', 'B': 's', 'C': 'd', 'D': 'a'}
|
||||
return arrows.get(ch3, '')
|
||||
return ch
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old)
|
||||
|
||||
def print_help():
|
||||
print("\n--- Controls ---")
|
||||
print(" [W / ↑] Up [S / ↓] Down")
|
||||
print(" [A / ←] Left [D / →] Right")
|
||||
print(" [Space/E/Enter] OK [B] Back")
|
||||
print(" [H] Home [M] Menu")
|
||||
print(" [+] Vol Up [-] Vol Down")
|
||||
print(" [0] Mute [P] Power")
|
||||
print(" [T] Type text [Q] Quit")
|
||||
print("----------------\n")
|
||||
|
||||
def main():
|
||||
tv_ip = None
|
||||
tv_port = 8121
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
tv_ip = sys.argv[1]
|
||||
if len(sys.argv) > 2:
|
||||
tv_port = int(sys.argv[2])
|
||||
else:
|
||||
found = discover()
|
||||
if found:
|
||||
tv_ip, tv_port, _ = found
|
||||
else:
|
||||
tv_ip = input("Enter TV IP address manually: ").strip()
|
||||
|
||||
if not tv_ip:
|
||||
print("No IP provided. Exiting.")
|
||||
return
|
||||
|
||||
sock = connect(tv_ip, tv_port)
|
||||
print_help()
|
||||
|
||||
try:
|
||||
while True:
|
||||
ch = getch()
|
||||
if not ch:
|
||||
continue
|
||||
|
||||
lower_ch = ch.lower()
|
||||
|
||||
if lower_ch == 'q' or ch == '\x03':
|
||||
print("\nExiting...")
|
||||
break
|
||||
|
||||
if lower_ch == 't':
|
||||
print()
|
||||
text = input("Enter text to send to TV (keyboard input): ")
|
||||
if text:
|
||||
send_text(sock, text)
|
||||
print(f"Sent text: {repr(text)}")
|
||||
continue
|
||||
|
||||
if lower_ch in KEY_MAP:
|
||||
keycode, name = KEY_MAP[lower_ch]
|
||||
send_key(sock, keycode)
|
||||
print(f"Sent: {name} (code {keycode})", flush=True)
|
||||
else:
|
||||
print(f"Unknown key: {repr(ch)} (type 'q' to quit)", flush=True)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted.")
|
||||
except Exception as e:
|
||||
print(f"\nConnection error: {e}")
|
||||
finally:
|
||||
try:
|
||||
sock.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user