#!/usr/bin/env python3 """ control server for roomba 69X series, make venv or uv it, pip stuff, run - I haven't solved the waking part but may just use a servo oh, I ran this on an orangpi3b, ymmv when it comes to TTYusb, this was RC for Bookworm and had an unstable data.stream handling creating a situation to open the port, then wait, then hopefully it works future - add servo, and keep wake commands in every X heatmap in progress for any kind of useful mapping """ import os, time, threading, base64, struct, serial, sys import cv2 import numpy as np from flask import Flask, jsonify, request print("Server starting... waiting 2 seconds for hardware to stabilize...") time.sleep(2) # --- CONFIG --- ROOMBA_PORT = "/dev/ttyUSB0"; BAUD_RATE = 115200; WIDTH, HEIGHT= 320, 320; JPEG_Q = 40; DRIVE_SPEED = 150; TURN_SPEED = 100 OP = {"START":bytes([128]),"SAFE":bytes([131]),"FULL":bytes([132]),"STOP_STREAM":bytes([150,0]),"DRIVE_DIRECT":145,"SENSORS":142,"MOTORS":138} app = Flask(__name__); lock = threading.Lock(); ser, cam = None, None # --- INIT --- try: print(f"Opening serial port {ROOMBA_PORT}..."); ser = serial.Serial(ROOMBA_PORT, BAUD_RATE, timeout=1) print("Resetting Roomba state..."); ser.write(OP["STOP_STREAM"]); time.sleep(0.2); ser.write(OP["START"]); time.sleep(0.2); ser.write(OP["SAFE"]); time.sleep(0.2) ser.reset_input_buffer(); print("OOOO Roomba is awake, silent, and ready for commands.") except Exception as e: print(f"XXXX Roomba init failed: {e}"); ser = None def find_and_init_camera(): for i in range(10): print(f"Initializing camera index {i}...") temp_cam = cv2.VideoCapture(i) if temp_cam and temp_cam.isOpened(): temp_cam.set(cv2.CAP_PROP_FRAME_WIDTH, WIDTH); temp_cam.set(cv2.CAP_PROP_FRAME_HEIGHT, HEIGHT); time.sleep(1.5) print(f"OOOO Camera initialized successfully on index {i}."); return temp_cam return None try: cam = find_and_init_camera() if cam is None: raise RuntimeError("XXXXX Could not find any working camera.") except Exception as e: print(f"XXXX Camera init failed: {e}"); cam = None # --- API EPs --- @app.route('/image') def image(): if not cam: return jsonify(error='camera-offline'), 503 with lock: for _ in range(3): cam.read() ok, frame = cam.read() if not ok: return jsonify(error='frame-grab-failed'), 500 # Overlay overlay = frame.copy() # Define Regions # CLOSE (Lower 30% of the image) - DANGER ZONE cv2.rectangle(overlay, (0, int(HEIGHT * 0.7)), (WIDTH, HEIGHT), (0, 0, 255), -1) # Red # MEDIUM (Middle 40-60%ish) - TARGET ZONE cv2.rectangle(overlay, (0, int(HEIGHT * 0.3)), (WIDTH, int(HEIGHT * 0.7)), (0, 255, 0), -1) # Green # Overlay merge to frame alpha = 0.15 # Transparency factor - review later frame_with_hud = cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0) # Add text labels cv2.putText(frame_with_hud, 'CLOSE (DANGER)', (10, HEIGHT - 15), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255,255,255), 1) cv2.putText(frame_with_hud, 'MEDIUM (TARGET)', (10, int(HEIGHT * 0.7) - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255,255,255), 1) # --- END OF HUD --- ok, buf = cv2.imencode('.jpg', frame_with_hud, [cv2.IMWRITE_JPEG_QUALITY, JPEG_Q]) if not ok: return jsonify(error='jpeg-encoding-failed'), 500 return jsonify(image_base64=base64.b64encode(buf).decode('utf-8')) @app.route('/status') def status(): if not ser: return jsonify(error='roomba-offline'), 503 with lock: try: ser.reset_input_buffer(); ser.write(bytes([OP["SENSORS"], 21])); time.sleep(0.05) data = ser.read(1); is_docked = (data[0] != 0) if data else False return jsonify(battery_percent=100.0, docked=is_docked) except Exception as e: return jsonify(battery_percent=100.0, docked=True, error=str(e)) @app.route('/undock', methods=['POST']) def undock(): if not ser: return jsonify(error='roomba-offline'), 503 with lock: try: ser.write(OP["FULL"]); time.sleep(0.2) drive_pkt = struct.pack('>Bhh', OP["DRIVE_DIRECT"], -DRIVE_SPEED, -DRIVE_SPEED); ser.write(drive_pkt); time.sleep(1.5) stop_pkt = struct.pack('>Bhh', OP["DRIVE_DIRECT"], 0, 0); ser.write(stop_pkt) ser.write(OP["SAFE"]); time.sleep(0.2) return jsonify(status="undock sequence complete") except Exception as e: return jsonify(error=str(e)), 500 @app.route('/vacuum', methods=['POST']) def vacuum(): if not ser: return jsonify(error='roomba-offline'), 503 with lock: try: data=request.get_json() or {}; is_on=data.get("on", False) motors_byte = 0b00000111 if is_on else 0b00000000 vacuum_pkt = struct.pack('BB', OP["MOTORS"], motors_byte); ser.write(vacuum_pkt) print(f"Vacuum motors turned {'ON' if is_on else 'OFF'}") return jsonify(status=f"vacuum {'on' if is_on else 'off'}") except Exception as e: return jsonify(error=str(e)), 500 @app.route('/bumper') def get_bumper(): if not ser: return jsonify(error='roomba-offline'), 503 with lock: try: ser.reset_input_buffer(); ser.write(bytes([OP["SENSORS"], 7])); time.sleep(0.05) data = ser.read(1); is_pressed = bool(data[0] & 0b00000011) if data else False return jsonify(bumped=is_pressed) except Exception as e: return jsonify(bumped=False, error=str(e)) @app.route('/move', methods=['POST']) def move(): if not ser: return jsonify(error='roomba-offline'), 503 data=request.get_json() or {}; maneuver=data.get('maneuver','stop'); duration=float(data.get('duration',0)) mapping={'forward':(DRIVE_SPEED,DRIVE_SPEED),'backward':(-DRIVE_SPEED,-DRIVE_SPEED),'left':(-TURN_SPEED,TURN_SPEED),'right':(TURN_SPEED,-TURN_SPEED),'stop':(0,0)} right_vel, left_vel = mapping.get(maneuver,(0,0)) with lock: try: ser.write(OP["SAFE"]); time.sleep(0.05) drive_pkt = struct.pack('>Bhh', OP["DRIVE_DIRECT"], right_vel, left_vel); ser.write(drive_pkt) if duration > 0: time.sleep(duration) stop_pkt = struct.pack('>Bhh', OP["DRIVE_DIRECT"], 0, 0); ser.write(stop_pkt) return jsonify(status=f"executed {maneuver}") except Exception as e: return jsonify(error=str(e)), 500 if __name__ == '__main__': if ser is None or cam is None: print('XXXX Missing hardware; API will error out.') else: print('OOOO Server ready. Latest ADD - Vision HUD.') app.run(host='0.0.0.0', port=5000, threaded=True)