#!/usr/bin/env python3 # Import modules import os # Prevent requests from loading optional tools which slow down start os.environ['PYTHONHTTPSVERIFY'] = '0' os.environ['REQUESTS_CA_BUNDLE'] = '' os.environ['SSL_CERT_FILE'] = '' import time, requests, random from ev3dev2.motor import LargeMotor, MediumMotor, MoveSteering, MoveTank, OUTPUT_A, OUTPUT_B, OUTPUT_C, OUTPUT_D, SpeedPercent from ev3dev2.sensor import INPUT_4, Sensor from ev3dev2.sensor.lego import UltrasonicSensor # Initialize program constants SERVER = "http://192.168.0.129:5000" DEVICE_NAME = "REXIE" POLL_INTERVAL = 1.0 HEARTBEAT_INTERVAL = 2.0 SCAN_ANGLE = 40 # IR sensor turn angle OBSTACLE_DISTANCE = 150 # mm # ------------------- # SERVER CLIENT class # ------------------- class ServerClient: # Initialize self def __init__(self, server, device): print("[SERVERCLIENT] Initializing") self.server = server self.device = device self.requests = requests # Send event def send_event(self, kind, detail): # Populate JSON payload payload = { "device": self.device, "type": kind, "value": detail } # Try sending JSON event to /event endpoint on Flask server try: r = self.requests.post(self.server + "/event", json=payload, timeout=POLL_INTERVAL, headers={"Connection": "close"} ) r.close() print("[EVENT]", payload) except Exception as e: print("[HTTP] Sending event failed with error '", e, "'") time.sleep(0.2) # Receive command def get_command(self): # Try receiving JSON command from /sent_command/ endpoint on Flask server try: r = self.requests.get(self.server + "/sent_command/" + self.device, timeout=POLL_INTERVAL, headers={"Connection": "close"} ) data = r.json() r.close() return data.get("command sent") except Exception as e: print("[HTTP] Receiving command failed with error '", e, "'") time.sleep(0.2) return None # -------------------- # HARDWARE LAYER class # -------------------- class Hardware: # Initialize self def __init__(self): print("[HARDWARE] Initializing") # MODIFY TO SUIT REQUIRED HARDWARE #self.ev3 = EV3Brick() # Initialize robot constants self.axle_track = 150 # mm distance between middle of track contact self.wheel_diameter = 45 # mm # Initialize motors self.motor_jaw = MediumMotor(OUTPUT_A) self.motor_left = LargeMotor(OUTPUT_B) self.motor_right = LargeMotor(OUTPUT_C) self.motor_sensor = MediumMotor(OUTPUT_D) self.move_tank = MoveTank(OUTPUT_B, OUTPUT_C) self.move_steering = MoveSteering(OUTPUT_B, OUTPUT_C) self.motor_sensor.reset() # Initialize sensors self.sensor_ir = Sensor(address='in1') # IR thermometer self.sensor_us = UltrasonicSensor(INPUT_4) def forward(self): print("[HW] Running command forward") def backward(self): print("[HW] Running command backward") def start(self): print("[HW] Running command start") def stop(self): print("[HW] Running command stop") def beep(self): print("[HW] Running command beep") # ------------------ # DEVICE LOGIC class # ------------------ class Device: # Initialize self def __init__(self, name, server): self.name = name self.server = ServerClient(server, name) self.hw = Hardware() self.last_heartbeat = 0 self.start_received = False self.waiting = False # Attack obstacle def attack_obstacle(self): distance_cm = self.hw.sensor_us.distance_centimeters if distance_cm is None: return False # Treat as no obstacle distance = distance_cm * 10 jaw = self.hw.motor_jaw tank = self.hw.move_tank if distance < OBSTACLE_DISTANCE: print("[OBS] Obstacle detected at", distance, "mm") msg = "Prey:" + str(distance) + "mm" self.server.send_event("STATUS", msg) self.server.send_event("EVENT", "01") # Charge forward tank.on_for_seconds(SpeedPercent(-60), SpeedPercent(-60), 1) # Jaw snap # jaw.on_for_seconds(SpeedPercent(80), 0.2) # jaw.on_for_seconds(SpeedPercent(-80), 0.2) jaw.on_to_position(SpeedPercent(80), -90) jaw.on_to_position(SpeedPercent(80), 0) self.server.send_event("EVENT", "04") return True return False # Handle command sent by Flask server def handle_command(self, cmd): print("[CMD] Receiving command '", cmd, "'") # MODIFY TO SUIT REQUIRED COMMANDS if cmd == "forward": self.hw.forward() elif cmd == "backward": self.hw.backward() elif cmd == "start": #self.hw.start() # If command is 'start', allow code to proceed to periodic() self.start_received = True self.waiting = False elif cmd == "stop": #self.hw.stop() self.server.send_event("EVENT", "02") self.waiting = True elif cmd == "beep": self.hw.beep() else: print("[CMD] Receiving unknown command '", cmd, "'") # Send status 'alive' to Flask server def heartbeat(self): self.server.send_event("STATUS", "alive") # Scan left, centre, right def heat_seek_step(self): left = self.hw.motor_left right = self.hw.motor_right tank = self.hw.move_tank jaw = self.hw.motor_jaw centre_temp = self.scan_direction(0) left_temp = self.scan_direction(-SCAN_ANGLE) right_temp = self.scan_direction(SCAN_ANGLE) # Find hottest direction temps = { "centre": centre_temp, "left": left_temp, "right": right_temp } direction = max(temps, key=temps.get) hottest = temps[direction] print("[SCAN] Temperatures found", temps, ". Hottest at", direction, hottest) msg = "Temp.:" + str(hottest) + "degC, Dir.:" + direction self.server.send_event("STATUS", msg) # Jaw animation # jaw.on_for_seconds(SpeedPercent(40), 0.2) # jaw.on_for_seconds(SpeedPercent(-40), 0.2) jaw.on_to_position(SpeedPercent(40), -90) jaw.on_to_position(SpeedPercent(40), 0) # Move based on direction if direction == "right": tank.on_for_seconds(SpeedPercent(-20), SpeedPercent(20), 0.3) elif direction == "left": tank.on_for_seconds(SpeedPercent(20), SpeedPercent(-20), 0.3) else: # Move forward toward heat tank.on_for_seconds(SpeedPercent(-30), SpeedPercent(-30), 1) # Define functions run periodically def periodic(self): # MODIFY TO SUIT REQUIRED CODE # If waiting, growl periodically if self.waiting: value = random.randint(1, 5) if value == 1: self.server.send_event("EVENT", "03") else: # Read ambient temperature ambient = self.read_ambient() # Attack takes priority if self.attack_obstacle(): return # Otherwise heat-seek self.heat_seek_step() # Scan def scan_direction(self, angle): scanner = self.hw.motor_sensor scanner.on_to_position(SpeedPercent(20), angle, brake=True, block=True) time.sleep(0.2) temperature = self.read_target() # Reset to centre scanner.on_to_position(SpeedPercent(20), 0, brake=True, block=True) return temperature # Send event 'ready' to Flask server def startup(self): print("[DEVICE] Starting up") self.server.send_event("EVENT", "ready") # Read ambient temperature def read_ambient(self): sensor = self.hw.sensor_ir sensor.mode = 'AMBIENT-C' time.sleep(0.05) ambient = sensor.value(0) / 100.0 print("[SCAN] Ambient temperature", ambient, "degC") return ambient # Read target temperature def read_target(self): sensor = self.hw.sensor_ir sensor.mode = 'TARGET-C' time.sleep(0.05) target = sensor.value(0) / 100.0 print("[SCAN] Target temperature", target, "degC") return target # Define main function def run(self): # Send event self.startup() last_cmd = None while True: # Send status now = time.time() if now - self.last_heartbeat > HEARTBEAT_INTERVAL: self.heartbeat() self.last_heartbeat = now # Receive command cmd = self.server.get_command() if cmd: if cmd != last_cmd: self.handle_command(cmd) # Send event command consumed to Flask server self.server.send_event("CMD_ACK", cmd) last_cmd = cmd else: last_cmd = None #self.periodic() if self.start_received: self.periodic() time.sleep(POLL_INTERVAL) # Declare main program if __name__ == "__main__": device = Device( DEVICE_NAME, SERVER ) device.run()