| 123456789101112131415161718192021222324252627282930 |
- """Disposable UI fixture; use distinct ports for concurrent native clients."""
- import argparse, base64, json, time, threading
- from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
- from pathlib import Path
- parser=argparse.ArgumentParser();parser.add_argument('--port',type=int,required=True);parser.add_argument('--image',required=True);args=parser.parse_args()
- image=base64.b64encode(Path(args.image).read_bytes()).decode()
- state={'mode':'agent','generation':1,'controlId':None,'actions':[],'notes':''}
- lock=threading.Lock()
- class Handler(BaseHTTPRequestHandler):
- def log_message(self,*args): pass
- def respond(self,value):
- data=json.dumps(value,ensure_ascii=False).encode();self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
- def do_GET(self):
- with lock:
- if self.path=='/fixture/state': return self.respond(state)
- if self.path.endswith('/frame'): return self.respond({'id':'22222222-2222-4222-8222-222222222222','generation':state['generation'],'data':image,'width':512,'height':512,'capturedAt':int(time.time()*1000)})
- if self.path.endswith('/computer'): return self.respond({**state,'canResume':True})
- return self.respond({'ok':True,'csrfToken':'fixture','features':['computerControl']})
- def do_POST(self):
- data=json.loads(self.rfile.read(int(self.headers.get('Content-Length','0'))) or b'{}')
- with lock:
- if self.path=='/fixture/reset': state.update(mode='agent',generation=1,controlId=None,actions=[],notes='')
- elif self.path.endswith('/take'):
- state.update(mode='human',generation=state['generation']+1,controlId='11111111-1111-4111-8111-111111111111')
- return self.respond({**state,'token':'fixture-private-control-token-1234567890','canResume':True})
- elif self.path.endswith('/input'): state['actions'].append(data.get('action'))
- elif self.path.endswith('/giveback'): state.update(mode='agent',generation=state['generation']+1,controlId=None,notes=data.get('notes',''))
- return self.respond({'ok':True})
- print(f'native computer fixture on 127.0.0.1:{args.port}',flush=True)
- ThreadingHTTPServer(('127.0.0.1',args.port),Handler).serve_forever()
|
