Adolfo Reyna
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
import concurrent.futures, json, re, socket, time, urllib.request
|
|
|
|
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:
|
|
return json.loads(r.read().decode('utf-8','replace'))
|
|
|
|
def call(url, name, args, timeout=4):
|
|
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.35)
|
|
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.0
|
|
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 candidates():
|
|
pairs=discover_udp()
|
|
pairs.update({('127.0.0.1',8080),('192.168.68.123',80),('192.168.68.126',8080)})
|
|
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)
|
|
names=[t.get('name','') for t in res.get('result',{}).get('tools',[])]
|
|
if 'draw_text' in names:
|
|
return url,names
|
|
except Exception:
|
|
return None
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=16) as ex:
|
|
targets=[r for r in ex.map(probe, candidates()) if r]
|
|
seen=set(); uniq=[]
|
|
for url,names in targets:
|
|
key=re.sub(r'^http://','',url).split('/')[0]
|
|
if key not in seen:
|
|
seen.add(key); uniq.append((url,names))
|
|
if len(uniq)>=3: break
|
|
|
|
lines=[f'Found {len(uniq)} target(s): '+', '.join(u for u,_ in uniq)]
|
|
for url,names in uniq:
|
|
try:
|
|
if 'clear_screen' in names:
|
|
try: call(url,'clear_screen',{'color':0},timeout=3)
|
|
except Exception: pass
|
|
res=call(url,'draw_text',{'text':'Hi','x':115,'y':95,'size':5,'color':65535},timeout=5)
|
|
if 'error' in res:
|
|
res=call(url,'draw_text',{'text':'Hi','x':115,'y':95,'size':5},timeout=5)
|
|
if 'error' in res:
|
|
lines.append('FAIL '+url+' '+json.dumps(res.get('error')))
|
|
else:
|
|
lines.append('OK '+url+' draw_text Hi')
|
|
except Exception as e:
|
|
lines.append(f'FAIL {url} {type(e).__name__}: {e}')
|
|
|
|
output='\n'.join(lines)+'\n'
|
|
open('/tmp/esp32_hi_result.txt','w').write(output)
|
|
print(output, end='')
|
|
if not uniq or any(l.startswith('FAIL') for l in lines):
|
|
raise SystemExit(1)
|