"""skyvison RC local receiver. Python 3.10+, stdlib dry run by default.""" import json, math, os, secrets, threading, time from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from pathlib import Path class Outputs: def __init__(self): self.enabled = os.getenv('RC_OUTPUTS') == 'enabled' self.adapter = os.getenv('RC_ADAPTER', 'dry-run') self.device = None if self.enabled: if self.adapter == 'serial': import serial self.device = serial.Serial(os.environ['RC_SERIAL_PORT'],115200,timeout=.1,write_timeout=.1) time.sleep(2) # Uno auto-reset on USB open. elif self.adapter == 'pca9685': import board, busio from adafruit_pca9685 import PCA9685 self.device=PCA9685(busio.I2C(board.SCL, board.SDA));self.device.frequency=50 else: raise ValueError('RC_OUTPUTS=enabled requires serial or pca9685 adapter') self.write(False,0,0) def write(self, armed, throttle, steering): # Only forward-only ESCs with a 1000 us stop pulse are supported. throttle_us=1000+round(throttle*1000) if armed else 1000 steering_us=1500+round(steering*500) if armed else 1500 if not self.enabled:return if self.adapter=='serial':self.device.write(f'{int(armed)},{throttle_us},{steering_us}\n'.encode()) else: self.device.channels[0].duty_cycle=round(throttle_us/20000*65535) self.device.channels[1].duty_cycle=round(steering_us/20000*65535) def close(self): self.write(False,0,0) if self.device is not None: if self.adapter=='pca9685': self.device.deinit() else:self.device.close() class Receiver: def __init__(self, outputs=None, clock=time.monotonic): self.outputs=outputs or Outputs();self.clock=clock;self.lock=threading.RLock() self.session='';self.seq=-1;self.last=0;self.armed=False;self.last_pair=0 def stop(self): self.armed=False;self.outputs.write(False,0,0) def tick(self): with self.lock: if self.clock()-self.last>.5:self.stop() def pair(self): with self.lock: # Do not let a second client take over an active link. if self.session and self.clock()-self.last<1:raise ValueError('Receiver is busy; stop/disconnect and wait one second') self.stop();self.session=secrets.token_hex(16);self.seq=-1;self.last=self.clock() return {'session':self.session,'outputs':'enabled' if self.outputs.enabled else 'dry-run','watchdogMs':500} def command(self, frame): with self.lock: self.tick() if not self.session or frame.get('session')!=self.session:raise ValueError('Session expired; pair again') if type(frame.get('seq')) is not int or frame['seq']<=self.seq:raise ValueError('Stale sequence') self.seq=frame['seq'] try: if frame.get('profile') not in ('car','boat'):raise ValueError('Aircraft profiles are simulator-only') for name in ('throttle','yaw','pitch','roll'): value=frame.get(name) if type(value) not in (int,float) or not math.isfinite(value) or not (-1<=value<=1):raise ValueError('Invalid channel') if frame['throttle']<0:raise ValueError('Reverse throttle is not supported') if type(frame.get('armed')) is not bool or type(frame.get('deadman')) is not bool:raise ValueError('Invalid arming state') if frame['armed'] and not self.armed and any(frame[name]!=0 for name in ('throttle','yaw','pitch','roll')):raise ValueError('Return all channels to neutral before arming') self.armed=frame['armed'];active=self.armed and frame['deadman'] self.outputs.write(active,frame['throttle'] if active else 0,frame['roll'] if active else 0) self.last=self.clock() return {'armed':self.armed,'seq':self.seq,'outputs':'enabled' if self.outputs.enabled else 'dry-run'} except Exception: self.stop();raise ROOT=Path(os.getenv('RC_WEB_ROOT',str(Path(__file__).resolve().parents[1]/'controller'))).resolve() def handler(receiver,key): class Handler(SimpleHTTPRequestHandler): def __init__(self,*args,**kwargs):super().__init__(*args,directory=str(ROOT),**kwargs) def end_headers(self): self.send_header('Cache-Control','no-store');self.send_header('X-Content-Type-Options','nosniff') self.send_header('Content-Security-Policy',"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'") super().end_headers() def response(self,status,body): data=json.dumps(body).encode();self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data) def do_POST(self): if self.path not in ('/api/session','/api/control'):return self.response(404,{'error':'Not found'}) # No CORS headers: cross-origin webpages cannot authenticate/send JSON commands. if not secrets.compare_digest(self.headers.get('X-RC-Key',''),key):return self.response(403,{'error':'Pairing key rejected'}) if self.headers.get('Content-Type','').split(';')[0]!='application/json':return self.response(415,{'error':'JSON required'}) try: n=int(self.headers.get('Content-Length','0')) if n<1 or n>2048:raise ValueError('Invalid frame length') self.connection.settimeout(.2) data=json.loads(self.rfile.read(n)) if not isinstance(data,dict):raise ValueError('Object required') result=receiver.pair() if self.path=='/api/session' else receiver.command(data) self.response(200,result) except (ValueError,TypeError,TimeoutError) as error:self.response(400,{'error':str(error)}) except Exception: with receiver.lock:receiver.stop() self.response(503,{'error':'Output adapter failed; stopped'}) return Handler def main(): key=os.getenv('RC_KEY') or secrets.token_urlsafe(18) if len(key)<12:raise ValueError('RC_KEY must have at least 12 characters') receiver=Receiver();port=int(os.getenv('RC_PORT','8080')) server=ThreadingHTTPServer((os.getenv('RC_BIND','0.0.0.0'),port),handler(receiver,key));server.daemon_threads=True closing=threading.Event() def watchdog(): while not closing.wait(.02): try:receiver.tick() except Exception:closing.set();os._exit(1) # external hardware watchdog is still required threading.Thread(target=watchdog,daemon=True).start() print(f'Controller: http://:{port} | pairing key: {key} | outputs: {"ENABLED" if receiver.outputs.enabled else "DRY RUN"}',flush=True) try:server.serve_forever() except KeyboardInterrupt:pass finally: closing.set();server.server_close();receiver.outputs.close() if __name__=='__main__':main()