Adolfo Reyna
164 lines
6.8 KiB
Python
164 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
import base64, concurrent.futures, json, os, re, socket, struct, time, urllib.request, urllib.error
|
|
from pathlib import Path
|
|
|
|
OUT = Path('/tmp/esp32_thumbs_result.txt')
|
|
W,H = 320,240
|
|
|
|
def rpc(url, method, params=None, timeout=4):
|
|
payload = json.dumps({'jsonrpc':'2.0','id':str(time.time()),'method':method,'params':params or {}}).encode()
|
|
req = urllib.request.Request(url, data=payload, headers={'Content-Type':'application/json'}, method='POST')
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
txt = r.read().decode('utf-8','replace')
|
|
try:
|
|
return json.loads(txt)
|
|
except Exception:
|
|
return {'raw': txt}
|
|
|
|
def call(url, name, args, timeout=7):
|
|
return rpc(url, 'tools/call', {'name': name, 'arguments': args}, timeout=timeout)
|
|
|
|
def discover_udp():
|
|
found = set()
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
sock.settimeout(0.45)
|
|
for host in ['255.255.255.255','192.168.68.255']:
|
|
try: sock.sendto(b'DISCOVER_SCREEN', (host, 5000))
|
|
except Exception: pass
|
|
end = time.time() + 1.2
|
|
while time.time() < end:
|
|
try:
|
|
data, addr = sock.recvfrom(512)
|
|
text = data.decode('utf-8','replace')
|
|
m = re.search(r'SCREEN_IP_(\d+)', text)
|
|
port = int(m.group(1)) if m else 80
|
|
found.add((addr[0], port))
|
|
except Exception:
|
|
pass
|
|
return found
|
|
|
|
def candidate_urls():
|
|
pairs = set(discover_udp())
|
|
# Add likely LAN hosts; the user said three devices, so find live MCP responders instead of stale aliases.
|
|
for last in range(100, 151):
|
|
pairs.add((f'192.168.68.{last}', 80))
|
|
# Local desktop-style clients if present.
|
|
for ip,port in [('127.0.0.1',8080),('192.168.68.150',8080),('192.168.68.129',8080)]:
|
|
pairs.add((ip,port))
|
|
return [f'http://{ip}:{port}/api/mcp' if port != 80 else f'http://{ip}/api/mcp' for ip,port in sorted(pairs)]
|
|
|
|
def probe(url):
|
|
try:
|
|
res = rpc(url, 'tools/list', {}, timeout=0.9)
|
|
tools = res.get('result',{}).get('tools', [])
|
|
names = [t.get('name','') for t in tools]
|
|
if any(n in names for n in ['draw_image','draw_raw_rgb565','draw_color_bmp','draw_text']):
|
|
return url, names
|
|
except Exception:
|
|
return None
|
|
|
|
def make_pbm():
|
|
pix = [[0]*W for _ in range(H)]
|
|
def rect(x0,y0,x1,y1,v=1):
|
|
for y in range(max(0,y0), min(H,y1)):
|
|
row = pix[y]
|
|
for x in range(max(0,x0), min(W,x1)):
|
|
row[x] = v
|
|
# chunky black thumbs-up silhouette centered
|
|
rect(75,130,125,185) # wrist
|
|
rect(120,95,165,185) # palm
|
|
rect(145,75,175,115) # raised thumb
|
|
rect(160,60,190,92) # thumb tip
|
|
rect(165,100,235,122) # fingers
|
|
rect(165,125,225,145)
|
|
rect(165,148,215,168)
|
|
rect(165,171,205,188)
|
|
# white cuts between fingers
|
|
rect(166,122,225,126,0); rect(166,145,218,149,0); rect(166,168,210,172,0)
|
|
header = f'P4\n{W} {H}\n'.encode()
|
|
body = bytearray()
|
|
for y in range(H):
|
|
for x0 in range(0,W,8):
|
|
b=0
|
|
for i in range(8):
|
|
if x0+i < W and pix[y][x0+i]: b |= 1 << (7-i)
|
|
body.append(b)
|
|
return base64.b64encode(header+body).decode()
|
|
|
|
def make_rgb565(w=200,h=160):
|
|
def rgb565(r,g,b): return ((r&248)<<8)|((g&252)<<3)|(b>>3)
|
|
bg=rgb565(20,35,70); yellow=rgb565(245,190,45); dark=rgb565(80,55,10)
|
|
pix=[bg]*(w*h)
|
|
def rect(x0,y0,x1,y1,c):
|
|
for y in range(max(0,y0), min(h,y1)):
|
|
off=y*w
|
|
for x in range(max(0,x0), min(w,x1)): pix[off+x]=c
|
|
rect(28,90,68,135,yellow); rect(65,58,105,135,yellow); rect(90,35,120,75,yellow); rect(112,22,142,50,yellow)
|
|
rect(103,62,170,80,yellow); rect(103,84,162,102,yellow); rect(103,106,154,124,yellow); rect(103,128,146,146,yellow)
|
|
rect(103,81,170,84,dark); rect(103,103,162,106,dark); rect(103,125,154,128,dark)
|
|
# big-endian bytes
|
|
raw=bytearray()
|
|
for p in pix: raw += struct.pack('>H', p)
|
|
return base64.b64encode(raw).decode(), w, h
|
|
|
|
def make_bmp(w=200,h=160):
|
|
# 24-bit BMP, bottom-up, blue background with yellow thumb
|
|
rowpad=(4-(w*3)%4)%4
|
|
data=bytearray()
|
|
def is_hand(x,y):
|
|
return (28<=x<68 and 90<=y<135) or (65<=x<105 and 58<=y<135) or (90<=x<120 and 35<=y<75) or (112<=x<142 and 22<=y<50) or (103<=x<170 and 62<=y<80) or (103<=x<162 and 84<=y<102) or (103<=x<154 and 106<=y<124) or (103<=x<146 and 128<=y<146)
|
|
for y in range(h-1,-1,-1):
|
|
for x in range(w):
|
|
if is_hand(x,y): data += bytes([45,190,245])
|
|
else: data += bytes([70,35,20])
|
|
data += b'\x00'*rowpad
|
|
size=54+len(data)
|
|
header=b'BM'+struct.pack('<IHHI',size,0,0,54)+struct.pack('<IiiHHIIiiII',40,w,h,1,24,0,len(data),2835,2835,0,0)
|
|
return base64.b64encode(header+data).decode(), w, h
|
|
|
|
PBM=make_pbm(); RGB,W2,H2=make_rgb565(); BMP,WB,HB=make_bmp()
|
|
urls = candidate_urls()
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=32) as ex:
|
|
probed = [r for r in ex.map(probe, urls) if r]
|
|
# De-dupe IP:port and keep at most the first 3 live screen/MCP devices, matching the request.
|
|
seen=set(); targets=[]
|
|
for url,names in probed:
|
|
key=re.sub(r'^http://','',url).split('/')[0]
|
|
if key not in seen:
|
|
seen.add(key); targets.append((url,names))
|
|
if len(targets)>=3: break
|
|
|
|
lines=[f'Found {len(targets)} target(s): '+', '.join(u for u,_ in targets)]
|
|
for url,names in targets:
|
|
ok=False; detail=''
|
|
try: call(url,'clear_screen',{'color':0},timeout=3)
|
|
except Exception: pass
|
|
attempts=[]
|
|
if 'draw_raw_rgb565' in names:
|
|
attempts.append(('draw_raw_rgb565', {'rgb565_base64':RGB,'x':60,'y':40,'w':W2,'h':H2}))
|
|
if 'draw_color_bmp' in names:
|
|
attempts.append(('draw_color_bmp', {'bmp_base64':BMP,'x':60,'y':40}))
|
|
if 'draw_image' in names:
|
|
attempts.append(('draw_image', {'pbm_base64':PBM,'x':0,'y':0}))
|
|
attempts.append(('draw_image', {'image_base64':BMP,'x':60,'y':40,'dither':True}))
|
|
for name,args in attempts:
|
|
try:
|
|
res=call(url,name,args,timeout=9)
|
|
if 'error' not in res:
|
|
ok=True; detail=f'{name} ok'; break
|
|
detail=f'{name} error: {res.get("error")}'
|
|
except Exception as e:
|
|
detail=f'{name} exception: {e}'
|
|
if not ok and 'draw_text' in names:
|
|
try:
|
|
res=call(url,'draw_text',{'text':'👍','x':130,'y':105,'size':4},timeout=4)
|
|
ok='error' not in res; detail='fallback draw_text 👍' if ok else str(res.get('error'))
|
|
except Exception as e: detail=f'fallback exception: {e}'
|
|
lines.append(('OK ' if ok else 'FAIL ')+url+' '+detail)
|
|
|
|
OUT.write_text('\n'.join(lines)+'\n')
|
|
print('\n'.join(lines))
|
|
if not targets or any(l.startswith('FAIL') for l in lines):
|
|
raise SystemExit(1)
|