Update app.py

app.py

@@ -76,28 +76,49 @@ class TrafficSimEnv(gym.Env):
         self.steps = 0
 
     def reset(self, seed=None, options=None):
+        """
+        Resets the environment and returns the initial observation and info.
+        """
         self.steps = 0
         self.current_signal = 30
         congestion_map = {"Low": 0, "Medium": 1, "High": 2}
         self.congestion_numeric = congestion_map.get(self.congestion_level, 0)
+
+        # Initial observation: [congestion level, current signal]
         obs = np.array([self.congestion_numeric, self.current_signal], dtype=np.float32)
-        return obs, {}  # Fixed format for DummyVecEnv
+
+        return obs, {}  # Returns observation and info (info can be empty)
+
 
     def step(self, action):
-        print("Step: ", action)
-        print("Obs before update: ", self.current_signal)
-        print("Action Taken: ", action)
-        signal_change = {0: -5, 1: 0, 2: +5}[action]
+        """
+        Takes an action and updates the environment state.
+        Returns 5 values: observation, reward, done, truncated, and info.
+        """
+
+        print(f"Action taken: {action}")
+        print(f"Signal before change: {self.current_signal}")
+        print(f"Congestion Level: {self.congestion_numeric}")
+        
+        # Signal changes based on the action
+        signal_change = {0: -5, 1: 0, 2: 5}[action]
         self.current_signal = max(20, min(60, self.current_signal + signal_change))
+
+        # Reward Calculation based on congestion level and signal
         reward = self._calculate_reward()
+
+        # Increment the step count
         self.steps += 1
-        done = self.steps >= 10
-        truncated = False  # Compatibility
+        done = self.steps >= 10  # End condition
+        truncated = False  # Default to False; can change based on custom conditions
+
+        # Ensure we always return 5 values: obs, reward, done, truncated, info
         obs = np.array([self.congestion_numeric, self.current_signal], dtype=np.float32)
-        info = {}  # Additional info (if needbe)
-        return obs, reward, done, truncated, info
+        print(f"Observation: {obs}")
 
+        info = {}  # Additional info (can remain empty or contain any useful data)
 
+        return obs, reward, done, truncated, info  # Must return 5 values here
 
     def _calculate_reward(self):
         if self.congestion_level == "High":
@@ -116,25 +137,37 @@ class TrafficSimEnv(gym.Env):
 # Prior to commit had a lot of errors regarding expected output errors
 def optimize_signal_rl(congestion_level):
     try:
+        # Create the environment with DummyVecEnv to wrap TrafficSimEnv
         env = DummyVecEnv([lambda: TrafficSimEnv(congestion_level)])
+
+        # Initialize PPO model (policy = "MlpPolicy", for multi-layer perceptron model)
         model = PPO("MlpPolicy", env, verbose=0)
+
+        # Train the model on the environment for 1000 timesteps
         model.learn(total_timesteps=1000)
-        obs = env.reset()
 
+        # Reset environment to start the simulation
+        obs, _ = env.reset()
+
+        # Loop through to simulate for 10 timesteps
         for _ in range(10):
             action, _ = model.predict(obs)
+            
+            # Step through the environment with the predicted action
             obs, reward, done, truncated, info = env.step(action)
-            logging.info(f"Step results: obs={obs}, reward={reward}, done={done}, info={info}")
 
-            obs = obs[0]  # Extract the first (and only) observation from the batch
-            reward = reward[0]  # Reward for the current action
-            done = done[0]  # Done flag for the current step
-            truncated = truncated[0]  # Truncated flag
+            # Extract the first value from each returned array (since env is wrapped in DummyVecEnv)
+            obs = obs[0]  # First observation (from batch)
+            reward = reward[0]  # Reward for the action (from batch)
+            done = done[0]  # Done flag (from batch)
+            truncated = truncated[0]  # Truncated flag (from batch)
 
-            if done or truncated:  # Check if episode ends (either 'done' or 'truncated')
+            # Stop when the environment signals that the episode is done or truncated
+            if done or truncated:
                 break
 
-        optimal_duration = int(obs[1]) if len(obs) > 1 else 30  # Get the optimal signal duration
+        # Extract the optimal signal duration (second observation value) from `obs`
+        optimal_duration = int(obs[1]) if len(obs) > 1 else 30  # Ensure the signal value is within range
         return f"Green for {optimal_duration}s, Red for {60 - optimal_duration}s"
     except Exception as e:
         logging.error(f"Error optimizing signal with RL: {e}")
@@ -142,6 +175,7 @@ def optimize_signal_rl(congestion_level):
 
 
 
+
 def process_traffic_image(image):
     """
     Orchestrates the traffic analysis workflow.