ObservationFunction changes not applied

[TGW795]

Hi.

I've been trying to writing a code to experiment with MARL using sumo-rl.parallel_env() and making my own ObservationFunction, but any changes I made do not apply. It was stated in README that we can use our original ObservationFunction by defining it in observations.py and passing it to the environment constructor, and I followed this flow. Is there anything I am doing wrong? (I'm using the example code of PettingZoo Multi-Agent API)

Thank you.

[LucasAlegre]

Hi,

Can you share your code?

[TGW795]

Sure. This is my code and changes. (I've been checking the behavior of several observation values, and I've not written about iteration part.)

• sumo_rl/environment/observations.py

import numpy as np
from gymnasium import spaces

from .traffic_signal import TrafficSignal


class ObservationFunction:
    """Abstract base class for observation functions."""

    def __init__(self, ts: TrafficSignal):
        """Initialize observation function."""
        self.ts = ts

    @abstractmethod
    def __call__(self):
        """Subclasses must override this method."""
        phase_id = [1 if self.ts.green_phase == i else 0 for i in range(self.ts.num_green_phases)]
        observation = np.array(phase_id, dtype=np.float32)
        return observation

    @abstractmethod
    def observation_space(self):
        """Subclasses must override this method."""
        return spaces.Box(
            low=np.zeros(self.ts.num_green_phases, dtype=np.float32),
            high=np.ones(self.ts.num_green_phases, dtype=np.float32),


class DefaultObservationFunction(ObservationFunction):
    """Default observation function for traffic signals."""

    def __init__(self, ts: TrafficSignal):
        """Initialize default observation function."""
        super().__init__(ts)

    def __call__(self) -> np.ndarray:
        """Return the default observation."""
        phase_id = [1 if self.ts.green_phase == i else 0 for i in range(self.ts.num_green_phases)]  # one-hot encoding
        min_green = [0 if self.ts.time_since_last_phase_change < self.ts.min_green + self.ts.yellow_time else 1]
        density = self.ts.get_lanes_density()
        queue = self.ts.get_lanes_queue()
        observation = np.array(phase_id + min_green + density + queue, dtype=np.float32)
        return observation

    def observation_space(self) -> spaces.Box:
        """Return the observation space."""
        return spaces.Box(
            low=np.zeros(self.ts.num_green_phases + 1 + 2 * len(self.ts.lanes), dtype=np.float32),
            high=np.ones(self.ts.num_green_phases + 1 + 2 * len(self.ts.lanes), dtype=np.float32),
        )

• sumo_rl/environment/env.py(#L99)

        observation_class: ObservationFunction = ObservationFunction,

I thought that we would obtain only phase_id by these modifications, but in fact, I got values defined in DefaultObservationFunction. (I've checked this issue by running a code same as an example of PettingZoo Multi-Agent API.)

[LucasAlegre]

ObservationFunction is an abstract class, you should not modify it. You have to create a new class that implements the abstract methods:

class MyObservationFunction(ObservationFunction):

    def __init__(self, ts: TrafficSignal):
        self.ts = ts

    def __call__(self):
        phase_id = [1 if self.ts.green_phase == i else 0 for i in range(self.ts.num_green_phases)]
        observation = np.array(phase_id, dtype=np.float32)
        return observation

    def observation_space(self):
        return spaces.Box(
            low=np.zeros(self.ts.num_green_phases, dtype=np.float32),
            high=np.ones(self.ts.num_green_phases, dtype=np.float32),

# In your experiment file:
env = sumo_rl.env(..., observation_class=MyObservationFunction)

[TGW795]

It worked! Thank you!
This was just an elementary mistake on my part :)