AlmostFeat : replacing emitter / receiver by fifo. Not working

Author: Cbampeta

scripts/launch_train_multiprocessing.py

@@ -14,6 +14,7 @@
 
 from config import *
 from TemporalResNetExtractor import TemporalResNetExtractor
+from CNN1DResNetExtractor import CNN1DResNetExtractor
 from onnx_utils import *
 
 from WebotsSimulationGymEnvironment import WebotsSimulationGymEnvironment
@@ -39,7 +40,7 @@ def make_env(simulation_rank: int, vehicle_rank: int):
 
     envs = SubprocVecEnv([lambda simulation_rank=simulation_rank, vehicle_rank=vehicle_rank : make_env(simulation_rank, vehicle_rank) for vehicle_rank in range(n_vehicles) for simulation_rank in range(n_simulations)])
 
-    ExtractorClass = TemporalResNetExtractor
+    ExtractorClass = CNN1DResNetExtractor
 
     policy_kwargs = dict(
         features_extractor_class=ExtractorClass,
@@ -128,6 +129,8 @@ def make_env(simulation_rank: int, vehicle_rank: int):
         else:
             model.learn(total_timesteps=500_000)
 
+        print("iteration over")
+
         model.save(save_path + str(i))
 
         i += 1

src/Simulateur/WebotsSimulationGymEnvironment.py

@@ -36,17 +36,18 @@ def __init__(self, simulation_rank: int, vehicle_rank: int):
 
         os.mkfifo(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}toserver.pipe")
         os.mkfifo(f"/tmp/autotech/serverto{simulation_rank}_{vehicle_rank}.pipe")
+        os.mkfifo(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}tosupervisor.pipe")
 
         #  --mode=fast --minimize --no-rendering --batch --stdout
         os.system(f"""
             webots {__file__.rsplit('/', 1)[0]}/worlds/piste{simulation_rank % n_map}.wbt --mode=fast --minimize --no-rendering --batch --stdout &
             echo $! {simulation_rank}_{vehicle_rank} >>/tmp/autotech/simulationranks
         """)
-        log(f"SERVER{simulation_rank}_{vehicle_rank} : {simulation_rank}_{vehicle_rank}toserver.pipe")
-        self.fifo_r = open(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}toserver.pipe", "rb")
         log(f"SERVER{simulation_rank}_{vehicle_rank} : serverto{simulation_rank}_{vehicle_rank}.pipe")
         self.fifo_w = open(f"/tmp/autotech/serverto{simulation_rank}_{vehicle_rank}.pipe", "wb")
-        log(f"SERVER{simulation_rank}_{vehicle_rank} : fifo opened :D and init done")
+        log(f"SERVER{simulation_rank}_{vehicle_rank} : {simulation_rank}_{vehicle_rank}toserver.pipe")
+        self.fifo_r = open(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}toserver.pipe", "rb")
+        
         log("-------------------------------------------------------------------")
 
     def reset(self, seed=0):
@@ -56,6 +57,7 @@ def reset(self, seed=0):
         # this is true for lidar_horizontal_resolution = camera_horizontal_resolution
         self.context = obs = np.zeros([2, context_size, lidar_horizontal_resolution], dtype=np.float32)
         info = {}
+        log(f"SERVER{self.simulation_rank}_{self.vehicle_rank} : finished reset")
         return obs, info
 
     def step(self, action):
@@ -64,6 +66,7 @@ def step(self, action):
         self.fifo_w.flush()
 
         # communication with the supervisor
+        log(f"SERVER{self.simulation_rank}_{self.vehicle_rank} : trying to read the fifo_r")
         cur_state   = np.frombuffer(self.fifo_r.read(np.dtype(np.float32).itemsize * (n_sensors + lidar_horizontal_resolution + camera_horizontal_resolution)), dtype=np.float32)
         log(f"SERVER{self.simulation_rank}_{self.vehicle_rank} : received {cur_state=}")
         reward      = np.frombuffer(self.fifo_r.read(np.dtype(np.float32).itemsize), dtype=np.float32)[0] # scalar
@@ -79,17 +82,13 @@ def step(self, action):
         lidar_obs = cur_state[:lidar_horizontal_resolution]
         camera_obs = cur_state[lidar_horizontal_resolution:]
 
-        # apply dropout to the camera
-        # p = 0.5
-        # camera_obs *= np.random.binomial(1, 1-p, camera_obs.shape) # random values in {0, 1}
-
         self.context = obs = np.concatenate([
             self.context[:, 1:],
             [lidar_obs[None], camera_obs[None]]
         ], axis=1)
-        # check if the context is correct
-        # if self.simulation_rank == 0:
-        #     print(f"{(obs[0] == 0).mean():.3f} {(obs[1] == 0).mean():.3f}")
+
+        log(f"SERVER{self.simulation_rank}_{self.vehicle_rank} : step over")
+
         return obs, reward, done, truncated, info
 
 

src/Simulateur/config.py

@@ -11,8 +11,8 @@
 lidar_max_range = 12.0
 device = "cuda" if is_available() else "cpu"
 
-context_size = 128
-lidar_horizontal_resolution = 128 # DON'T CHANGE THIS VALUE PLS
-camera_horizontal_resolution = 128 # DON'T CHANGE THIS VALUE PLS
+context_size = 1
+lidar_horizontal_resolution = 1024 # DON'T CHANGE THIS VALUE PLS
+camera_horizontal_resolution = 1024 # DON'T CHANGE THIS VALUE PLS
 
-B_DEBUG = False
+B_DEBUG = True

src/Simulateur/controllers/controllerVehicleDriver/controllerVehicleDriver.py

@@ -1,9 +1,23 @@
 import numpy as np
+import psutil
 import time
+import os
+import re 
+import sys
 
+# add src/Simulateur to sys.path
+path = __file__.rsplit('/', 3)[0]
+sys.path.insert(0, path)
+
+from config import *
 from vehicle import Driver
 
 
+
+def log(s: str):
+    if True:
+        print(s, file=open("/tmp/autotech/logs", "a"))
+
 class VehicleDriver(Driver):
     """
     This class is a subclass of the Driver class and is used to control the vehicle.
@@ -16,8 +30,10 @@ def __init__(self):
         basicTimeStep = int(self.getBasicTimeStep())
         self.sensorTime = basicTimeStep // 4
 
-        self.i = int(self.getName().split("_")[-1])
+        self.v_min = 1 
+        self.v_max = 9 
 
+        self.i = int(self.getName().split("_")[-1])
         # Lidar
         self.lidar = self.getDevice("Hokuyo")
         self.lidar.enable(self.sensorTime)
@@ -32,14 +48,28 @@ def __init__(self):
         self.touch_sensor.enable(self.sensorTime)
 
         # Communication
-        self.receiver = self.getDevice("TT02_receiver")
-        self.receiver.enable(self.sensorTime)
-        self.receiver.setChannel(2 * self.i) # corwe ponds the the supervisor's emitter channel
-        self.emitter = self.getDevice("TT02_emitter")
-        self.emitter.setChannel(2 * self.i + 1) # corresponds the the supervisor's receiver channel
 
-        # Last data received from the supervisor (steering angle)
-        self.last_data = np.zeros(2, dtype=np.float32)
+        proc = psutil.Process(os.getpid()) #current
+        parent = proc.parent() #parent
+        grandparent = parent.parent() if parent else None #grandparent
+        pppid = str(grandparent.pid)
+
+
+        self.simulation_rank = int(
+            re.search(
+                pppid + r" (\d+)",
+                open("/tmp/autotech/simulationranks", "r").read(),
+                re.MULTILINE
+            ).group(1)
+        )
+
+        log(f"CLIENT{self.simulation_rank}/{self.i} : serverto{self.simulation_rank}_{self.i}.pipe")
+        self.fifo_r = open(f"/tmp/autotech/serverto{self.simulation_rank}_{self.i}.pipe", "rb")
+        log(f"CLIENT{self.simulation_rank}/{self.i} : {self.simulation_rank}_{self.i}tosupervisor.pipe")
+        self.fifo_w = open(f"/tmp/autotech/{self.simulation_rank}_{self.i}tosupervisor.pipe", "wb")
+
+
+
 
     #Vérification de l"état de la voiture
     def observe(self):
@@ -78,14 +108,17 @@ def observe(self):
     def step(self):
         # sends observation to the supervisor
 
-        self.emitter.send(self.observe().tobytes())
+        # First to be executed
+        self.fifo_w.write(self.observe().tobytes())
+        
+        log(f"CLIENT{self.simulation_rank}/{self.i} : trying to read from fifo")    
+        action = np.frombuffer(self.fifo_r.read(np.dtype(np.int64).itemsize * 2), dtype=np.int64)
+        log(f"CLIENT{self.simulation_rank}/{self.i} : received {action=}")
 
-        if self.receiver.getQueueLength() > 0:
-            while self.receiver.getQueueLength() > 1:
-                self.receiver.nextPacket()
-            self.last_data = np.frombuffer(self.receiver.getBytes(), dtype=np.float32)
+        # Simulation step
 
-        action_steering, action_speed = self.last_data
+        action_steering = np.linspace(-.4, .4, n_actions_steering, dtype=np.float32)[action[0], None]
+        action_speed = np.linspace(self.v_min, self.v_max, n_actions_speed, dtype=np.float32)[action[1], None]
 
         cur_angle = self.getSteeringAngle()
         dt = self.getBasicTimeStep()

src/Simulateur/controllers/controllerWorldSupervisor/controllerWorldSupervisor.py

@@ -7,7 +7,6 @@
 
 from controller import Supervisor
 supervisor = Supervisor()
-
 import torch.nn as nn
 
 import psutil
@@ -82,7 +81,7 @@ def log(s: str):
 
 
 
-class WebotsVehicleGymEnvironment(gym.Env):
+class WebotsVehicleManager:
     """
     One environment for each vehicle
 
@@ -102,37 +101,36 @@ def __init__(self, vehicle_rank: int):
         # negative value so that the first reset is not skipped
         self.last_reset = -1e6
 
-        # Emitter
-        self.emitter = supervisor.getDevice(f"supervisor_emitter_{vehicle_rank}")
-        self.emitter.setChannel(2 * self.vehicle_rank)
+        proc = psutil.Process(os.getpid()) #current
+        parent = proc.parent() #parent
+        grandparent = parent.parent() if parent else None #grandparent
+        pppid = str(grandparent.pid)
 
-        # Receiver
-        self.receiver = supervisor.getDevice(f"supervisor_receiver_{vehicle_rank}")
-        self.receiver.enable(self.sensorTime)
-        self.receiver.setChannel(2 * self.vehicle_rank + 1)
 
+        self.simulation_rank = int(
+            re.search(
+                pppid + r" (\d+)",
+                open("/tmp/autotech/simulationranks", "r").read(),
+                re.MULTILINE
+            ).group(1)
+        )
 
-        log(f"CLIENT{simulation_rank}/{vehicle_rank}_{vehicle_rank} : begins init")
-        log(f"CLIENT{simulation_rank}/{vehicle_rank}_{vehicle_rank} : {simulation_rank}_{vehicle_rank}toserver.pipe")
-        self.fifo_w = open(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}toserver.pipe", "wb")
-        log(f"CLIENT{simulation_rank}/{vehicle_rank} : serverto{simulation_rank}_{vehicle_rank}.pipe")
-        self.fifo_r = open(f"/tmp/autotech/serverto{simulation_rank}_{vehicle_rank}.pipe", "rb")
 
-        # Last data received from the car
-        self.last_data = np.zeros(n_sensors + lidar_horizontal_resolution + camera_horizontal_resolution, dtype=np.float32)
+        log(f"SUPERVISOR{simulation_rank}/{vehicle_rank}_{vehicle_rank} : begins init")
+        log(f"SUPERVISOR{simulation_rank}/{self.vehicle_rank} : {self.simulation_rank}_{self.vehicle_rank}tosupervisor.pipe")
+        self.fifo_r = open(f"/tmp/autotech/{self.simulation_rank}_{self.vehicle_rank}tosupervisor.pipe", "rb")
+        log(f"SUPERVISOR{simulation_rank}/{vehicle_rank}_{vehicle_rank} : {simulation_rank}_{vehicle_rank}toserver.pipe")
+        self.fifo_w = open(f"/tmp/autotech/{simulation_rank}_{vehicle_rank}toserver.pipe", "wb")
+        
 
         self.translation_field = supervisor.getFromDef(f"TT02_{self.vehicle_rank}").getField("translation") # may cause access issues ...
         self.rotation_field = supervisor.getFromDef(f"TT02_{self.vehicle_rank}").getField("rotation") # may cause access issues ...
 
     # returns the lidar data of all vehicles
     def observe(self):
-        # gets from Receiver
-        if self.receiver.getQueueLength() > 0:
-            while self.receiver.getQueueLength() > 1:
-                self.receiver.nextPacket()
-            self.last_data = np.frombuffer(self.receiver.getBytes(), dtype=np.float32)
+        # gets from Vehicle
 
-        return self.last_data
+        return np.frombuffer(self.fifo_r.read(np.dtype(np.float32).itemsize * (n_sensors + lidar_horizontal_resolution + camera_horizontal_resolution)), dtype=np.float32)
 
     # reset the gym environment reset
     def reset(self, seed=None):
@@ -158,10 +156,9 @@ def reset(self, seed=None):
         return obs, info
 
     # step function of the gym environment
-    def step(self, action):
-        action_steering = np.linspace(-.4, .4, n_actions_steering, dtype=np.float32)[action[0], None]
-        action_speed = np.linspace(self.v_min, self.v_max, n_actions_speed, dtype=np.float32)[action[1], None]
-        self.emitter.send(np.array([action_steering, action_speed], dtype=np.float32).tobytes())
+    def step(self):
+        #action_steering = np.linspace(-.4, .4, n_actions_steering, dtype=np.float32)[action[0], None]
+        #action_speed = np.linspace(self.v_min, self.v_max, n_actions_speed, dtype=np.float32)[action[1], None]
 
         # we should add a beacon sensor pointing upwards to detect the beacon
         obs = self.observe()
@@ -191,7 +188,7 @@ def step(self, action):
 
 def main():
 
-    envs = [WebotsVehicleGymEnvironment(i) for i in range(n_vehicles)]
+    envs = [WebotsVehicleManager(i) for i in range(n_vehicles)]
     log(f"CLIENT ALL : envs created")
     # check_env(env)
 
@@ -218,22 +215,17 @@ def main():
         log(f"CLIENT ALL : begin step")
         #Prédiction pour séléctionner une action à partir de l"observation
         for e in envs:
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : trying to read from fifo")
-            action = np.frombuffer(e.fifo_r.read(np.dtype(np.int64).itemsize * 2), dtype=np.int64)
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : received {action=}")
-
-            obs, reward, done, truncated, info = e.step(action)
-
+            obs, reward, done, truncated, info = e.step()
             if done:
                 obs, info = e.reset()
 
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : sending {obs=}")
+            log(f"SUPERVISOR{simulation_rank}/{e.vehicle_rank} : sending {obs=}")
             e.fifo_w.write(obs.tobytes())
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : sending {reward=}")
+            log(f"SUPERVISOR{simulation_rank}/{e.vehicle_rank} : sending {reward=}")
             e.fifo_w.write(reward.tobytes())
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : sending {done=}")
+            log(f"SUPERVISOR{simulation_rank}/{e.vehicle_rank} : sending {done=}")
             e.fifo_w.write(done.tobytes())
-            log(f"CLIENT{simulation_rank}/{e.vehicle_rank} : sending {truncated=}")
+            log(f"SUPERVISOR{simulation_rank}/{e.vehicle_rank} : sending {truncated=}")
             e.fifo_w.write(truncated.tobytes())
             e.fifo_w.flush()