API Reference

This page is generated automatically from the source docstrings. It documents the public classes most users will need to interact with. For anything not listed here, browse the source on GitHub.

The environment

pgtg.environment.PGTGEnv

Bases: Env

Class representing the modular PGTG environment with driver profiles.

Source code in pgtg/environment.py

class PGTGEnv(gym.Env):
    """Class representing the modular PGTG environment with driver profiles."""

    metadata: ClassVar[dict[str, Any]] = {
        "render_modes": ["human", "rgb_array", "pil_image"],
        "render_fps": 4,
    }

    @dataclass
    class RNGStates:
        map_rng: np.random.Generator
        car_rng: np.random.Generator

        @classmethod
        def from_seed(cls, seed: int | None) -> PGTGEnv.RNGStates:
            """Initialize all generators from a single master seed."""
            child_seeds = np.random.SeedSequence(seed).spawn(2)
            return cls(*(np.random.default_rng(s) for s in child_seeds))

        def to_dict(self) -> dict[str, Any]:
            """Returns the current state of all generators."""
            return {f.name: getattr(self, f.name).bit_generator.state for f in fields(self)}

        def from_dict(self, state_dict: dict[str, Any]) -> None:
            for name, state in state_dict.items():
                getattr(self, name).bit_generator.state = state

    map_path: str | None
    map_plan: MapPlan
    noise_path: list[tuple[str, tuple[int, int]]]
    termination_reason: TerminationReason | None
    cars: list[Car]
    _car_positions: np.ndarray

    def __init__(
        self,
        map_path: str | None = None,
        *,
        render_mode: str | None = None,
        only_render_obs: bool = False,
        traffic_density: float = 0.0,
        traffic_light_phases_duration: tuple[int, int, int] = (10, 3, 10),
        ignore_traffic_collisions: bool = False,
        car_spawn_exclusion_radius: int = 1,
        observation_context: (
            dict[str, Any] | Iterable[str | MF | int] | ObservationContext | None
        ) = None,
        reward_context: dict[str, Any] | RewardContext | None = None,
        random_map_context: dict[str, Any] | RandomMapContext | None = None,
        obstacles: Sequence[SteppableFeature] | None = None,
        driver_profile_weights: dict[DriverProfile | str, float] | None = None,
        driver_profile_config: DriverProfileConfig | dict | str | None = None,
        traffic_rules_config: TrafficRulesConfig | dict | str | None = None,
        rules_active: list[str] | None = None,
        tiles: str | None = None,
        obstacle_masks: str | None = None,
        traffic_lanes: str | None = None,
        info_level: str | None = "essential",
        seed: int | None = None,
        log_level: int | None = None,
    ):
        """Initialize the PGTG environment; usually called via `gymnasium.make()`.

        Args:
            map_path: Path to a map .json file, or the name of a bundled benchmark map
                (see `pgtg.map.defaults.DEFAULT_MAPS`). With `None`, a random map is
                generated each episode according to the
                [`random_map_context`][pgtg.contexts.RandomMapContext].
            render_mode: One of `None`, `'human'`, `'rgb_array'`, or `'pil_image'`.
            only_render_obs: Render the agent's observation instead of the entire map:
                the view is cropped to the observation window *and* only features in
                `observation_context.obs_features` are drawn. Applies to all render modes.
            traffic_density: Fraction of car-lane cells initially occupied by cars.
                `0.0` disables traffic.
            traffic_light_phases_duration: Steps spent in the (green, yellow, red) phases,
                respectively.
            ignore_traffic_collisions: If true, colliding with traffic does not terminate the
                episode.
            car_spawn_exclusion_radius: Chebyshev radius around the agent in which cars
                never spawn. `1` excludes the agent's cell plus its 8 neighbors; negative
                values disable the exclusion. `1` is the minimum necessary value to exclude
                unavoidable collisions with newly spawned cars (cars move first).
            observation_context: [`ObservationContext`][pgtg.contexts.ObservationContext],
                dict of its fields, or a bare iterable of observed features (shorthand for
                `{"obs_features": ...}`). Controls window size/sliding and which channels
                the agent observes.
            reward_context: [`RewardContext`][pgtg.contexts.RewardContext] or dict of its fields
                (rewards & penalties).
            random_map_context:  [`RandomMapContext`][pgtg.contexts.RandomMapContext] or dict of
                its fields; used only when `map_path` is `None`.
            obstacles: Explicit obstacle handlers. Defaults for all generatable obstacles
                are filled in automatically; handlers given here take precedence.
            driver_profile_weights: Mapping of driver profile to sampling weight.
                Overrides the weights from `driver_profile_config`.
            driver_profile_config:
                [`DriverProfileConfig`][pgtg.traffic.drivers.DriverProfileConfig], dict,
                or path to a .json/.yaml file defining driver behaviors and weights.
            traffic_rules_config: [`TrafficRulesConfig`][pgtg.traffic.rules.TrafficRulesConfig],
                dict, or path to a .json/.yaml file with additional traffic rules
                (merged with the built-in defaults).
            rules_active: Names of the loaded traffic rules to activate. `None` activates
                all loaded rules, an empty list none.
            tiles: Path to a custom tiles .json/.yaml file overriding the default tile
                templates. Tile size is inferred from the file; all three tile-data files
                must share the same dimensions, so non-default sizes require matching
                `obstacle_masks` and `traffic_lanes` files.
            obstacle_masks: Path to a custom obstacle-masks file.
            traffic_lanes: Path to a custom traffic-lanes file.
            info_level: How much information [`get_info()`][pgtg.environment.PGTGEnv.get_info]
                (and thus the info dict of [`reset()`][pgtg.environment.PGTGEnv.reset]/
                [`step()`][pgtg.environment.PGTGEnv.step]) contains. `'essential'` (default)
                returns cheap scalars only (position, velocity, flat tire, tile type, termination
                reason, subgoal progress). `'restorable'` additionally returns the
                full episode state needed by
                [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state]
                (cars, RNG states, visited set, active subgoals, driver/traffic-rule stats).
                `'none'`/`None` makes [`get_info()`][pgtg.environment.PGTGEnv.get_info]
                return an empty dict; [`step()`][pgtg.environment.PGTGEnv.step] still adds `cost`
                and `is_success`.
            seed: Master seed for all randomness.
                [`reset(seed=...)`][pgtg.environment.PGTGEnv.reset] restarts the episode sequence
                deterministically.
            log_level: Level of the `"pgtg"` logger. `None` (default) leaves it as it is
                (`logging.WARNING` until something changes it). The logger is process-wide,
                so passing a level here affects every PGTG environment.
        """
        self.__setup_logger(log_level)

        # Master entropy source. Every reset() spawns fresh child streams from this
        # sequence of *different* episodes, while reset(seed=...) restarts the sequence.
        # => Episodes only depend on their seed and number; not on previous episode behavior.
        self.seed = seed
        self._seed_seq = np.random.SeedSequence(seed)
        self.rng_states = PGTGEnv.RNGStates.from_seed(seed)

        # Register all obstacle handlers:
        # - because they are generatable, or
        # - because they were explicitly passed.
        explicit = list(obstacles) if obstacles is not None else []
        covered = {
            int(o.observation_feature) for o in explicit if o.observation_feature is not None
        }
        auto_filled = [
            factory()
            for factory in ObstacleRegistry.generatable()
            if int(factory.observation_feature) not in covered
        ]
        self.obstacles: list[SteppableFeature] = [*explicit, *auto_filled]
        self._default_obstacles = obstacles is None

        # Resolve the tile data once (paths -> preprocessed datasets);
        # None = the packaged default file. The loaders memoize per file,
        # so the datasets are shared and built once, whether default or custom.
        self._tiles = tile_registry.tiles_from_file(tiles) if tiles is not None else None
        self._obstacle_masks = (
            tile_registry.obstacle_masks_from_file(obstacle_masks)
            if obstacle_masks is not None
            else None
        )
        self._traffic_lanes = (
            tile_registry.traffic_lanes_from_file(traffic_lanes)
            if traffic_lanes is not None
            else None
        )
        # Infer tile size from tile data and validate three datasets share the same dims
        self.tile_width, self.tile_height = tile_registry.resolve_tile_dims(
            tiles=self._tiles,
            obstacle_masks=self._obstacle_masks,
            traffic_lanes=self._traffic_lanes,
        )

        # Only record requested info dict entries to avoid expensive computations
        # (e.g., when no restorations are needed)
        self.info_level = "none" if info_level is None else str(info_level).lower()
        if self.info_level not in ("none", "essential", "restorable"):
            raise ValueError(
                f"info_level must be one of 'none', 'essential', 'restorable' or None, "
                f"got {info_level!r}"
            )

        # Resolve map path.
        # - a default map name resolved to its bundled .json
        # - a custom map path to a custom .json
        # - None: generate a random map each episode
        if map_path is not None and map_path in DEFAULT_MAPS:
            self.map_path = str(DEFAULT_MAPS[map_path])
        else:
            self.map_path = map_path
        if self.map_path is not None:
            self.logger.debug("Loading map from path %s", self.map_path)
            self.map_plan = json_file_to_map_plan(self.map_path)
            self.map = EpisodeMap(
                self.map_plan,
                tiles=self._tiles,
                obstacle_masks=self._obstacle_masks,
                traffic_lanes=self._traffic_lanes,
            )

        self.rm_context = RandomMapContext.model_validate(
            (random_map_context or {}),
            context={"obstacles": self.obstacles},
        )

        # Each branch decides which obstacle features can occur in this env.
        # It yields both the auto-observe set and which default handlers are worth keeping around
        # `keep_features = None` means "keep every handler".
        if self.map_path is not None:
            # Only add obstacles to the auto-observe if they are generated on this *fixed* map
            present = self.map.feature_union()
            self._generatable_here = {
                int(o.observation_feature)
                for o in self.obstacles
                if o.observation_feature is not None and int(o.observation_feature) & present
            }
            keep_features = self._generatable_here
        elif self.rm_context.obstacle_probability > 0:
            # Only add obstacles generally generatable on the *random* maps to the obs space
            self._generatable_here = {
                int(feature)
                for feature, weight in self.rm_context.obs_feature_weights.items()
                if weight > 0
            }
            # Zero-weight handlers stay registered even though they are not auto-observed
            keep_features = None
        else:
            # No obstacles generatable, empty set for auto-observe
            self._generatable_here = set()
            keep_features = set()

        # Construct reward and observation contexts
        self.reward_context = RewardContext.model_validate(reward_context or {})
        self.obs_context = ObservationContext.model_validate(
            observation_context or {},
            context={
                "obstacles": self.obstacles,
                "tile_dims": (self.tile_width, self.tile_height),
                "generatable_here": self._generatable_here,
            },
        )

        # Drop default handlers that can never trigger here, so they cost nothing per step.
        # Applied only after the observation context is built: an obstacle with observe=True
        # keeps its channel even when this env can never produce it.
        if self._default_obstacles and keep_features is not None:
            self.obstacles = [
                obstacle
                for obstacle in self.obstacles
                if obstacle.observation_feature is not None
                and int(obstacle.observation_feature) in keep_features
            ]

        # Generate observation_space and action_space for gymnasium API
        if self.map_path is not None:
            # Fixed size
            self.observation_space = self.obs_context.gen_obs_space(
                self.map.width,
                self.map.height,
            )
        else:
            # Size inferred by tile_dim * map_dim
            self.observation_space = self.obs_context.gen_obs_space(
                self.rm_context.width * self.tile_width,
                self.rm_context.height * self.tile_height,
            )
        self.action_space = spaces.Discrete(9)
        self.reward_range = (-np.inf, np.inf)

        # Internal state: Traffic rules to apply *on each step*
        self.rule_engine = TrafficRuleEngine()
        self._load_traffic_rules(traffic_rules_config, rules_active)

        # Internal state: traffic lights. Prepend so their check runs before any random obstacle
        self._traffic_light = TrafficLight(phases_duration=traffic_light_phases_duration)
        self.obstacles.insert(0, self._traffic_light)

        # Internal state: Every obstacle keeps track of its own RNG state. Deterministic
        # handlers (the traffic light) own no stream and stay out of the seeding sequence.
        # NOTE: rebuild if `self.obstacles` is mutated later.
        self._rng_handlers: list[HasRng] = [h for h in self.obstacles if isinstance(h, HasRng)]
        self._index_obstacle_handlers()  # faster access

        self.traffic_density = traffic_density
        self.ignore_traffic_collisions = ignore_traffic_collisions
        self.car_spawn_exclusion_radius = int(car_spawn_exclusion_radius)

        # Warnings: Emit warnings for any obstacle that is generatable but not observable:
        #           The agent is probably unable learn to avoid such obstacles
        if self.traffic_density > 0 and not self.obs_context.is_observed(MF.TRAFFIC):
            self.logger.warning(
                "traffic_density=%s spawns cars, but MF.TRAFFIC is not in the observation "
                "features %s — cars will be invisible to the agent yet can still terminate "
                "the episode on collision.",
                self.traffic_density,
                self.obs_context.obs_features,
            )

        for feature in self._generatable_here:
            if not self.obs_context.is_observed(feature):
                self.logger.warning(
                    "'%s' can appear in this environment but is not in the observation "
                    "features %s — an agent will not be able to learn to avoid it.",
                    FeatureRegistry.describe(feature),
                    self.obs_context.obs_features,
                )

        # Internal state: Driver profiles for different driver behaviors; default and custom
        if driver_profile_config is None:
            self._driver_profile_config = DriverProfileConfig.defaults()
        else:
            self._driver_profile_config = DriverProfileConfig.model_validate(driver_profile_config)

        if driver_profile_weights is not None:
            normalized_weights = {
                (p.value if isinstance(p, DriverProfile) else str(p)): w
                for p, w in driver_profile_weights.items()
            }
            self._driver_profile_config = self._driver_profile_config.model_copy(
                update={"weights": normalized_weights}
            )

        self.driver_profile_dist = self._driver_profile_config.normalized_weights
        self._refresh_profile_sampler()

        # Some irrelevant rendering state
        self.render_mode = render_mode
        self.only_render_obs = only_render_obs
        self.window_size = 720
        self.window = None
        self.clock = None

        # Start the first episode, so a freshly constructed env is immediately steppable
        self._start_episode(np.random.SeedSequence(self._seed_seq.entropy))

    def __setup_logger(self, log_level: int | None) -> None:
        self.logger = PGTGLogger.get_logger()
        if log_level is not None:
            # Shared by every env in the process; only touch it when asked to.
            self.logger.setLevel(log_level)
        self.logger.debug(
            "Initialized PGTGEnv with logger (level %d)", self.logger.getEffectiveLevel()
        )

    def _load_traffic_rules(
        self,
        traffic_rules_config: TrafficRulesConfig | dict | str | None,
        rules_active: list[str] | None,
    ) -> None:
        """Load traffic rules into the rule engine and select which are active.

        The *loaded* rules are the built-in defaults plus any custom rules from
        `traffic_rules_config` (a custom rule reusing a default's name is a conflict and
        raises). `rules_active` then selects which of the loaded rules are active by name:
        `None` activates every loaded rule; a list activates only the named ones (an empty
        list activates none). An unknown name raises ValueError. Str-path inputs are
        resolved by TrafficRulesConfig's model validator.

        Because of name-based selection, names have to be unique.
        """
        loaded: dict[str, TrafficRule] = {}
        for rule in TrafficRulesConfig.defaults().rules:
            loaded[rule.name] = rule
        if traffic_rules_config is not None:
            for rule in TrafficRulesConfig.model_validate(traffic_rules_config).rules:
                if rule.name in loaded:
                    raise ValueError(f"Rule with name {rule.name} already exists.")
                loaded[rule.name] = rule

        if rules_active is None:
            active = list(loaded.values())
        else:
            missing = [name for name in rules_active if name not in loaded]
            if missing:
                raise ValueError(
                    f"Unknown traffic rule name(s) in rules_active: {sorted(set(missing))}. "
                    f"Loaded rules: {sorted(loaded)}."
                )
            active = [loaded[name] for name in rules_active]

        if active:
            self.rule_engine.add_rules_from_config(TrafficRulesConfig(rules=active))

    def add_traffic_rule(self, rule_dict: dict[str, Any]) -> None:
        """Adds a new traffic rule."""
        self.rule_engine.add_rule_from_dict(rule_dict)

    def remove_traffic_rule(self, rule_name: str) -> bool:
        """Removes a traffic rule by name. Returns whether the rule existed."""
        return self.rule_engine.remove_rule(rule_name)

    def get_agent_direction_string(self) -> str:
        """Returns the agent's current direction as used by the traffic rules."""
        return self.rule_engine.get_agent_direction(self)

    def _reseed_obstacle_rngs(self, seed_seq: np.random.SeedSequence) -> None:
        """Load each obstacle generator's state from the next child of ``seed_seq``.

        Pass a sequence already advanced past the two map/car children so obstacle streams
        line up identically whether they are seeded at construction or on ``reset()``. States
        are loaded onto the existing Generator objects in place so handler references stay live.
        """
        for handler in self._rng_handlers:
            (child,) = seed_seq.spawn(1)
            handler.rng.bit_generator.state = np.random.default_rng(child).bit_generator.state

    def reset(
        self,
        *,
        seed: int | None = None,
        **kwargs: Any,
    ) -> tuple[dict, dict[str, Any]]:
        """Starts a fresh episode, possibly reinitializing the seed sequence for future episodes.

        Args:
            seed: Optional seed to reseed all associated rng streams.

        Returns:
            observation: The resulting observation after the reset.
            info: Additional information about the state of the environment. Analogous to the info
                returned by step().
        """
        with TimedExecution("reset", PGTGLogger.te_cb, "pgtg"):
            self.logger.debug("Resetting environment.")
            super().reset(seed=seed, **kwargs)

            if seed is not None:
                self.seed = seed
                self._seed_seq = np.random.SeedSequence(seed)
            self._start_episode(self._seed_seq)

            if self.render_mode == "human":
                self.render()
            return (self.get_observation(), self.get_info())

    def _start_episode(self, seed_seq: np.random.SeedSequence) -> None:
        """Builds a complete, steppable episode from the next children of ``seed_seq``."""
        # The seeding order is crucial: map and car RNGs first, then one child per obstacle.
        fresh = PGTGEnv.RNGStates(*(np.random.default_rng(s) for s in seed_seq.spawn(2)))
        self.rng_states.from_dict(fresh.to_dict())
        self._reseed_obstacle_rngs(seed_seq)

        # Generate a new random map, or re-instantiate the fixed one
        if self.map_path is None:
            self.logger.debug("Generating random map.")
            self.map_plan = generate_map(self.rm_context, self.rng_states.map_rng)
        self.map = EpisodeMap(
            self.map_plan,
            tiles=self._tiles,
            obstacle_masks=self._obstacle_masks,
            traffic_lanes=self._traffic_lanes,
        )

        # Agent state
        self.position = np.array(self.rng_states.map_rng.choice(self.map.starters))
        self.velocity = np.zeros(2, dtype=int)
        self.termination_reason = None
        self.flat_tire = False
        self.subgoals_reached = 0

        # Episode history, seeded with the starting cell
        start = tuple(self.position)
        self.positions_path = [start]
        self.tile_path = [start]
        self.visited_set = {start}
        self.noise_path = []

        # Traffic
        self._traffic_light.phase_counter = 0
        self.cars = []
        self._next_car_id = 0
        self._refresh_profile_sampler()
        if self.traffic_density > 0:
            self._create_initial_traffic()
        self._update_car_positions_array()
        self._occupied: Counter[tuple[int, int]] = Counter(car.position for car in self.cars)

        self.logger.debug("Initial position: (%d, %d)", *self.position)

    def _update_car_positions_array(self) -> None:
        """Rebuilds the (n_cars, 2) position array used by spatial masking.

        np.fromiter over the flat coordinates is several times faster than
        np.array on a list of tuples, which matters at one rebuild per step."""
        n = len(self.cars)
        if n:
            flat = np.fromiter(
                (coord for car in self.cars for coord in car.position),
                dtype=np.int64,
                count=2 * n,
            )
            self._car_positions = flat.reshape(n, 2)
        else:
            self._car_positions = np.empty((0, 2), dtype=np.int64)

    def _refresh_profile_sampler(self) -> None:
        """Precomputes the cumulative-weight table `_select_driver_profile` samples
        from. Called at construction and on every reset, so changes to
        `driver_profile_dist` between episodes still take effect."""
        self._profile_names = list(self.driver_profile_dist.keys())
        cumulative = np.cumsum(
            np.fromiter(self.driver_profile_dist.values(), dtype=np.float64),
        )
        # weights are normalized; pin the last bin against float drift
        cumulative[-1] = 1.0
        self._profile_cumulative = cumulative

    def _select_driver_profile(self) -> str:
        """Samples a driver profile according to the configured weights.

        Inverse-CDF sampling on the precomputed cumulative weights: one uniform
        draw + searchsorted, instead of `Generator.choice(p=...)`, which rebuilt
        and validated the probability vector on every call (~25 µs → ~1 µs; this
        runs once per spawned car, ~1000× per dense reset)."""
        draw = self.rng_states.car_rng.random()
        return self._profile_names[
            int(np.searchsorted(self._profile_cumulative, draw, side="right"))
        ]

    def _should_car_stop_at_traffic_light(self, car: Car, light_phase: str) -> bool:
        """Decides whether a car stops at a traffic light, based on its driver profile."""
        behavior = self._driver_profile_config.behaviors[car.driver_profile]

        if light_phase == SC.GREEN:
            return False
        if light_phase == SC.YELLOW:
            return self.rng_states.car_rng.random() < behavior.yellow_light_stop_probability
        if light_phase == SC.RED:
            return self.rng_states.car_rng.random() < behavior.red_light_stop_probability

        return True

    def _should_car_move(self, car: Car, delay_draw: float, speed_draw: float) -> bool:
        """Decides whether a car moves this step, based on its driver profile.

        The two uniform draws are taken from a per-step batch drawn in `step()`
        """
        behavior = self._driver_profile_config.behaviors[car.driver_profile]

        # still in delay mode
        if car.last_action_delay > 0:
            car.last_action_delay -= 1
            return False

        # sleepy, delay a few time steps
        if delay_draw < behavior.reaction_delay_probability:
            car.last_action_delay = int(self.rng_states.car_rng.integers(1, 4))
            return False

        # stochastic movement based on `speed_multiplier` (0.0 = never moves, 1.0 = always moves)
        return speed_draw < behavior.speed_multiplier

    def _decompose_velocity(self, velocity: np.ndarray | None = None) -> list[np.ndarray]:
        """Decompose a velocity into unit substeps along the ideal straight line.

        The velocity vector is rasterised into a connected sequence of moves that
        each advance at most one cell per axis, so the substep walk in ``step`` can
        check every cell the straight-line trajectory passes through.
        The number of substeps is the Chebyshev (L-infinity) norm, i.e., the dominant
        axis advances exactly one cell per substep; diagonal movement allowed.

        Each sampled point is rounded *half away from zero* in exact integer
        arithmetic, which guarantees two properties:

        * **Reflection / rotation symmetry.** ``decompose(-v) == -decompose(v)``
          element-wise, so mirroring or rotating the map mirrors/rotates every
          trajectory. The environment has no built-in axis bias.
        * **Reproducibility w/o FPEs.** The rounding runs on ``int64`` via ``//``.
          So a point landing on an exact half-cell boundary is resolved by definition
          half away from zero, instead of by rounding mode.
          Ties occur exactly when ``max(|vx|, |vy|) / gcd(|vx|, |vy|)`` is even,
          e.g. ``(2, 1)`` or ``(10, -5)``.
          While 0.5 is exactly representable in float64, integer arithmetic still
          avoids any risk of introducing FPEs.
          The decomposition also reads only the relative displacement, never
          the agent's absolute position; making it independent of the position.

        Trade-off: at exact-half ties this rule is not *reversal*-symmetric:
        the diagonal substep is taken adjacent to the start, so the same segment
        traversed in opposite directions visits different middle cells
            (forward ``(2,   1)`` passes through ``(1, 1)``,
             reverse ``(-2, -1)`` passes through ``(1, 0)``).
        Reversal symmetry would require a fixed global tie-break axis, which would
        break the isotropy above; isotropy is the more useful invariant.
        """
        if velocity is None:
            velocity = self.velocity

        num_steps = int(np.max(np.abs(velocity)))
        if num_steps == 0:
            return []

        offsets = np.outer(np.arange(num_steps + 1), np.asarray(velocity, dtype=np.int64))
        path = np.sign(offsets) * ((2 * np.abs(offsets) + num_steps) // (2 * num_steps))

        steps = np.diff(path, axis=0)
        return list(steps)

    def generate_frame(
        self,
        hide_positions: bool = False,
        show_observation_window: bool = True,
        only_observation: bool = False,
    ) -> Image:
        """Renders the current state as a PIL image.

        Args:
            hide_positions: Hide the path taken in the current episode.
            show_observation_window: Dim everything outside the observation window.
            only_observation: Render the agent's observation: crop to the observation
                window and draw only the observed features.
        """
        if only_observation:
            pic = graphic.create_observation_window(
                self, show_path=(not hide_positions), only_observed_features=True
            )
        else:
            pic = graphic.create_map(
                self,
                show_path=(not hide_positions),
                show_observation_window=show_observation_window,
            )

        if pic is None:
            from PIL import Image as PILImage

            pic = PILImage.new("RGBA", (400, 400), (255, 255, 255, 255))

        if pic.mode != "RGBA":
            pic = pic.convert("RGBA")

        return pic

    def render(self) -> Image | npt.NDArray | None:  # type: ignore[override]
        """Returns a rendered representation of the game according to the render mode.

        In `'human'` mode this draws the current frame into the pygame window and
        returns `None`. [`reset()`][pgtg.environment.PGTGEnv.reset] and
        [`step()`][pgtg.environment.PGTGEnv.step] automatically call `render()`.

        Returns:
            The rendered representation, or `None` if the render mode is `"human"` or `None`.
        """

        match self.render_mode:
            case None:
                return None
            case "human":
                self._render_frame_for_human()
                return None
            case "rgb_array":
                return np.asarray(
                    self.generate_frame(only_observation=self.only_render_obs).convert("RGB")
                )
            case "pil_image":
                return self.generate_frame(only_observation=self.only_render_obs)
            case _:
                raise ValueError(
                    f"render_mode={self.render_mode!r} is not supported; expected one of "
                    f"None, {', '.join(map(repr, self.metadata['render_modes']))}."
                )

    def close(self) -> None:
        """Closes the pygame window opened by the `'human'` render mode, if any."""
        if self.window is not None:
            pygame.display.quit()
            pygame.quit()
            self.window = None
            self.clock = None

    def _render_frame_for_human(self) -> None:
        """Renders the current state of the environment in a window."""

        pil_image = self.generate_frame(only_observation=self.only_render_obs)

        if self.window is None:
            pygame.init()
            pygame.display.init()
            self.window = pygame.display.set_mode(
                (
                    self.window_size * (pil_image.size[0] / pil_image.size[1]),
                    self.window_size,
                ),
            )
            pygame.display.set_caption("PGTG")

        if self.clock is None:
            self.clock = pygame.time.Clock()

        pygame_image = pygame.image.fromstring(
            pil_image.tobytes(),
            pil_image.size,
            pil_image.mode,
        ).convert()

        pygame_image = pygame.transform.scale(
            pygame_image,
            (
                self.window_size * (pil_image.size[0] / pil_image.size[1]),
                self.window_size,
            ),
        )

        self.window.blit(pygame_image, pygame_image.get_rect())
        pygame.event.pump()
        pygame.display.update()

        # delay to keep the framerate stable
        self.clock.tick(self.metadata["render_fps"])

    def _within_spawn_exclusion(self, pos: tuple[int, int] | np.ndarray) -> bool:
        """True if ``pos`` is within ``car_spawn_exclusion_radius`` (Chebyshev) of the agent.

        Keeps cars from spawning on or right next to the agent, which would be an
        unavoidable crash. A negative radius disables the exclusion.
        """
        r = self.car_spawn_exclusion_radius
        if r < 0:
            return False
        ax, ay = int(self.position[0]), int(self.position[1])
        return abs(int(pos[0]) - ax) <= r and abs(int(pos[1]) - ay) <= r

    def _create_initial_traffic(self) -> None:
        """Creates a number of cars defined by the traffic density."""

        candidates = [
            p for p in self.map.traffic_spawnable_positions if not self._within_spawn_exclusion(p)
        ]

        num_cars = int(len(self.map.traffic_spawnable_positions) * self.traffic_density)

        if num_cars > 0 and candidates:
            chosen_indices = self.rng_states.car_rng.choice(
                len(candidates),
                size=min(num_cars, len(candidates)),
                replace=False,
            )
            initial_car_positions = [candidates[idx] for idx in chosen_indices]
        else:
            initial_car_positions = []

        self.cars = [
            self.__spawn_car_at(initial_car_position)
            for initial_car_position in initial_car_positions
        ]

    def _get_next_car_position_and_route(
        self,
        car: Car,
        occupied: Counter[tuple[int, int]],
        delay_draw: float,
        speed_draw: float,
    ) -> tuple[Position, str] | None:
        """The car's next position and route, or None if it has nowhere left to go."""
        if not self._should_car_move(car, delay_draw, speed_draw):
            car.patience_counter += 1
            return (car.position, car.route)

        if car.route is None:
            car.patience_counter += 1
            return None

        behavior = self._driver_profile_config.behaviors[car.driver_profile]

        # Candidate moves come from the map's precomputed transition table
        # (N/E/S/W probing order, 'all'-lane routes pre-sorted); static traffic lanes per episode
        for (
            possible_position,
            ori,
            square_lanes,
            all_routes,
            has_traffic_light,
        ) in self.map.car_transitions.get(car.position, ()):
            if all_routes is not None:
                # Entered via an 'all' lane: pick any of the cell's other routes.
                chosen_idx = int(self.rng_states.car_rng.integers(len(all_routes)))

                car.patience_counter = 0
                return possible_position, all_routes[chosen_idx].direction

            car_following_lane = LR.get_by_str(car.route, ori)
            if car_following_lane not in square_lanes:
                continue

            # Traffic follow traffic light rules; mostly that is.
            if has_traffic_light:
                traffic_light_phase = self.get_traffic_light_phase()
                if traffic_light_phase in [
                    SC.YELLOW,
                    SC.RED,
                ] and self._should_car_stop_at_traffic_light(
                    car,
                    traffic_light_phase,
                ):
                    car.patience_counter += 1
                    return (car.position, car.route)

            # Overtaking; two cars can occupy the same cell!
            if (
                not occupied[possible_position] > 0
                or behavior.min_following_distance == 0  # TODO: Not distinguishing everything > 0
                or (
                    car.patience_counter > (behavior.patience_level * 10)
                    and self.rng_states.car_rng.random() < (1.0 - behavior.patience_level)
                )
            ):
                car.patience_counter = 0
                return (possible_position, car.route)

            car.patience_counter += 1
            return (car.position, car.route)

        car.patience_counter += 1
        return None

    def __spawn_car_at(self, position: Position) -> Car:
        # Ordered non-'all' routes for this cell, cached on the map
        routes = self.map.spawn_routes(*position)
        chosen_idx = int(self.rng_states.car_rng.integers(len(routes)))

        driver_profile = self._select_driver_profile()

        new_car = Car(
            id=self._next_car_id,
            position=position,
            route=routes[chosen_idx],
            driver_profile=driver_profile,
        )
        self._next_car_id += 1

        return new_car

    def _spawn_new_car(self) -> Car | None:
        """Creates a new car and returns it. It still has to be stored.

        Returns:
            Newly created car, or None when the map has no usable car spawner.
        """

        usable = self.map.usable_car_spawners
        if not usable:
            self.logger.warning(
                "A car left the map but there is no usable car spawner to respawn it on, so "
                "it is not replaced. Check that the traffic-lane data marks spawner cells "
                "and gives them at least one non-'all' route.",
            )
            return None

        # Prefer spawners outside the agent's exclusion zone; fall back to the full set.
        spawners = [s for s in usable if not self._within_spawn_exclusion(s)] or usable
        spawner_idx = int(self.rng_states.car_rng.integers(len(spawners)))
        return self.__spawn_car_at(Position._make(spawners[spawner_idx]))

    def get_traffic_light_phase(self) -> str:
        """Returns the current traffic light phase (one of `"green"`, `"yellow"`, `"red"`)."""
        return self._traffic_light.phase

    def get_driver_profile_stats(self) -> dict[str, Any]:
        """Returns statistics about the current driver profile distribution."""
        profile_counts = dict.fromkeys(self._driver_profile_config.behaviors, 0)

        for car in self.cars:
            if car.driver_profile in profile_counts:
                profile_counts[car.driver_profile] += 1

        total_cars = len(self.cars)
        if total_cars > 0:
            profile_percentages = {k: (v / total_cars) * 100 for k, v in profile_counts.items()}
        else:
            profile_percentages = dict.fromkeys(profile_counts.keys(), 0)

        return {
            "counts": profile_counts,
            "percentages": profile_percentages,
            "total_cars": total_cars,
            "configured_percentages": {k: v * 100 for k, v in self.driver_profile_dist.items()},
        }

    def get_subgoal_compass_direction(
        self,
        position: Annotated[NDArray[np.int_], Literal[2]] | None = None,
    ) -> tuple[int, tuple[int, int], bool]:
        """Calculate direction indicators to the nearest subgoal.

        Args:
            position: Position to calculate from. (default: the agent's current position)

        Returns:
            compass: index of the nearest subgoal direction, clockwise from north:
                0=N, 1=NE, 2=E, 3=SE, 4=S, 5=SW, 6=W, 7=NW. When there is no active subgoal
                the sentinel 8 is returned.
            distances: Tuple with the signed x and y offset to the nearest subgoal.
            in_window: True iff nearest subgoal is within the observation window.
        """
        if position is None:
            position = self.position

        active_subgoals = self.map.active_subgoals
        if active_subgoals.ndim < 2 or len(active_subgoals) == 0:
            return 8, (0, 0), False

        # Signed offsets toward the nearest subgoal (positive dx = east, positive dy = south).
        deltas = active_subgoals - position
        nearest = deltas[np.argmin(np.sum(np.abs(deltas), axis=1))]
        dx, dy = int(nearest[0]), int(nearest[1])

        dir_index = int(np.round(np.arctan2(dx, -dy) / (np.pi / 4))) % 8

        sub_x, sub_y = position[0] + dx, position[1] + dy
        tl_x, tl_y, br_x, br_y = self.get_obs_window_boundaries(position)
        in_window = bool(tl_x <= sub_x <= br_x and tl_y <= sub_y <= br_y)

        return dir_index, (dx, dy), in_window

    def _index_obstacle_handlers(self) -> None:
        """Build the feature-bit -> handler index used by `_dispatch_map_features`.

        The obstacle set is fixed after construction.
        Call this again if `self.obstacles` is mutated afterwards after env creation.
        """
        self._always_handlers: list[SteppableFeature] = []
        self._handlers_by_feature: dict[int, list[SteppableFeature]] = {}
        for handler in self.obstacles:
            feature = int(handler.observation_feature)
            single_bit = feature and (feature & (feature - 1)) == 0
            if handler.observation_gated and single_bit:
                self._handlers_by_feature.setdefault(feature, []).append(handler)
            else:
                self._always_handlers.append(handler)

    def _dispatch_map_features(self, local_ctx: StepContext) -> bool:
        """Dispatch obstacle handlers for the current sub-step.

        Non-gated handlers (e.g. the traffic light) always run.
        Single-bit gated obstacles are dispatched straight from
        `self._handlers_by_feature` for the feature bits present at the current cell.

        Returns `True` iff the sub-step loop should break.
        """
        for handler in self._always_handlers:
            if handler.act(self, local_ctx):
                return True

        cur_x, cur_y = int(local_ctx.current_position[0]), int(local_ctx.current_position[1])
        mask = self.map.get_features_at(cur_x, cur_y, default=0)

        while mask:
            bit = mask & -mask
            mask ^= bit
            for handler in self._handlers_by_feature.get(bit, ()):
                if handler.act(self, local_ctx):
                    return True
        return False

    def step(
        self,
        action: int,
    ) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]:
        """Performs the given action and returns the results.

        Args:
            action: The action to perform.

        Returns:
            observation: The resulting observation after the transition.
            reward: The reward for the action.
            terminated: Whether or not the episode has ended.
            truncated: Always `False`. No native truncation.
            info: Additional information about the state of the environment.
                Analogous to the info returned by reset().
        """
        with TimedExecution("step", PGTGLogger.te_cb, "pgtg"):
            if self.terminated:
                raise RuntimeError("Already done, step has no further effect")

            ### STEP SETUP
            self._traffic_light.tick()
            self.rule_engine.triggered_rules.clear()

            ### CARS MOVE FIRST ###
            # `_occupied` persists across steps: the loop below keeps it in sync
            new_cars = []
            occupied = self._occupied
            # One batched draw for all cars' (delay, speed) decisions
            draws = self.rng_states.car_rng.random(2 * len(self.cars)).tolist() if self.cars else []
            for i, car in enumerate(self.cars):
                next_position_and_route = self._get_next_car_position_and_route(
                    car,
                    occupied,
                    draws[2 * i],
                    draws[2 * i + 1],
                )

                occupied[car.position] -= 1
                if next_position_and_route is None:
                    replacement = self._spawn_new_car()
                    if replacement is None:
                        continue  # no usable spawner: the car leaves without a successor
                    new_cars.append(replacement)
                else:
                    car.position, car.route = next_position_and_route
                    new_cars.append(car)
                occupied[new_cars[-1].position] += 1

            self.cars = new_cars
            self._update_car_positions_array()

            ### AGENT STEP ###
            # Fill local context, so obstacles and traffic rules can modify dynamics
            local_ctx = StepContext(
                current_position=self.position.copy(),
                acceleration=np.array(ACTIONS_TO_ACCELERATION[action]),
            )
            self.velocity = self.velocity + local_ctx.acceleration
            local_ctx.remaining_substeps = [*self._decompose_velocity(), None]

            # Always rebound by the loop below (it runs at least once, for the None sentinel)
            current_pos_t: tuple[int, int] = tuple(local_ctx.current_position)
            while local_ctx.remaining_substeps:
                # queue with pop so handlers can interrupt movements
                velocity_substep = local_ctx.remaining_substeps.pop(0)
                local_ctx.velocity_substep = velocity_substep
                current_pos_t = tuple(local_ctx.current_position)

                # Verify termination conditions
                if not self.map.inside_map(*current_pos_t):
                    self.termination_reason = TerminationReason.OUT_OF_BOUNDS

                elif self.map.feature_at(*current_pos_t, MF.WALL):
                    self.termination_reason = TerminationReason.WALL_COLLISION

                elif not self.ignore_traffic_collisions and occupied[current_pos_t] > 0:
                    self.termination_reason = TerminationReason.TRAFFIC_COLLISION

                if self.termination_reason is not None:  # same penalty for all crashes
                    local_ctx.sum_penalty += self.reward_context.crash_penalty
                    break

                if self.map.feature_at(*current_pos_t, MF.FINAL_GOAL):
                    local_ctx.sum_reward += (
                        self.reward_context.individual_subgoal_reward(self.map.num_subgoals)
                        + self.reward_context.final_goal_bonus
                    )
                    self.subgoals_reached += 1
                    local_ctx.is_success = True
                    self.termination_reason = TerminationReason.GOAL
                    break

                # Subgoal reward handling; on final goal this was applied already as well
                if self.map.feature_at(*current_pos_t, MF.SUBGOAL):
                    local_ctx.sum_reward += self.reward_context.individual_subgoal_reward(
                        self.map.num_subgoals,
                    )
                    self.subgoals_reached += 1
                    self.map.set_subgoals_to_used(*current_pos_t)

                # This was just the verification for the last movement step (None sentinel)
                if local_ctx.velocity_substep is None:
                    continue

                # Walk all obstacles and apply their effects. Returning true, means the step loop
                # is interrupted. Note, that if the obstacle moves the agent, it should not return
                # true but instead set remaining_substeps to `[None]` to verify the last move
                if self._dispatch_map_features(local_ctx):
                    break

                # Traffic rules are evaluated at the cell being left, after this cell's
                # rewards/obstacles but before moving. Same return contract as above for obstacles.
                if self.rule_engine.apply_rules(self, local_ctx):
                    break

                # Finally move the agent one substep along the decomposed velocity
                local_ctx.current_position += local_ctx.velocity_substep
                current_pos_t = tuple(local_ctx.current_position)
                self.tile_path.append(current_pos_t)

            if self.flat_tire:
                self.velocity = np.zeros(2, dtype=int)

            old_pos_tuple = tuple(self.position)
            moved = old_pos_tuple != current_pos_t
            # Standing still means *choosing* to stay put: no acceleration and no movement.
            if not moved:
                if not local_ctx.acceleration.any():  # Only discourage neutral action
                    local_ctx.sum_penalty += self.reward_context.standing_still_penalty
            elif current_pos_t in self.visited_set:
                local_ctx.sum_penalty += self.reward_context.already_visited_penalty

            self.position = local_ctx.current_position
            self.visited_set.add(current_pos_t)
            self.positions_path.append(tuple(self.position))

            if self.render_mode == "human":
                self.render()

            info = self.get_info()
            info["cost"] = local_ctx.sum_penalty
            info["is_success"] = local_ctx.is_success
            reward = (
                local_ctx.sum_reward
                if self.reward_context.separate_reward_cost
                else local_ctx.sum_reward - local_ctx.sum_penalty
            )

            return (self.get_observation(), reward, self.terminated, False, info)

    def light_step(
        self,
        action: int,
    ) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]:
        """Copies the environment and steps the copy. The original stays unchanged.

        **Note:** this deep-copies the entire environment (including the map grid and all cars),
        which is costly. Stepping and restoring with
        [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state] might be faster.

        Args:
            action: The action to perform.

        Returns:
            observation: The resulting observation after the transition.
            reward: The reward for the action.
            terminated: Whether or not the episode has ended.
            truncated: Always `False`. No native truncation.
            info: Additional information about the state of the environment.
                Analogous to the info returned by reset().
        """

        env_copy = copy.deepcopy(self)
        return env_copy.step(action)

    def set_to_state(self, state: dict[str, Any]) -> tuple[dict, dict[str, Any]]:
        """
        Sets the environment to a given state.

        Restores the full mutable episode state captured by
        [`get_info()`][pgtg.environment.PGTGEnv.get_info]:
            position, velocity, flat tire, RNG states, traffic-light phase, cars,
            subgoal consumption, the visited set, the subgoal counter, and the termination reason,
        so that setting two environments to the same state (built from the same map) and
        choosing the same actions yields the same future.

        Leaving out the rng states is supported to obtain an uncertain future.

        Append-only history (`positions_path`/`tile_path`/`noise_path`) is reset
        to a clean baseline at this state. The map itself is not re-created, so `state`
        must come from an environment built on the same map.


        Args:
            state: The state to set the environment to (a dict as returned by
                get_info(level="restorable")).

        Returns:
            observation: The observation after setting the state.
            info: Additional information about the state of the environment.
        """

        # TODO: Possibly fix map rebuild for procedural resets.
        #       Our StateRestorationWrapper should have the code for this already. Just port it.

        missing = [key for key in ("cars", "rng_states") if key not in state]
        if missing:
            raise ValueError(
                f"set_to_state requires a 'restorable' state dict (missing keys: {missing}). "
                "Capture the state with get_info(level='restorable') or construct the "
                "environment with info_level='restorable'."
            )

        self.position = np.array([state["x"], state["y"]])
        self.velocity = np.array([state["x_velocity"], state["y_velocity"]])
        self.flat_tire = state["flat_tire"]

        if "rng_states" in state:
            self.rng_states.from_dict(state["rng_states"])

        obstacle_rng_states = state.get("obstacle_rng_states")
        if obstacle_rng_states:
            for handler in self._rng_handlers:
                saved = obstacle_rng_states.get(int(handler.observation_feature))
                if saved is not None:
                    handler.rng.bit_generator.state = saved

        self._traffic_light.phase_counter = state.get("traffic_light_phase_counter", 0)

        self.cars = []
        if state["cars"] is not None and len(state["cars"]) > 0:
            self.cars = [Car.model_validate(car_data) for car_data in state["cars"]]
            self._next_car_id = self.cars[-1].id + 1

        self._update_car_positions_array()
        self._occupied = Counter(car.position for car in self.cars)

        reason_value = state.get("termination_reason")
        self.termination_reason = (
            TerminationReason(reason_value) if reason_value is not None else None
        )
        self.subgoals_reached = state.get("subgoals_reached", 0)
        self.visited_set = {
            tuple(cell) for cell in state.get("visited_set", [tuple(self.position)])
        }

        if "active_subgoals" in state:
            target_active = {tuple(cell) for cell in state["active_subgoals"]}
            for x in range(self.map.width):
                for y in range(self.map.height):
                    if not (
                        self.map.feature_at(x, y, MF.SUBGOAL)
                        or self.map.feature_at(x, y, MF.USED_SUBGOAL)
                    ):
                        continue
                    if (x, y) in target_active:
                        self.map.add_feature_at(x, y, MF.SUBGOAL)
                        self.map.remove_feature_at(x, y, MF.USED_SUBGOAL)
                    else:
                        self.map.remove_feature_at(x, y, MF.SUBGOAL)
                        self.map.add_feature_at(x, y, MF.USED_SUBGOAL)
            self.map.active_subgoals = np.array(state["active_subgoals"], dtype=int).reshape(-1, 2)

        self.positions_path = [tuple(self.position)]
        self.tile_path = [tuple(self.position)]
        self.noise_path = []

        return self.get_observation(), self.get_info()

    @property
    def terminated(self) -> bool:
        """True iff the episode has ended (internally; does not capture, e.g., gym.TimeLimit)"""
        return self.termination_reason is not None

    @property
    def current_tile(self) -> tuple[int, int]:
        """Returns the current tile (in map coordinates) the agent occupies."""
        # Uses the clamped `global_position`: on the terminating step the agent can be one
        # cell outside the map, which would otherwise select a non-existent tile.
        return get_tile_from_global(*self.global_position, self.tile_width, self.tile_height)

    @property
    def global_position(self) -> np.ndarray:
        """Global (x, y) position of the agent, clamped into the map."""
        # On the terminating step the raw position can be one cell out of bounds -> clip
        return np.clip(self.position, [0, 0], [self.map.width - 1, self.map.height - 1])

    @property
    def position_in_tile(self) -> np.ndarray:
        """Local position of the agent within the current tile."""
        # Clamped: an out-of-bounds raw position could otherwise wrap to the opposite tile edge.
        return self.global_position % np.array([self.tile_width, self.tile_height])  # type: ignore[no-any-return]

    def get_obs_window_boundaries(
        self,
        position: Annotated[NDArray[np.int_], Literal[2]] | None = None,
    ) -> tuple[int, int, int, int]:
        """
        Returns the boundaries of the observation window as
        `(top_left_x, top_left_y, bottom_right_x, bottom_right_y)`.

        Args:
            position: Position to take the window around (default: the agent's position).
                Clamped into the map (see [`current_tile`][pgtg.environment.PGTGEnv.current_tile]):
                the agent can be one cell out of bounds on the terminating step,
                so the window is what would be observed from the last valid cell.
                A non-default position gives the window that *would* be observed from there.
        """
        if position is None:
            position = self.position
        pos_x, pos_y = np.clip(position, [0, 0], [self.map.width - 1, self.map.height - 1])

        if self.obs_context.sliding:
            return (
                pos_x - self.obs_context.half_window_x,
                pos_y - self.obs_context.half_window_y,
                pos_x + self.obs_context.half_window_x,
                pos_y + self.obs_context.half_window_y,
            )
        tile_x, tile_y = get_tile_from_global(pos_x, pos_y, self.tile_width, self.tile_height)
        return (
            tile_x * self.tile_width,
            tile_y * self.tile_height,
            tile_x * self.tile_width + self.tile_width - 1,
            tile_y * self.tile_height + self.tile_height - 1,
        )

    def get_observation(self) -> dict[str, Any]:
        """Returns the current observation visible to the agent.

        Returns:
            An element from within the observation space.
        """
        with TimedExecution("get_observation", PGTGLogger.te_cb, "pgtg"):
            obs_window_bounds = self.get_obs_window_boundaries()
            map_cutout = self.map.get_map_cutout(
                *obs_window_bounds,
                fill_squares_outside_map_with=MF.WALL,
            )

            # Build all one-hot observation windows
            oh_maps = self.__encode_obs_onehot(map_cutout)

            # Traffic is a special case
            if self.obs_context.is_observed(MF.TRAFFIC):
                mask, rel_coords = get_spatial_mask(
                    self._car_positions,
                    (obs_window_bounds[0], obs_window_bounds[2] + 1),
                    (obs_window_bounds[1], obs_window_bounds[3] + 1),
                    offset_max=False,
                )
                valid_coords = rel_coords[mask].astype(int)
                if self.logger.isEnabledFor(logging.DEBUG):  # only iterate cars if logging enabled
                    self.logger.debug(
                        "Observation: %d cars in total, %d cars in observation window",
                        len(self._car_positions),
                        sum(mask),
                    )

                oh_maps[MF.TRAFFIC][valid_coords[:, 0], valid_coords[:, 1]] = 1

            # Aggregate all windows and append any additional features
            return self.obs_context.gen_observation(self, oh_maps)

    def __encode_obs_onehot(
        self,
        map_cutout: np.ndarray,
    ) -> dict[Any, np.ndarray]:
        cutout = np.asarray(map_cutout, dtype=np.int64)
        phase = self.get_traffic_light_phase()

        oh_maps: dict[Any, np.ndarray] = {}
        for feature in self.obs_context.obs_features:
            if feature == MF.TRAFFIC:
                oh_maps[feature] = np.zeros(cutout.shape, dtype=np.int8)
                continue
            present = ((cutout & int(feature)) != 0).astype(np.int8)
            if feature == MF.TRAFFIC_LIGHT:
                for sc in (SC.GREEN, SC.YELLOW, SC.RED):
                    oh_maps[sc] = present if sc == phase else np.zeros_like(present)
            else:
                oh_maps[feature] = present

        return oh_maps

    def get_info(self, level: str | None = None) -> dict[str, Any]:
        """Returns additional information about the state of the environment.

        Args:
            level: Overrides the env's `info_level` for this call. `'essential'`
                returns cheap scalars only, `'restorable'` additionally returns the
                full episode state consumed by
                [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state],
                `'none'` returns an empty dict. `None` (default) uses the env's `info_level`.
        """

        with TimedExecution("get_info", PGTGLogger.te_cb, "pgtg"):
            level = self.info_level if level is None else str(level).lower()
            if level == "none":
                return {}

            tile_x, tile_y = np.clip(
                self.current_tile,
                [0, 0],
                [self.map.tiles_hor - 1, self.map.tiles_ver - 1],
            )

            tile_exits = self.map_plan.tiles[tile_y][tile_x]["exits"]
            current_tile_type = "".join(str(_exit) for _exit in tile_exits)

            info: dict[str, Any] = {
                "x": int(self.position[0]),
                "y": int(self.position[1]),
                "x_velocity": int(self.velocity[0]),
                "y_velocity": int(self.velocity[1]),
                "flat_tire": bool(self.flat_tire),
                "current_tile_type": current_tile_type,
                "termination_reason": self.termination_reason.value
                if self.termination_reason
                else None,
                "subgoal_completion_rate": (
                    self.subgoals_reached / self.map.num_subgoals if self.map.num_subgoals else 0.0
                ),
                "subgoals_reached": self.subgoals_reached,
            }

            if level != "restorable":
                return info

            # Episode progress needed to make set_to_state() a faithful restore.
            info.update(
                {
                    "visited_set": sorted((int(x), int(y)) for x, y in self.visited_set),
                    "active_subgoals": self.map.active_subgoals.tolist(),
                    "cars": [car.model_dump(mode="json") for car in self.cars],
                    "driver_profile_stats": self.get_driver_profile_stats(),
                    "traffic_rules": {
                        "active_rules": list(self.rule_engine.rules),
                        "triggered_rules": [r.name for r in self.rule_engine.triggered_rules],
                        "braking_applied": bool(self.rule_engine.triggered_rules),
                        "agent_direction": self.get_agent_direction_string(),
                    },
                    "rng_states": self.rng_states.to_dict(),
                    "obstacle_rng_states": {
                        int(handler.observation_feature): handler.rng.bit_generator.state
                        for handler in self._rng_handlers
                    },
                    "traffic_light_phase_counter": self._traffic_light.phase_counter,
                }
            )
            return info

    def applicable_actions(self) -> list[int]:
        """The applicable actions. Constant for PGTG, except on a terminated env.

        Returns:
            applicable_actions: A list of applicable actions.
        """

        if not self.terminated:
            return list(range(9))
        return []

__init__

__init__(
    map_path: str | None = None,
    *,
    render_mode: str | None = None,
    only_render_obs: bool = False,
    traffic_density: float = 0.0,
    traffic_light_phases_duration: tuple[int, int, int] = (
        10,
        3,
        10,
    ),
    ignore_traffic_collisions: bool = False,
    car_spawn_exclusion_radius: int = 1,
    observation_context: dict[str, Any]
    | Iterable[str | MapFeature | int]
    | ObservationContext
    | None = None,
    reward_context: dict[str, Any]
    | RewardContext
    | None = None,
    random_map_context: dict[str, Any]
    | RandomMapContext
    | None = None,
    obstacles: Sequence[SteppableFeature] | None = None,
    driver_profile_weights: dict[DriverProfile | str, float]
    | None = None,
    driver_profile_config: DriverProfileConfig
    | dict
    | str
    | None = None,
    traffic_rules_config: TrafficRulesConfig
    | dict
    | str
    | None = None,
    rules_active: list[str] | None = None,
    tiles: str | None = None,
    obstacle_masks: str | None = None,
    traffic_lanes: str | None = None,
    info_level: str | None = "essential",
    seed: int | None = None,
    log_level: int | None = None,
)

Initialize the PGTG environment; usually called via gymnasium.make().

Parameters:

Name	Type	Description	Default
map_path	str | None	Path to a map .json file, or the name of a bundled benchmark map (see pgtg.map.defaults.DEFAULT_MAPS). With None, a random map is generated each episode according to the random_map_context.	None
render_mode	str | None	One of None, 'human', 'rgb_array', or 'pil_image'.	None
only_render_obs	bool	Render the agent's observation instead of the entire map: the view is cropped to the observation window and only features in observation_context.obs_features are drawn. Applies to all render modes.	False
traffic_density	float	Fraction of car-lane cells initially occupied by cars. 0.0 disables traffic.	0.0
traffic_light_phases_duration	tuple[int, int, int]	Steps spent in the (green, yellow, red) phases, respectively.	(10, 3, 10)
ignore_traffic_collisions	bool	If true, colliding with traffic does not terminate the episode.	False
car_spawn_exclusion_radius	int	Chebyshev radius around the agent in which cars never spawn. 1 excludes the agent's cell plus its 8 neighbors; negative values disable the exclusion. 1 is the minimum necessary value to exclude unavoidable collisions with newly spawned cars (cars move first).	1
observation_context	dict[str, Any] | Iterable[str | MapFeature | int] | ObservationContext | None	ObservationContext, dict of its fields, or a bare iterable of observed features (shorthand for {"obs_features": ...}). Controls window size/sliding and which channels the agent observes.	None
reward_context	dict[str, Any] | RewardContext | None	RewardContext or dict of its fields (rewards & penalties).	None
random_map_context	dict[str, Any] | RandomMapContext | None	RandomMapContext or dict of its fields; used only when map_path is None.	None
obstacles	Sequence[SteppableFeature] | None	Explicit obstacle handlers. Defaults for all generatable obstacles are filled in automatically; handlers given here take precedence.	None
driver_profile_weights	dict[DriverProfile | str, float] | None	Mapping of driver profile to sampling weight. Overrides the weights from driver_profile_config.	None
driver_profile_config	DriverProfileConfig | dict | str | None	DriverProfileConfig, dict, or path to a .json/.yaml file defining driver behaviors and weights.	None
traffic_rules_config	TrafficRulesConfig | dict | str | None	TrafficRulesConfig, dict, or path to a .json/.yaml file with additional traffic rules (merged with the built-in defaults).	None
rules_active	list[str] | None	Names of the loaded traffic rules to activate. None activates all loaded rules, an empty list none.	None
tiles	str | None	Path to a custom tiles .json/.yaml file overriding the default tile templates. Tile size is inferred from the file; all three tile-data files must share the same dimensions, so non-default sizes require matching obstacle_masks and traffic_lanes files.	None
obstacle_masks	str | None	Path to a custom obstacle-masks file.	None
traffic_lanes	str | None	Path to a custom traffic-lanes file.	None
info_level	str | None	How much information get_info() (and thus the info dict of reset()/ step()) contains. 'essential' (default) returns cheap scalars only (position, velocity, flat tire, tile type, termination reason, subgoal progress). 'restorable' additionally returns the full episode state needed by set_to_state() (cars, RNG states, visited set, active subgoals, driver/traffic-rule stats). 'none'/None makes get_info() return an empty dict; step() still adds cost and is_success.	'essential'
seed	int | None	Master seed for all randomness. reset(seed=...) restarts the episode sequence deterministically.	None
log_level	int | None	Level of the "pgtg" logger. None (default) leaves it as it is (logging.WARNING until something changes it). The logger is process-wide, so passing a level here affects every PGTG environment.	None

Source code in pgtg/environment.py

def __init__(
    self,
    map_path: str | None = None,
    *,
    render_mode: str | None = None,
    only_render_obs: bool = False,
    traffic_density: float = 0.0,
    traffic_light_phases_duration: tuple[int, int, int] = (10, 3, 10),
    ignore_traffic_collisions: bool = False,
    car_spawn_exclusion_radius: int = 1,
    observation_context: (
        dict[str, Any] | Iterable[str | MF | int] | ObservationContext | None
    ) = None,
    reward_context: dict[str, Any] | RewardContext | None = None,
    random_map_context: dict[str, Any] | RandomMapContext | None = None,
    obstacles: Sequence[SteppableFeature] | None = None,
    driver_profile_weights: dict[DriverProfile | str, float] | None = None,
    driver_profile_config: DriverProfileConfig | dict | str | None = None,
    traffic_rules_config: TrafficRulesConfig | dict | str | None = None,
    rules_active: list[str] | None = None,
    tiles: str | None = None,
    obstacle_masks: str | None = None,
    traffic_lanes: str | None = None,
    info_level: str | None = "essential",
    seed: int | None = None,
    log_level: int | None = None,
):
    """Initialize the PGTG environment; usually called via `gymnasium.make()`.

    Args:
        map_path: Path to a map .json file, or the name of a bundled benchmark map
            (see `pgtg.map.defaults.DEFAULT_MAPS`). With `None`, a random map is
            generated each episode according to the
            [`random_map_context`][pgtg.contexts.RandomMapContext].
        render_mode: One of `None`, `'human'`, `'rgb_array'`, or `'pil_image'`.
        only_render_obs: Render the agent's observation instead of the entire map:
            the view is cropped to the observation window *and* only features in
            `observation_context.obs_features` are drawn. Applies to all render modes.
        traffic_density: Fraction of car-lane cells initially occupied by cars.
            `0.0` disables traffic.
        traffic_light_phases_duration: Steps spent in the (green, yellow, red) phases,
            respectively.
        ignore_traffic_collisions: If true, colliding with traffic does not terminate the
            episode.
        car_spawn_exclusion_radius: Chebyshev radius around the agent in which cars
            never spawn. `1` excludes the agent's cell plus its 8 neighbors; negative
            values disable the exclusion. `1` is the minimum necessary value to exclude
            unavoidable collisions with newly spawned cars (cars move first).
        observation_context: [`ObservationContext`][pgtg.contexts.ObservationContext],
            dict of its fields, or a bare iterable of observed features (shorthand for
            `{"obs_features": ...}`). Controls window size/sliding and which channels
            the agent observes.
        reward_context: [`RewardContext`][pgtg.contexts.RewardContext] or dict of its fields
            (rewards & penalties).
        random_map_context:  [`RandomMapContext`][pgtg.contexts.RandomMapContext] or dict of
            its fields; used only when `map_path` is `None`.
        obstacles: Explicit obstacle handlers. Defaults for all generatable obstacles
            are filled in automatically; handlers given here take precedence.
        driver_profile_weights: Mapping of driver profile to sampling weight.
            Overrides the weights from `driver_profile_config`.
        driver_profile_config:
            [`DriverProfileConfig`][pgtg.traffic.drivers.DriverProfileConfig], dict,
            or path to a .json/.yaml file defining driver behaviors and weights.
        traffic_rules_config: [`TrafficRulesConfig`][pgtg.traffic.rules.TrafficRulesConfig],
            dict, or path to a .json/.yaml file with additional traffic rules
            (merged with the built-in defaults).
        rules_active: Names of the loaded traffic rules to activate. `None` activates
            all loaded rules, an empty list none.
        tiles: Path to a custom tiles .json/.yaml file overriding the default tile
            templates. Tile size is inferred from the file; all three tile-data files
            must share the same dimensions, so non-default sizes require matching
            `obstacle_masks` and `traffic_lanes` files.
        obstacle_masks: Path to a custom obstacle-masks file.
        traffic_lanes: Path to a custom traffic-lanes file.
        info_level: How much information [`get_info()`][pgtg.environment.PGTGEnv.get_info]
            (and thus the info dict of [`reset()`][pgtg.environment.PGTGEnv.reset]/
            [`step()`][pgtg.environment.PGTGEnv.step]) contains. `'essential'` (default)
            returns cheap scalars only (position, velocity, flat tire, tile type, termination
            reason, subgoal progress). `'restorable'` additionally returns the
            full episode state needed by
            [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state]
            (cars, RNG states, visited set, active subgoals, driver/traffic-rule stats).
            `'none'`/`None` makes [`get_info()`][pgtg.environment.PGTGEnv.get_info]
            return an empty dict; [`step()`][pgtg.environment.PGTGEnv.step] still adds `cost`
            and `is_success`.
        seed: Master seed for all randomness.
            [`reset(seed=...)`][pgtg.environment.PGTGEnv.reset] restarts the episode sequence
            deterministically.
        log_level: Level of the `"pgtg"` logger. `None` (default) leaves it as it is
            (`logging.WARNING` until something changes it). The logger is process-wide,
            so passing a level here affects every PGTG environment.
    """
    self.__setup_logger(log_level)

    # Master entropy source. Every reset() spawns fresh child streams from this
    # sequence of *different* episodes, while reset(seed=...) restarts the sequence.
    # => Episodes only depend on their seed and number; not on previous episode behavior.
    self.seed = seed
    self._seed_seq = np.random.SeedSequence(seed)
    self.rng_states = PGTGEnv.RNGStates.from_seed(seed)

    # Register all obstacle handlers:
    # - because they are generatable, or
    # - because they were explicitly passed.
    explicit = list(obstacles) if obstacles is not None else []
    covered = {
        int(o.observation_feature) for o in explicit if o.observation_feature is not None
    }
    auto_filled = [
        factory()
        for factory in ObstacleRegistry.generatable()
        if int(factory.observation_feature) not in covered
    ]
    self.obstacles: list[SteppableFeature] = [*explicit, *auto_filled]
    self._default_obstacles = obstacles is None

    # Resolve the tile data once (paths -> preprocessed datasets);
    # None = the packaged default file. The loaders memoize per file,
    # so the datasets are shared and built once, whether default or custom.
    self._tiles = tile_registry.tiles_from_file(tiles) if tiles is not None else None
    self._obstacle_masks = (
        tile_registry.obstacle_masks_from_file(obstacle_masks)
        if obstacle_masks is not None
        else None
    )
    self._traffic_lanes = (
        tile_registry.traffic_lanes_from_file(traffic_lanes)
        if traffic_lanes is not None
        else None
    )
    # Infer tile size from tile data and validate three datasets share the same dims
    self.tile_width, self.tile_height = tile_registry.resolve_tile_dims(
        tiles=self._tiles,
        obstacle_masks=self._obstacle_masks,
        traffic_lanes=self._traffic_lanes,
    )

    # Only record requested info dict entries to avoid expensive computations
    # (e.g., when no restorations are needed)
    self.info_level = "none" if info_level is None else str(info_level).lower()
    if self.info_level not in ("none", "essential", "restorable"):
        raise ValueError(
            f"info_level must be one of 'none', 'essential', 'restorable' or None, "
            f"got {info_level!r}"
        )

    # Resolve map path.
    # - a default map name resolved to its bundled .json
    # - a custom map path to a custom .json
    # - None: generate a random map each episode
    if map_path is not None and map_path in DEFAULT_MAPS:
        self.map_path = str(DEFAULT_MAPS[map_path])
    else:
        self.map_path = map_path
    if self.map_path is not None:
        self.logger.debug("Loading map from path %s", self.map_path)
        self.map_plan = json_file_to_map_plan(self.map_path)
        self.map = EpisodeMap(
            self.map_plan,
            tiles=self._tiles,
            obstacle_masks=self._obstacle_masks,
            traffic_lanes=self._traffic_lanes,
        )

    self.rm_context = RandomMapContext.model_validate(
        (random_map_context or {}),
        context={"obstacles": self.obstacles},
    )

    # Each branch decides which obstacle features can occur in this env.
    # It yields both the auto-observe set and which default handlers are worth keeping around
    # `keep_features = None` means "keep every handler".
    if self.map_path is not None:
        # Only add obstacles to the auto-observe if they are generated on this *fixed* map
        present = self.map.feature_union()
        self._generatable_here = {
            int(o.observation_feature)
            for o in self.obstacles
            if o.observation_feature is not None and int(o.observation_feature) & present
        }
        keep_features = self._generatable_here
    elif self.rm_context.obstacle_probability > 0:
        # Only add obstacles generally generatable on the *random* maps to the obs space
        self._generatable_here = {
            int(feature)
            for feature, weight in self.rm_context.obs_feature_weights.items()
            if weight > 0
        }
        # Zero-weight handlers stay registered even though they are not auto-observed
        keep_features = None
    else:
        # No obstacles generatable, empty set for auto-observe
        self._generatable_here = set()
        keep_features = set()

    # Construct reward and observation contexts
    self.reward_context = RewardContext.model_validate(reward_context or {})
    self.obs_context = ObservationContext.model_validate(
        observation_context or {},
        context={
            "obstacles": self.obstacles,
            "tile_dims": (self.tile_width, self.tile_height),
            "generatable_here": self._generatable_here,
        },
    )

    # Drop default handlers that can never trigger here, so they cost nothing per step.
    # Applied only after the observation context is built: an obstacle with observe=True
    # keeps its channel even when this env can never produce it.
    if self._default_obstacles and keep_features is not None:
        self.obstacles = [
            obstacle
            for obstacle in self.obstacles
            if obstacle.observation_feature is not None
            and int(obstacle.observation_feature) in keep_features
        ]

    # Generate observation_space and action_space for gymnasium API
    if self.map_path is not None:
        # Fixed size
        self.observation_space = self.obs_context.gen_obs_space(
            self.map.width,
            self.map.height,
        )
    else:
        # Size inferred by tile_dim * map_dim
        self.observation_space = self.obs_context.gen_obs_space(
            self.rm_context.width * self.tile_width,
            self.rm_context.height * self.tile_height,
        )
    self.action_space = spaces.Discrete(9)
    self.reward_range = (-np.inf, np.inf)

    # Internal state: Traffic rules to apply *on each step*
    self.rule_engine = TrafficRuleEngine()
    self._load_traffic_rules(traffic_rules_config, rules_active)

    # Internal state: traffic lights. Prepend so their check runs before any random obstacle
    self._traffic_light = TrafficLight(phases_duration=traffic_light_phases_duration)
    self.obstacles.insert(0, self._traffic_light)

    # Internal state: Every obstacle keeps track of its own RNG state. Deterministic
    # handlers (the traffic light) own no stream and stay out of the seeding sequence.
    # NOTE: rebuild if `self.obstacles` is mutated later.
    self._rng_handlers: list[HasRng] = [h for h in self.obstacles if isinstance(h, HasRng)]
    self._index_obstacle_handlers()  # faster access

    self.traffic_density = traffic_density
    self.ignore_traffic_collisions = ignore_traffic_collisions
    self.car_spawn_exclusion_radius = int(car_spawn_exclusion_radius)

    # Warnings: Emit warnings for any obstacle that is generatable but not observable:
    #           The agent is probably unable learn to avoid such obstacles
    if self.traffic_density > 0 and not self.obs_context.is_observed(MF.TRAFFIC):
        self.logger.warning(
            "traffic_density=%s spawns cars, but MF.TRAFFIC is not in the observation "
            "features %s — cars will be invisible to the agent yet can still terminate "
            "the episode on collision.",
            self.traffic_density,
            self.obs_context.obs_features,
        )

    for feature in self._generatable_here:
        if not self.obs_context.is_observed(feature):
            self.logger.warning(
                "'%s' can appear in this environment but is not in the observation "
                "features %s — an agent will not be able to learn to avoid it.",
                FeatureRegistry.describe(feature),
                self.obs_context.obs_features,
            )

    # Internal state: Driver profiles for different driver behaviors; default and custom
    if driver_profile_config is None:
        self._driver_profile_config = DriverProfileConfig.defaults()
    else:
        self._driver_profile_config = DriverProfileConfig.model_validate(driver_profile_config)

    if driver_profile_weights is not None:
        normalized_weights = {
            (p.value if isinstance(p, DriverProfile) else str(p)): w
            for p, w in driver_profile_weights.items()
        }
        self._driver_profile_config = self._driver_profile_config.model_copy(
            update={"weights": normalized_weights}
        )

    self.driver_profile_dist = self._driver_profile_config.normalized_weights
    self._refresh_profile_sampler()

    # Some irrelevant rendering state
    self.render_mode = render_mode
    self.only_render_obs = only_render_obs
    self.window_size = 720
    self.window = None
    self.clock = None

    # Start the first episode, so a freshly constructed env is immediately steppable
    self._start_episode(np.random.SeedSequence(self._seed_seq.entropy))

reset

reset(
    *, seed: int | None = None, **kwargs: Any
) -> tuple[dict, dict[str, Any]]

Starts a fresh episode, possibly reinitializing the seed sequence for future episodes.

Parameters:

Name	Type	Description	Default
seed	int | None	Optional seed to reseed all associated rng streams.	None

Returns:

Name	Type	Description
observation	dict	The resulting observation after the reset.
info	dict[str, Any]	Additional information about the state of the environment. Analogous to the info returned by step().

Source code in pgtg/environment.py

def reset(
    self,
    *,
    seed: int | None = None,
    **kwargs: Any,
) -> tuple[dict, dict[str, Any]]:
    """Starts a fresh episode, possibly reinitializing the seed sequence for future episodes.

    Args:
        seed: Optional seed to reseed all associated rng streams.

    Returns:
        observation: The resulting observation after the reset.
        info: Additional information about the state of the environment. Analogous to the info
            returned by step().
    """
    with TimedExecution("reset", PGTGLogger.te_cb, "pgtg"):
        self.logger.debug("Resetting environment.")
        super().reset(seed=seed, **kwargs)

        if seed is not None:
            self.seed = seed
            self._seed_seq = np.random.SeedSequence(seed)
        self._start_episode(self._seed_seq)

        if self.render_mode == "human":
            self.render()
        return (self.get_observation(), self.get_info())

step

step(
    action: int,
) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]

Performs the given action and returns the results.

Parameters:

Name	Type	Description	Default
action	int	The action to perform.	required

Returns:

Name	Type	Description
observation	dict	The resulting observation after the transition.
reward	SupportsFloat	The reward for the action.
terminated	bool	Whether or not the episode has ended.
truncated	bool	Always False. No native truncation.
info	dict[str, Any]	Additional information about the state of the environment. Analogous to the info returned by reset().

Source code in pgtg/environment.py

def step(
    self,
    action: int,
) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]:
    """Performs the given action and returns the results.

    Args:
        action: The action to perform.

    Returns:
        observation: The resulting observation after the transition.
        reward: The reward for the action.
        terminated: Whether or not the episode has ended.
        truncated: Always `False`. No native truncation.
        info: Additional information about the state of the environment.
            Analogous to the info returned by reset().
    """
    with TimedExecution("step", PGTGLogger.te_cb, "pgtg"):
        if self.terminated:
            raise RuntimeError("Already done, step has no further effect")

        ### STEP SETUP
        self._traffic_light.tick()
        self.rule_engine.triggered_rules.clear()

        ### CARS MOVE FIRST ###
        # `_occupied` persists across steps: the loop below keeps it in sync
        new_cars = []
        occupied = self._occupied
        # One batched draw for all cars' (delay, speed) decisions
        draws = self.rng_states.car_rng.random(2 * len(self.cars)).tolist() if self.cars else []
        for i, car in enumerate(self.cars):
            next_position_and_route = self._get_next_car_position_and_route(
                car,
                occupied,
                draws[2 * i],
                draws[2 * i + 1],
            )

            occupied[car.position] -= 1
            if next_position_and_route is None:
                replacement = self._spawn_new_car()
                if replacement is None:
                    continue  # no usable spawner: the car leaves without a successor
                new_cars.append(replacement)
            else:
                car.position, car.route = next_position_and_route
                new_cars.append(car)
            occupied[new_cars[-1].position] += 1

        self.cars = new_cars
        self._update_car_positions_array()

        ### AGENT STEP ###
        # Fill local context, so obstacles and traffic rules can modify dynamics
        local_ctx = StepContext(
            current_position=self.position.copy(),
            acceleration=np.array(ACTIONS_TO_ACCELERATION[action]),
        )
        self.velocity = self.velocity + local_ctx.acceleration
        local_ctx.remaining_substeps = [*self._decompose_velocity(), None]

        # Always rebound by the loop below (it runs at least once, for the None sentinel)
        current_pos_t: tuple[int, int] = tuple(local_ctx.current_position)
        while local_ctx.remaining_substeps:
            # queue with pop so handlers can interrupt movements
            velocity_substep = local_ctx.remaining_substeps.pop(0)
            local_ctx.velocity_substep = velocity_substep
            current_pos_t = tuple(local_ctx.current_position)

            # Verify termination conditions
            if not self.map.inside_map(*current_pos_t):
                self.termination_reason = TerminationReason.OUT_OF_BOUNDS

            elif self.map.feature_at(*current_pos_t, MF.WALL):
                self.termination_reason = TerminationReason.WALL_COLLISION

            elif not self.ignore_traffic_collisions and occupied[current_pos_t] > 0:
                self.termination_reason = TerminationReason.TRAFFIC_COLLISION

            if self.termination_reason is not None:  # same penalty for all crashes
                local_ctx.sum_penalty += self.reward_context.crash_penalty
                break

            if self.map.feature_at(*current_pos_t, MF.FINAL_GOAL):
                local_ctx.sum_reward += (
                    self.reward_context.individual_subgoal_reward(self.map.num_subgoals)
                    + self.reward_context.final_goal_bonus
                )
                self.subgoals_reached += 1
                local_ctx.is_success = True
                self.termination_reason = TerminationReason.GOAL
                break

            # Subgoal reward handling; on final goal this was applied already as well
            if self.map.feature_at(*current_pos_t, MF.SUBGOAL):
                local_ctx.sum_reward += self.reward_context.individual_subgoal_reward(
                    self.map.num_subgoals,
                )
                self.subgoals_reached += 1
                self.map.set_subgoals_to_used(*current_pos_t)

            # This was just the verification for the last movement step (None sentinel)
            if local_ctx.velocity_substep is None:
                continue

            # Walk all obstacles and apply their effects. Returning true, means the step loop
            # is interrupted. Note, that if the obstacle moves the agent, it should not return
            # true but instead set remaining_substeps to `[None]` to verify the last move
            if self._dispatch_map_features(local_ctx):
                break

            # Traffic rules are evaluated at the cell being left, after this cell's
            # rewards/obstacles but before moving. Same return contract as above for obstacles.
            if self.rule_engine.apply_rules(self, local_ctx):
                break

            # Finally move the agent one substep along the decomposed velocity
            local_ctx.current_position += local_ctx.velocity_substep
            current_pos_t = tuple(local_ctx.current_position)
            self.tile_path.append(current_pos_t)

        if self.flat_tire:
            self.velocity = np.zeros(2, dtype=int)

        old_pos_tuple = tuple(self.position)
        moved = old_pos_tuple != current_pos_t
        # Standing still means *choosing* to stay put: no acceleration and no movement.
        if not moved:
            if not local_ctx.acceleration.any():  # Only discourage neutral action
                local_ctx.sum_penalty += self.reward_context.standing_still_penalty
        elif current_pos_t in self.visited_set:
            local_ctx.sum_penalty += self.reward_context.already_visited_penalty

        self.position = local_ctx.current_position
        self.visited_set.add(current_pos_t)
        self.positions_path.append(tuple(self.position))

        if self.render_mode == "human":
            self.render()

        info = self.get_info()
        info["cost"] = local_ctx.sum_penalty
        info["is_success"] = local_ctx.is_success
        reward = (
            local_ctx.sum_reward
            if self.reward_context.separate_reward_cost
            else local_ctx.sum_reward - local_ctx.sum_penalty
        )

        return (self.get_observation(), reward, self.terminated, False, info)

render

render() -> Image | npt.NDArray | None

Returns a rendered representation of the game according to the render mode.

In 'human' mode this draws the current frame into the pygame window and returns None. reset() and step() automatically call render().

Returns:

Type	Description
Image | NDArray | None	The rendered representation, or None if the render mode is "human" or None.

Source code in pgtg/environment.py

def render(self) -> Image | npt.NDArray | None:  # type: ignore[override]
    """Returns a rendered representation of the game according to the render mode.

    In `'human'` mode this draws the current frame into the pygame window and
    returns `None`. [`reset()`][pgtg.environment.PGTGEnv.reset] and
    [`step()`][pgtg.environment.PGTGEnv.step] automatically call `render()`.

    Returns:
        The rendered representation, or `None` if the render mode is `"human"` or `None`.
    """

    match self.render_mode:
        case None:
            return None
        case "human":
            self._render_frame_for_human()
            return None
        case "rgb_array":
            return np.asarray(
                self.generate_frame(only_observation=self.only_render_obs).convert("RGB")
            )
        case "pil_image":
            return self.generate_frame(only_observation=self.only_render_obs)
        case _:
            raise ValueError(
                f"render_mode={self.render_mode!r} is not supported; expected one of "
                f"None, {', '.join(map(repr, self.metadata['render_modes']))}."
            )

current_tile property

current_tile: tuple[int, int]

Returns the current tile (in map coordinates) the agent occupies.

global_position property

global_position: ndarray

Global (x, y) position of the agent, clamped into the map.

position_in_tile property

position_in_tile: ndarray

Local position of the agent within the current tile.

get_agent_direction_string

get_agent_direction_string() -> str

Returns the agent's current direction as used by the traffic rules.

Source code in pgtg/environment.py

def get_agent_direction_string(self) -> str:
    """Returns the agent's current direction as used by the traffic rules."""
    return self.rule_engine.get_agent_direction(self)

get_observation

get_observation() -> dict[str, Any]

Returns the current observation visible to the agent.

Returns:

Type	Description
dict[str, Any]	An element from within the observation space.

Source code in pgtg/environment.py

def get_observation(self) -> dict[str, Any]:
    """Returns the current observation visible to the agent.

    Returns:
        An element from within the observation space.
    """
    with TimedExecution("get_observation", PGTGLogger.te_cb, "pgtg"):
        obs_window_bounds = self.get_obs_window_boundaries()
        map_cutout = self.map.get_map_cutout(
            *obs_window_bounds,
            fill_squares_outside_map_with=MF.WALL,
        )

        # Build all one-hot observation windows
        oh_maps = self.__encode_obs_onehot(map_cutout)

        # Traffic is a special case
        if self.obs_context.is_observed(MF.TRAFFIC):
            mask, rel_coords = get_spatial_mask(
                self._car_positions,
                (obs_window_bounds[0], obs_window_bounds[2] + 1),
                (obs_window_bounds[1], obs_window_bounds[3] + 1),
                offset_max=False,
            )
            valid_coords = rel_coords[mask].astype(int)
            if self.logger.isEnabledFor(logging.DEBUG):  # only iterate cars if logging enabled
                self.logger.debug(
                    "Observation: %d cars in total, %d cars in observation window",
                    len(self._car_positions),
                    sum(mask),
                )

            oh_maps[MF.TRAFFIC][valid_coords[:, 0], valid_coords[:, 1]] = 1

        # Aggregate all windows and append any additional features
        return self.obs_context.gen_observation(self, oh_maps)

get_info

get_info(level: str | None = None) -> dict[str, Any]

Returns additional information about the state of the environment.

Parameters:

Name	Type	Description	Default
level	str | None	Overrides the env's info_level for this call. 'essential' returns cheap scalars only, 'restorable' additionally returns the full episode state consumed by set_to_state(), 'none' returns an empty dict. None (default) uses the env's info_level.	None

Source code in pgtg/environment.py

def get_info(self, level: str | None = None) -> dict[str, Any]:
    """Returns additional information about the state of the environment.

    Args:
        level: Overrides the env's `info_level` for this call. `'essential'`
            returns cheap scalars only, `'restorable'` additionally returns the
            full episode state consumed by
            [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state],
            `'none'` returns an empty dict. `None` (default) uses the env's `info_level`.
    """

    with TimedExecution("get_info", PGTGLogger.te_cb, "pgtg"):
        level = self.info_level if level is None else str(level).lower()
        if level == "none":
            return {}

        tile_x, tile_y = np.clip(
            self.current_tile,
            [0, 0],
            [self.map.tiles_hor - 1, self.map.tiles_ver - 1],
        )

        tile_exits = self.map_plan.tiles[tile_y][tile_x]["exits"]
        current_tile_type = "".join(str(_exit) for _exit in tile_exits)

        info: dict[str, Any] = {
            "x": int(self.position[0]),
            "y": int(self.position[1]),
            "x_velocity": int(self.velocity[0]),
            "y_velocity": int(self.velocity[1]),
            "flat_tire": bool(self.flat_tire),
            "current_tile_type": current_tile_type,
            "termination_reason": self.termination_reason.value
            if self.termination_reason
            else None,
            "subgoal_completion_rate": (
                self.subgoals_reached / self.map.num_subgoals if self.map.num_subgoals else 0.0
            ),
            "subgoals_reached": self.subgoals_reached,
        }

        if level != "restorable":
            return info

        # Episode progress needed to make set_to_state() a faithful restore.
        info.update(
            {
                "visited_set": sorted((int(x), int(y)) for x, y in self.visited_set),
                "active_subgoals": self.map.active_subgoals.tolist(),
                "cars": [car.model_dump(mode="json") for car in self.cars],
                "driver_profile_stats": self.get_driver_profile_stats(),
                "traffic_rules": {
                    "active_rules": list(self.rule_engine.rules),
                    "triggered_rules": [r.name for r in self.rule_engine.triggered_rules],
                    "braking_applied": bool(self.rule_engine.triggered_rules),
                    "agent_direction": self.get_agent_direction_string(),
                },
                "rng_states": self.rng_states.to_dict(),
                "obstacle_rng_states": {
                    int(handler.observation_feature): handler.rng.bit_generator.state
                    for handler in self._rng_handlers
                },
                "traffic_light_phase_counter": self._traffic_light.phase_counter,
            }
        )
        return info

applicable_actions

applicable_actions() -> list[int]

The applicable actions. Constant for PGTG, except on a terminated env.

Returns:

Name	Type	Description
applicable_actions	list[int]	A list of applicable actions.

Source code in pgtg/environment.py

def applicable_actions(self) -> list[int]:
    """The applicable actions. Constant for PGTG, except on a terminated env.

    Returns:
        applicable_actions: A list of applicable actions.
    """

    if not self.terminated:
        return list(range(9))
    return []

get_obs_window_boundaries

get_obs_window_boundaries(
    position: Annotated[NDArray[int_], Literal[2]]
    | None = None,
) -> tuple[int, int, int, int]

Returns the boundaries of the observation window as (top_left_x, top_left_y, bottom_right_x, bottom_right_y).

Parameters:

Name	Type	Description	Default
position	Annotated[NDArray[int_], Literal[2]] | None	Position to take the window around (default: the agent's position). Clamped into the map (see current_tile): the agent can be one cell out of bounds on the terminating step, so the window is what would be observed from the last valid cell. A non-default position gives the window that would be observed from there.	None

Source code in pgtg/environment.py

def get_obs_window_boundaries(
    self,
    position: Annotated[NDArray[np.int_], Literal[2]] | None = None,
) -> tuple[int, int, int, int]:
    """
    Returns the boundaries of the observation window as
    `(top_left_x, top_left_y, bottom_right_x, bottom_right_y)`.

    Args:
        position: Position to take the window around (default: the agent's position).
            Clamped into the map (see [`current_tile`][pgtg.environment.PGTGEnv.current_tile]):
            the agent can be one cell out of bounds on the terminating step,
            so the window is what would be observed from the last valid cell.
            A non-default position gives the window that *would* be observed from there.
    """
    if position is None:
        position = self.position
    pos_x, pos_y = np.clip(position, [0, 0], [self.map.width - 1, self.map.height - 1])

    if self.obs_context.sliding:
        return (
            pos_x - self.obs_context.half_window_x,
            pos_y - self.obs_context.half_window_y,
            pos_x + self.obs_context.half_window_x,
            pos_y + self.obs_context.half_window_y,
        )
    tile_x, tile_y = get_tile_from_global(pos_x, pos_y, self.tile_width, self.tile_height)
    return (
        tile_x * self.tile_width,
        tile_y * self.tile_height,
        tile_x * self.tile_width + self.tile_width - 1,
        tile_y * self.tile_height + self.tile_height - 1,
    )

get_subgoal_compass_direction

get_subgoal_compass_direction(
    position: Annotated[NDArray[int_], Literal[2]]
    | None = None,
) -> tuple[int, tuple[int, int], bool]

Calculate direction indicators to the nearest subgoal.

Parameters:

Name	Type	Description	Default
position	Annotated[NDArray[int_], Literal[2]] | None	Position to calculate from. (default: the agent's current position)	None

Returns:

Name	Type	Description
compass	int	index of the nearest subgoal direction, clockwise from north: 0=N, 1=NE, 2=E, 3=SE, 4=S, 5=SW, 6=W, 7=NW. When there is no active subgoal the sentinel 8 is returned.
distances	tuple[int, int]	Tuple with the signed x and y offset to the nearest subgoal.
in_window	bool	True iff nearest subgoal is within the observation window.

Source code in pgtg/environment.py

def get_subgoal_compass_direction(
    self,
    position: Annotated[NDArray[np.int_], Literal[2]] | None = None,
) -> tuple[int, tuple[int, int], bool]:
    """Calculate direction indicators to the nearest subgoal.

    Args:
        position: Position to calculate from. (default: the agent's current position)

    Returns:
        compass: index of the nearest subgoal direction, clockwise from north:
            0=N, 1=NE, 2=E, 3=SE, 4=S, 5=SW, 6=W, 7=NW. When there is no active subgoal
            the sentinel 8 is returned.
        distances: Tuple with the signed x and y offset to the nearest subgoal.
        in_window: True iff nearest subgoal is within the observation window.
    """
    if position is None:
        position = self.position

    active_subgoals = self.map.active_subgoals
    if active_subgoals.ndim < 2 or len(active_subgoals) == 0:
        return 8, (0, 0), False

    # Signed offsets toward the nearest subgoal (positive dx = east, positive dy = south).
    deltas = active_subgoals - position
    nearest = deltas[np.argmin(np.sum(np.abs(deltas), axis=1))]
    dx, dy = int(nearest[0]), int(nearest[1])

    dir_index = int(np.round(np.arctan2(dx, -dy) / (np.pi / 4))) % 8

    sub_x, sub_y = position[0] + dx, position[1] + dy
    tl_x, tl_y, br_x, br_y = self.get_obs_window_boundaries(position)
    in_window = bool(tl_x <= sub_x <= br_x and tl_y <= sub_y <= br_y)

    return dir_index, (dx, dy), in_window

get_traffic_light_phase

get_traffic_light_phase() -> str

Returns the current traffic light phase (one of "green", "yellow", "red").

Source code in pgtg/environment.py

def get_traffic_light_phase(self) -> str:
    """Returns the current traffic light phase (one of `"green"`, `"yellow"`, `"red"`)."""
    return self._traffic_light.phase

get_driver_profile_stats

get_driver_profile_stats() -> dict[str, Any]

Returns statistics about the current driver profile distribution.

Source code in pgtg/environment.py

def get_driver_profile_stats(self) -> dict[str, Any]:
    """Returns statistics about the current driver profile distribution."""
    profile_counts = dict.fromkeys(self._driver_profile_config.behaviors, 0)

    for car in self.cars:
        if car.driver_profile in profile_counts:
            profile_counts[car.driver_profile] += 1

    total_cars = len(self.cars)
    if total_cars > 0:
        profile_percentages = {k: (v / total_cars) * 100 for k, v in profile_counts.items()}
    else:
        profile_percentages = dict.fromkeys(profile_counts.keys(), 0)

    return {
        "counts": profile_counts,
        "percentages": profile_percentages,
        "total_cars": total_cars,
        "configured_percentages": {k: v * 100 for k, v in self.driver_profile_dist.items()},
    }

light_step

light_step(
    action: int,
) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]

Copies the environment and steps the copy. The original stays unchanged.

Note: this deep-copies the entire environment (including the map grid and all cars), which is costly. Stepping and restoring with set_to_state() might be faster.

Parameters:

Name	Type	Description	Default
action	int	The action to perform.	required

Returns:

Name	Type	Description
observation	dict	The resulting observation after the transition.
reward	SupportsFloat	The reward for the action.
terminated	bool	Whether or not the episode has ended.
truncated	bool	Always False. No native truncation.
info	dict[str, Any]	Additional information about the state of the environment. Analogous to the info returned by reset().

Source code in pgtg/environment.py

def light_step(
    self,
    action: int,
) -> tuple[dict, SupportsFloat, bool, bool, dict[str, Any]]:
    """Copies the environment and steps the copy. The original stays unchanged.

    **Note:** this deep-copies the entire environment (including the map grid and all cars),
    which is costly. Stepping and restoring with
    [`set_to_state()`][pgtg.environment.PGTGEnv.set_to_state] might be faster.

    Args:
        action: The action to perform.

    Returns:
        observation: The resulting observation after the transition.
        reward: The reward for the action.
        terminated: Whether or not the episode has ended.
        truncated: Always `False`. No native truncation.
        info: Additional information about the state of the environment.
            Analogous to the info returned by reset().
    """

    env_copy = copy.deepcopy(self)
    return env_copy.step(action)

set_to_state

set_to_state(
    state: dict[str, Any],
) -> tuple[dict, dict[str, Any]]

Sets the environment to a given state.

Restores the full mutable episode state captured by get_info(): position, velocity, flat tire, RNG states, traffic-light phase, cars, subgoal consumption, the visited set, the subgoal counter, and the termination reason, so that setting two environments to the same state (built from the same map) and choosing the same actions yields the same future.

Leaving out the rng states is supported to obtain an uncertain future.

Append-only history (positions_path/tile_path/noise_path) is reset to a clean baseline at this state. The map itself is not re-created, so state must come from an environment built on the same map.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	The state to set the environment to (a dict as returned by get_info(level="restorable")).	required

Returns:

Name	Type	Description
observation	dict	The observation after setting the state.
info	dict[str, Any]	Additional information about the state of the environment.

Source code in pgtg/environment.py

def set_to_state(self, state: dict[str, Any]) -> tuple[dict, dict[str, Any]]:
    """
    Sets the environment to a given state.

    Restores the full mutable episode state captured by
    [`get_info()`][pgtg.environment.PGTGEnv.get_info]:
        position, velocity, flat tire, RNG states, traffic-light phase, cars,
        subgoal consumption, the visited set, the subgoal counter, and the termination reason,
    so that setting two environments to the same state (built from the same map) and
    choosing the same actions yields the same future.

    Leaving out the rng states is supported to obtain an uncertain future.

    Append-only history (`positions_path`/`tile_path`/`noise_path`) is reset
    to a clean baseline at this state. The map itself is not re-created, so `state`
    must come from an environment built on the same map.


    Args:
        state: The state to set the environment to (a dict as returned by
            get_info(level="restorable")).

    Returns:
        observation: The observation after setting the state.
        info: Additional information about the state of the environment.
    """

    # TODO: Possibly fix map rebuild for procedural resets.
    #       Our StateRestorationWrapper should have the code for this already. Just port it.

    missing = [key for key in ("cars", "rng_states") if key not in state]
    if missing:
        raise ValueError(
            f"set_to_state requires a 'restorable' state dict (missing keys: {missing}). "
            "Capture the state with get_info(level='restorable') or construct the "
            "environment with info_level='restorable'."
        )

    self.position = np.array([state["x"], state["y"]])
    self.velocity = np.array([state["x_velocity"], state["y_velocity"]])
    self.flat_tire = state["flat_tire"]

    if "rng_states" in state:
        self.rng_states.from_dict(state["rng_states"])

    obstacle_rng_states = state.get("obstacle_rng_states")
    if obstacle_rng_states:
        for handler in self._rng_handlers:
            saved = obstacle_rng_states.get(int(handler.observation_feature))
            if saved is not None:
                handler.rng.bit_generator.state = saved

    self._traffic_light.phase_counter = state.get("traffic_light_phase_counter", 0)

    self.cars = []
    if state["cars"] is not None and len(state["cars"]) > 0:
        self.cars = [Car.model_validate(car_data) for car_data in state["cars"]]
        self._next_car_id = self.cars[-1].id + 1

    self._update_car_positions_array()
    self._occupied = Counter(car.position for car in self.cars)

    reason_value = state.get("termination_reason")
    self.termination_reason = (
        TerminationReason(reason_value) if reason_value is not None else None
    )
    self.subgoals_reached = state.get("subgoals_reached", 0)
    self.visited_set = {
        tuple(cell) for cell in state.get("visited_set", [tuple(self.position)])
    }

    if "active_subgoals" in state:
        target_active = {tuple(cell) for cell in state["active_subgoals"]}
        for x in range(self.map.width):
            for y in range(self.map.height):
                if not (
                    self.map.feature_at(x, y, MF.SUBGOAL)
                    or self.map.feature_at(x, y, MF.USED_SUBGOAL)
                ):
                    continue
                if (x, y) in target_active:
                    self.map.add_feature_at(x, y, MF.SUBGOAL)
                    self.map.remove_feature_at(x, y, MF.USED_SUBGOAL)
                else:
                    self.map.remove_feature_at(x, y, MF.SUBGOAL)
                    self.map.add_feature_at(x, y, MF.USED_SUBGOAL)
        self.map.active_subgoals = np.array(state["active_subgoals"], dtype=int).reshape(-1, 2)

    self.positions_path = [tuple(self.position)]
    self.tile_path = [tuple(self.position)]
    self.noise_path = []

    return self.get_observation(), self.get_info()

generate_frame

generate_frame(
    hide_positions: bool = False,
    show_observation_window: bool = True,
    only_observation: bool = False,
) -> Image

Renders the current state as a PIL image.

Parameters:

Name	Type	Description	Default
hide_positions	bool	Hide the path taken in the current episode.	False
show_observation_window	bool	Dim everything outside the observation window.	True
only_observation	bool	Render the agent's observation: crop to the observation window and draw only the observed features.	False

Source code in pgtg/environment.py

def generate_frame(
    self,
    hide_positions: bool = False,
    show_observation_window: bool = True,
    only_observation: bool = False,
) -> Image:
    """Renders the current state as a PIL image.

    Args:
        hide_positions: Hide the path taken in the current episode.
        show_observation_window: Dim everything outside the observation window.
        only_observation: Render the agent's observation: crop to the observation
            window and draw only the observed features.
    """
    if only_observation:
        pic = graphic.create_observation_window(
            self, show_path=(not hide_positions), only_observed_features=True
        )
    else:
        pic = graphic.create_map(
            self,
            show_path=(not hide_positions),
            show_observation_window=show_observation_window,
        )

    if pic is None:
        from PIL import Image as PILImage

        pic = PILImage.new("RGBA", (400, 400), (255, 255, 255, 255))

    if pic.mode != "RGBA":
        pic = pic.convert("RGBA")

    return pic

add_traffic_rule

add_traffic_rule(rule_dict: dict[str, Any]) -> None

Adds a new traffic rule.

Source code in pgtg/environment.py

def add_traffic_rule(self, rule_dict: dict[str, Any]) -> None:
    """Adds a new traffic rule."""
    self.rule_engine.add_rule_from_dict(rule_dict)

remove_traffic_rule

remove_traffic_rule(rule_name: str) -> bool

Removes a traffic rule by name. Returns whether the rule existed.

Source code in pgtg/environment.py

def remove_traffic_rule(self, rule_name: str) -> bool:
    """Removes a traffic rule by name. Returns whether the rule existed."""
    return self.rule_engine.remove_rule(rule_name)

Configuration contexts

pgtg.contexts.ObservationContext

Bases: BaseModel

What the agent observes, and how large the observed window is.

Pass it to PGTGEnv as an instance, a dict of its fields, or a bare iterable of feature names (which is interpreted as obs_features).

Source code in pgtg/contexts.py

class ObservationContext(BaseModel):
    """What the agent observes, and how large the observed window is.

    Pass it to `PGTGEnv` as an instance, a `dict` of its fields, or a bare iterable of
    feature names (which is interpreted as `obs_features`).
    """

    # Names are resolved to bits by `convert_str_keys_to_enum` below. No `str` survives validation
    # Everything downstream (`obs_mask`, `is_observed`) can assume ints.
    obs_features: set[MF | int] = Field(
        default_factory=lambda: {
            MF.WALL,
            FeatureRegistry.category("GOAL"),
            MF.TRAFFIC,
            MF.TRAFFIC_LIGHT,
        },
    )
    """Which map features become observation planes.

    Pass string names (e.g. `"ice"`) as a list and they are resolved to the enum, or pass
    `MapFeature` members directly. `TRAFFIC_LIGHT` expands into three color planes. Only
    the structural features need listing: obstacle planes are added per environment based
    on whether the obstacle can actually appear there.
    """

    window_size: tuple[int, int] | None = None
    """Size of the observed window; a single `int` is treated as a square.

    `None` observes the currently occupied tile only.
    Anything else activates sliding window mode.
    """

    subgoal_direction: bool = False
    """Add a `Discrete(9)` compass pointing toward the next subgoal."""

    global_pos: bool = False
    """Add the agent's absolute `(x, y)` position to the observation."""

    local_pos: bool = True
    """Include the agent's position within the tile it occupies."""

    velocity: bool = True
    """Include the agent's velocity."""

    # Per-instance tile dims, injected by the env via the validation context ("tile_dims").
    # The default only applies to a context built without an env; it is the packaged tile
    # size, read once here rather than per instance.
    _tile_dims: tuple[int, int] = PrivateAttr(default=infer_tile_dims())
    sliding: bool = False

    @property
    def obs_mask(self) -> int:
        mask = 0
        for feature in self.obs_features:
            mask |= int(feature)
        return mask

    @field_validator("obs_features", mode="before")
    @classmethod
    def convert_str_keys_to_enum(cls, data: Any) -> Any:
        # Resolves feature and category *names* to their bit(mask).
        if isinstance(data, (str, bytes)) or not isinstance(data, Iterable):
            return data
        features = set()
        for key in data:
            if not isinstance(key, str):
                features.add(key)
                continue
            try:
                features.add(MF.from_str(key))
            except KeyError:
                raise ValueError(
                    f"unknown observation feature {key!r}. Known names: "
                    f"{', '.join(MF.known_names())}."
                ) from None
        return features

    @model_validator(mode="before")
    @classmethod
    def expand_shorthands(cls, data):
        # A bare iterable of feature names is shorthand for the obs_features field.
        # Non-iterables (an ObservationContext, None) pass through for pydantic to handle.
        if isinstance(data, Iterable) and not isinstance(data, (dict, str, bytes)):
            data = {"obs_features": data}

        # A single int window size means a square window.
        if isinstance(data, dict) and isinstance(data.get("window_size"), int):
            return {**data, "window_size": (data["window_size"], data["window_size"])}
        return data

    @model_validator(mode="after")
    def after_validator(self, info: ValidationInfo) -> ObservationContext:
        # Auto-observe: resolve each obstacle's tri-state `observe`.
        #   True  -> always observed;
        #   False -> never observed;
        #   None  -> observed iff the obstacle can actually appear in this env, i.e. its
        #            feature bit is in `generatable_here` (generated on a random map or
        #            present on a loaded map; injected by the env via the validation
        #            context, empty when the model is built standalone).
        ctx = info.context or {}
        generatable_here = {int(f) for f in ctx.get("generatable_here", ())}
        for obstacle in ctx.get("obstacles", ()):
            observe = getattr(obstacle, "observe", None)
            feature = getattr(obstacle, "observation_feature", None)
            if observe is False:
                continue
            if observe is None and (feature is None or int(feature) not in generatable_here):
                continue
            if feature is None:
                raise ValueError(
                    f"Obstacle {getattr(obstacle, 'name', obstacle)!r} has observe=True but no "
                    "observation_feature. Pass an observation_feature, or define it with the "
                    "@pgtg_obstacle decorator (which auto-allocates a unique bit)."
                )
            self.obs_features.add(feature)
        ctx_tile_dims = (info.context or {}).get("tile_dims")
        if ctx_tile_dims is not None:
            self._tile_dims = tuple(ctx_tile_dims)

        if self.window_size is not None:
            self.sliding = True
        else:
            self.window_size = self._tile_dims

        leaking = [
            f for f in self.obs_features if int(f) & FeatureRegistry.category("UNOBSERVABLE")
        ]
        if leaking:
            raise ValueError(
                "obs_features contains unobservable feature(s) "
                f"{FeatureRegistry.describe_many(leaking)}; "
                "these expose privileged map structure (e.g. car lanes, spawners, used subgoals, "
                "start and exit markers) and must not be observed."
            )

        return self

    @property
    def half_window_x(self) -> int:
        assert self.window_size is not None
        return self.window_size[0] // 2

    @property
    def half_window_y(self) -> int:
        assert self.window_size is not None
        return self.window_size[1] // 2

    def is_observed(self, feature: MF | int) -> bool:
        """Whether the given feature(s) are observed by this context."""
        return bool(int(feature) & self.obs_mask)

    def gen_obs_space(self, map_width: int = -1, map_height: int = -1) -> spaces.Dict:
        """Creates the Gymnasium observation space for this context.

        Args:
            map_width: The map width in _cells_ (i.e., `map.tiles_hor * tile_width`).
                only required if `global_pos=True`.
            map_height: The map height in _cells_ (i.e., `map.tiles_ver * tile_height`).
                only required if `global_pos=True`.
        """
        obs_space: dict[str, gym.Space] = {}

        assert self.window_size is not None
        # An even sliding window has no centre cell, silently mis-centering every observation.
        if self.sliding and not all(dim % 2 == 1 for dim in self.window_size):
            raise ValueError(
                f"sliding window_size dimensions must both be odd to have a centre "
                f"cell, got {self.window_size}."
            )

        if self.local_pos:
            obs_space["position"] = spaces.MultiDiscrete(list(self._tile_dims), dtype=np.int32)

        if self.velocity:
            obs_space["velocity"] = spaces.Box(
                low=-99,
                high=99,
                shape=(2,),
                dtype=np.int32,
            )

        if self.subgoal_direction:
            obs_space["subgoal_direction"] = spaces.Discrete(9)

        if self.global_pos:
            obs_space["global_position"] = spaces.Box(
                low=np.array([0, 0], dtype=np.int32),
                high=np.array([map_width - 1, map_height - 1], dtype=np.int32),
                shape=(2,),
                dtype=np.int32,
            )

        if self.obs_mask > 0:
            # One traffic-light bit becomes one channel per signal colour.
            channels: set[MF | int | SC] = set(self.obs_features)
            if MF.TRAFFIC_LIGHT in channels:
                channels.remove(MF.TRAFFIC_LIGHT)
                channels.update(SC)

            def _channel_order(feature: Any) -> tuple[int, int, str]:
                if isinstance(feature, SC):
                    return (1, 0, str(feature.value))
                return (0, int(feature), "")

            obs_space["map"] = spaces.Dict(
                {
                    feature: spaces.MultiBinary(self.window_size)
                    for feature in sorted(channels, key=_channel_order)
                },
            )

        return spaces.Dict(obs_space)

    def gen_observation(self, env: PGTGEnv, oh_maps: dict[Any, np.ndarray]) -> dict[str, Any]:
        """Generates the observation dict for the given environment and
        pre-initialized one-hot maps."""
        obs: dict[str, Any] = {}

        if self.local_pos:
            obs["position"] = env.position_in_tile.astype(np.int32)

        if self.velocity:
            obs["velocity"] = env.velocity.astype(np.int32)

        if self.subgoal_direction:
            obs["subgoal_direction"], _, _ = env.get_subgoal_compass_direction()

        if self.global_pos:
            obs["global_position"] = env.global_position.astype(np.int32)

        if self.obs_mask > 0:
            obs["map"] = oh_maps

        return obs

obs_features class-attribute instance-attribute

obs_features: set[MapFeature | int] = Field(
    default_factory=lambda: {
        MF.WALL,
        FeatureRegistry.category("GOAL"),
        MF.TRAFFIC,
        MF.TRAFFIC_LIGHT,
    }
)

Which map features become observation planes.

Pass string names (e.g. "ice") as a list and they are resolved to the enum, or pass MapFeature members directly. TRAFFIC_LIGHT expands into three color planes. Only the structural features need listing: obstacle planes are added per environment based on whether the obstacle can actually appear there.

window_size class-attribute instance-attribute

window_size: tuple[int, int] | None = None

Size of the observed window; a single int is treated as a square.

None observes the currently occupied tile only. Anything else activates sliding window mode.

subgoal_direction class-attribute instance-attribute

subgoal_direction: bool = False

Add a Discrete(9) compass pointing toward the next subgoal.

global_pos class-attribute instance-attribute

global_pos: bool = False

Add the agent's absolute (x, y) position to the observation.

local_pos class-attribute instance-attribute

local_pos: bool = True

Include the agent's position within the tile it occupies.

velocity class-attribute instance-attribute

velocity: bool = True

Include the agent's velocity.

is_observed

is_observed(feature: MapFeature | int) -> bool

Whether the given feature(s) are observed by this context.

Source code in pgtg/contexts.py

def is_observed(self, feature: MF | int) -> bool:
    """Whether the given feature(s) are observed by this context."""
    return bool(int(feature) & self.obs_mask)

gen_obs_space

gen_obs_space(
    map_width: int = -1, map_height: int = -1
) -> spaces.Dict

Creates the Gymnasium observation space for this context.

Parameters:

Name	Type	Description	Default
map_width	int	The map width in cells (i.e., map.tiles_hor * tile_width). only required if global_pos=True.	-1
map_height	int	The map height in cells (i.e., map.tiles_ver * tile_height). only required if global_pos=True.	-1

Source code in pgtg/contexts.py

def gen_obs_space(self, map_width: int = -1, map_height: int = -1) -> spaces.Dict:
    """Creates the Gymnasium observation space for this context.

    Args:
        map_width: The map width in _cells_ (i.e., `map.tiles_hor * tile_width`).
            only required if `global_pos=True`.
        map_height: The map height in _cells_ (i.e., `map.tiles_ver * tile_height`).
            only required if `global_pos=True`.
    """
    obs_space: dict[str, gym.Space] = {}

    assert self.window_size is not None
    # An even sliding window has no centre cell, silently mis-centering every observation.
    if self.sliding and not all(dim % 2 == 1 for dim in self.window_size):
        raise ValueError(
            f"sliding window_size dimensions must both be odd to have a centre "
            f"cell, got {self.window_size}."
        )

    if self.local_pos:
        obs_space["position"] = spaces.MultiDiscrete(list(self._tile_dims), dtype=np.int32)

    if self.velocity:
        obs_space["velocity"] = spaces.Box(
            low=-99,
            high=99,
            shape=(2,),
            dtype=np.int32,
        )

    if self.subgoal_direction:
        obs_space["subgoal_direction"] = spaces.Discrete(9)

    if self.global_pos:
        obs_space["global_position"] = spaces.Box(
            low=np.array([0, 0], dtype=np.int32),
            high=np.array([map_width - 1, map_height - 1], dtype=np.int32),
            shape=(2,),
            dtype=np.int32,
        )

    if self.obs_mask > 0:
        # One traffic-light bit becomes one channel per signal colour.
        channels: set[MF | int | SC] = set(self.obs_features)
        if MF.TRAFFIC_LIGHT in channels:
            channels.remove(MF.TRAFFIC_LIGHT)
            channels.update(SC)

        def _channel_order(feature: Any) -> tuple[int, int, str]:
            if isinstance(feature, SC):
                return (1, 0, str(feature.value))
            return (0, int(feature), "")

        obs_space["map"] = spaces.Dict(
            {
                feature: spaces.MultiBinary(self.window_size)
                for feature in sorted(channels, key=_channel_order)
            },
        )

    return spaces.Dict(obs_space)

gen_observation

gen_observation(
    env: PGTGEnv, oh_maps: dict[Any, ndarray]
) -> dict[str, Any]

Generates the observation dict for the given environment and pre-initialized one-hot maps.

Source code in pgtg/contexts.py

def gen_observation(self, env: PGTGEnv, oh_maps: dict[Any, np.ndarray]) -> dict[str, Any]:
    """Generates the observation dict for the given environment and
    pre-initialized one-hot maps."""
    obs: dict[str, Any] = {}

    if self.local_pos:
        obs["position"] = env.position_in_tile.astype(np.int32)

    if self.velocity:
        obs["velocity"] = env.velocity.astype(np.int32)

    if self.subgoal_direction:
        obs["subgoal_direction"], _, _ = env.get_subgoal_compass_direction()

    if self.global_pos:
        obs["global_position"] = env.global_position.astype(np.int32)

    if self.obs_mask > 0:
        obs["map"] = oh_maps

    return obs

pgtg.contexts.RewardContext

Bases: BaseModel

The reward and penalty structure.

Penalties are given as positive magnitudes and are subtracted from the reward, unless separate_reward_cost reports them separately.

Source code in pgtg/contexts.py

class RewardContext(BaseModel):
    """The reward and penalty structure.

    Penalties are given as positive magnitudes and are *subtracted* from the reward,
    unless `separate_reward_cost` reports them separately.
    """

    # Rewards
    subgoal_reward: float = 1
    """Reward for reaching a subgoal, or the final goal."""

    final_goal_bonus: float = 0
    """Extra reward added on top of `subgoal_reward` when the final goal is reached."""

    # Penalties. Given as positive magnitudes; they are *subtracted* from the reward
    crash_penalty: float = 1.0
    """Subtracted for hitting a wall, leaving the map, or colliding with traffic.
    Ends the episode."""

    traffic_light_penalty: float = 0.5
    """Subtracted for running a red light. Does not end the episode."""

    standing_still_penalty: float = 0
    """Subtracted each step the agent does not move."""

    already_visited_penalty: float = 0
    """Subtracted for entering a cell already visited this episode."""

    # Meta
    split_subgoal_reward: bool = False
    """Divide `subgoal_reward` by the number of subgoals.

    Keeps the total reward per track constant regardless of how long the track is.
    """

    separate_reward_cost: bool = False
    """Report penalties only as `info["cost"]` without subtracting them from the reward.

    For safe RL, where reward and cost are optimized separately.
    """

    # We can verify predefined penalties, but the ones defined by the user through
    # custom obstacles are not known. At least verify the known ones here.
    _PENALTY_FIELDS: ClassVar[tuple[str, ...]] = (
        "crash_penalty",
        "traffic_light_penalty",
        "standing_still_penalty",
        "already_visited_penalty",
    )

    @model_validator(mode="after")
    def check_finite_and_warn_on_negative_penalties(self) -> RewardContext:
        # Rejects NaN/infinite values and flags penalties that are secretly rewards
        for name in (*self._PENALTY_FIELDS, "subgoal_reward", "final_goal_bonus"):
            value = getattr(self, name)
            if not math.isfinite(value):
                raise ValueError(f"{name} must be a finite number, got {value!r}.")

        negative = [name for name in self._PENALTY_FIELDS if getattr(self, name) < 0]
        if negative:
            PGTGLogger.get_logger().warning(
                "Negative penalt%s %s: penalties are subtracted from the reward, so a "
                "negative one rewards the behaviour it is named after. Use a positive value "
                "to discourage it.",
                "y" if len(negative) == 1 else "ies",
                ", ".join(f"{name}={getattr(self, name)}" for name in negative),
                stacklevel=2,
            )
        return self

    def individual_subgoal_reward(self, num_subgoals: int) -> float:
        if self.split_subgoal_reward and num_subgoals > 0:
            return self.subgoal_reward / num_subgoals
        return self.subgoal_reward

subgoal_reward class-attribute instance-attribute

subgoal_reward: float = 1

Reward for reaching a subgoal, or the final goal.

final_goal_bonus class-attribute instance-attribute

final_goal_bonus: float = 0

Extra reward added on top of subgoal_reward when the final goal is reached.

crash_penalty class-attribute instance-attribute

crash_penalty: float = 1.0

Subtracted for hitting a wall, leaving the map, or colliding with traffic. Ends the episode.

traffic_light_penalty class-attribute instance-attribute

traffic_light_penalty: float = 0.5

Subtracted for running a red light. Does not end the episode.

standing_still_penalty class-attribute instance-attribute

standing_still_penalty: float = 0

Subtracted each step the agent does not move.

already_visited_penalty class-attribute instance-attribute

already_visited_penalty: float = 0

Subtracted for entering a cell already visited this episode.

split_subgoal_reward class-attribute instance-attribute

split_subgoal_reward: bool = False

Divide subgoal_reward by the number of subgoals.

Keeps the total reward per track constant regardless of how long the track is.

separate_reward_cost class-attribute instance-attribute

separate_reward_cost: bool = False

Report penalties only as info["cost"] without subtracting them from the reward.

For safe RL, where reward and cost are optimized separately.

pgtg.contexts.RandomMapContext

Bases: BaseModel

Procedural map generation parameters, used only when no map_path is given.

Source code in pgtg/contexts.py

class RandomMapContext(BaseModel):
    """Procedural map generation parameters, used only when no `map_path` is given."""

    width: int = 4
    """Map width in tiles."""

    height: int = 4
    """Map height in tiles."""

    connections_percentage: float = 0.5
    """Fraction of the possible tile connections to realize, in `[0, 1]`.
    `0.0` leaves only a minimal start-to-goal path, `1.0` connects every neighboring tile.
    """

    start_position: tuple[int, int] | tuple[int, int, str] | str = (
        0,
        -1,
        "west",
    )
    """Start tile on a border, with an optional entry direction, or `"random"`.

    A negative coordinate counts from the far edge, so `(0, -1, "west")` is the
    bottom-left tile, entered from the west.
    """

    goal_position: tuple[int, int] | tuple[int, int, str] | str = (
        -1,
        0,
        "east",
    )
    """Goal tile on a border, with an optional exit direction, or `"random"`."""

    min_dist_start_goal: int | None = None
    """Minimum path length between start and goal. Requires both positions to be `"random"`."""

    obstacle_probability: float = 0.0
    """Per-tile probability of receiving an obstacle, in `[0, 1]`."""

    # MF | int: a custom obstacle's feature is a plain int bit
    obs_feature_weights: dict[MF | int, float] = Field(
        default_factory=lambda: dict.fromkeys(random_obstacle_features(), 1.0),
    )
    """Relative weights deciding which obstacle type is placed, normalized internally.

    Defaults to each obstacle's registered `obstacle_weight` (`1` for the built-ins).
    String keys are accepted. A weight of `0` excludes that obstacle from generation, and
    with the default `observe=None` therefore also from the observation.
    """

    @field_validator("obs_feature_weights", mode="before")
    @classmethod
    def convert_str_keys_to_enum(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        return {MF.from_str(k) if isinstance(k, str) else k: v for k, v in data.items()}

    @model_validator(mode="after")
    def validate_obstacle_weights(self, info: ValidationInfo) -> RandomMapContext:
        ctx = info.context or {}

        random_obstacles = random_obstacle_features()

        instance_weights: dict[MF | int, float] = {
            obs.observation_feature: obs.obstacle_weight
            for obs in ctx.get("obstacles", ())
            if isinstance(obs, Generatable) and obs.observation_feature in random_obstacles
        }
        generatable_features = list(dict.fromkeys([*random_obstacles, *instance_weights]))
        generatable_set = set(generatable_features)

        user_set = "obs_feature_weights" in self.model_fields_set
        user_weights = self.obs_feature_weights if user_set else {}

        bad = [feat for feat in user_weights if feat not in generatable_set]
        if bad:
            raise ValueError(
                "Only random/generatable obstacles can have weights "
                f"(remove weights from: {FeatureRegistry.describe_many(bad)})"
            )

        if user_set:
            self.obs_feature_weights = {
                feat: user_weights.get(feat, 0.0) for feat in generatable_features
            }
        else:
            self.obs_feature_weights = {
                feat: instance_weights.get(
                    feat,
                    FeatureRegistry.weight_of(feat, 1.0 if feat in random_obstacles else 0.0),
                )
                for feat in generatable_features
            }
        if self.obstacle_probability <= 0:
            return self

        for feature, weight in self.obs_feature_weights.items():
            if weight < 0:
                raise ValueError(
                    f"Weight for {FeatureRegistry.describe(feature)} must be non-negative."
                )

        norm_factor = sum(self.obs_feature_weights.values())

        if norm_factor == 0:
            PGTGLogger.get_logger().warning(
                "All generatable obstacle weights are 0, so no obstacles will be placed "
                f"despite obstacle_probability={self.obstacle_probability}. Set a positive "
                "weight for at least one obstacle, or obstacle_probability=0, to silence this.",
                stacklevel=2,
            )
            self.obstacle_probability = 0.0
            return self

        for feature, weight in self.obs_feature_weights.items():
            if weight > 0:
                self.obs_feature_weights[feature] = weight / norm_factor

        return self

width class-attribute instance-attribute

width: int = 4

Map width in tiles.

height class-attribute instance-attribute

height: int = 4

Map height in tiles.

connections_percentage class-attribute instance-attribute

connections_percentage: float = 0.5

Fraction of the possible tile connections to realize, in [0, 1]. 0.0 leaves only a minimal start-to-goal path, 1.0 connects every neighboring tile.

start_position class-attribute instance-attribute

start_position: (
    tuple[int, int] | tuple[int, int, str] | str
) = (0, -1, "west")

Start tile on a border, with an optional entry direction, or "random".

A negative coordinate counts from the far edge, so (0, -1, "west") is the bottom-left tile, entered from the west.

goal_position class-attribute instance-attribute

goal_position: (
    tuple[int, int] | tuple[int, int, str] | str
) = (-1, 0, "east")

Goal tile on a border, with an optional exit direction, or "random".

min_dist_start_goal class-attribute instance-attribute

min_dist_start_goal: int | None = None

Minimum path length between start and goal. Requires both positions to be "random".

obstacle_probability class-attribute instance-attribute

obstacle_probability: float = 0.0

Per-tile probability of receiving an obstacle, in [0, 1].

obs_feature_weights class-attribute instance-attribute

obs_feature_weights: dict[MapFeature | int, float] = Field(
    default_factory=lambda: dict.fromkeys(
        random_obstacle_features(), 1.0
    )
)

Relative weights deciding which obstacle type is placed, normalized internally.

Defaults to each obstacle's registered obstacle_weight (1 for the built-ins). String keys are accepted. A weight of 0 excludes that obstacle from generation, and with the default observe=None therefore also from the observation.

Obstacles

pgtg.obstacles.pgtg_obstacle

pgtg_obstacle(
    name: str,
    observation_feature: MapFeature | int | None = None,
    obstacle_weight: float = 1.0,
    observe: bool | None = None,
    color: tuple[int, int, int] | None = None,
    category: str | None = "RANDOM_OBSTACLE",
    **default_kwargs: Any,
) -> Callable[
    [Callable[[PGTGEnv, StepContext], bool]],
    ObstacleFactory,
]

Source code in pgtg/obstacles/registry.py

def pgtg_obstacle(
    name: str,
    observation_feature: MF | int | None = None,
    obstacle_weight: float = 1.0,
    observe: bool | None = None,
    color: tuple[int, int, int] | None = None,
    category: str | None = "RANDOM_OBSTACLE",  # Currently, only RANDOM_OBSTACLE is supported/tested
    **default_kwargs: Any,
) -> Callable[[Callable[[PGTGEnv, StepContext], bool]], ObstacleFactory]:
    # Auto-assign a unique feature bit
    if observation_feature is None:
        existing = FeatureRegistry.feature_for_name(name) if name else None
        observation_feature = existing if existing is not None else FeatureRegistry.allocate(name)
    else:
        FeatureRegistry.register_name(observation_feature, name)

    if category != "RANDOM_OBSTACLE":
        raise ValueError(f"category={category!r} is not supported yet; only 'RANDOM_OBSTACLE' is.")
    if category is not None:  # RANDOM_OBSTACLE -> auto-generate & observe
        FeatureRegistry.add_to_category(category, observation_feature)

    def decorator(behavior: Callable[[PGTGEnv, StepContext], bool]) -> ObstacleFactory:
        func = cast(_ObstacleBehavior, behavior)  # For type hints

        register_feature_color(observation_feature, color, name)

        func.observation_feature = observation_feature
        func.name = name

        FeatureRegistry.set_weight(observation_feature, obstacle_weight)  # global generation weight

        factory = cast(
            ObstacleFactory,
            _ObstacleFactoryWraps(
                func,
                name=name,
                observation_feature=observation_feature,
                obstacle_weight=obstacle_weight,
                observe=observe,
                default_kwargs=default_kwargs,
            ),
        )
        # Use update_wrapper instead of @wraps, to have thread-safe, easily accessible rngs
        update_wrapper(factory, behavior)

        ObstacleRegistry.register(name, factory)
        return factory

    return decorator

pgtg.obstacles.Obstacle

Bases: SteppableFeature

Source code in pgtg/obstacles/base.py

class Obstacle(SteppableFeature):
    rng: np.random.Generator
    obstacle_weight: float

    def __init__(
        self,
        observation_feature: MF | int,
        behavior_fn: Callable[[PGTGEnv, StepContext], bool],
        act_probability: float,
        act_penalty: float = 0.0,
        obstacle_weight: float = 0.0,
        *,
        observe: bool | None = None,
        name: str = "",
    ):
        self.observation_feature = observation_feature
        self.observation_gated = True
        self.behavior_fn = behavior_fn
        self.act_probability = act_probability
        self.act_penalty = act_penalty
        self.obstacle_weight = obstacle_weight
        self.observe = observe
        self.rng = np.random.default_rng()
        self.name = name

    def act(self, env: PGTGEnv, local_ctx: StepContext) -> bool:
        """Returns `True` iff processing should stop for this `velocity_substep`."""
        if not env.map.feature_at(*tuple(local_ctx.current_position), self.observation_feature):
            return False

        if self.rng.random() < self.act_probability:
            env.noise_path.append((self.name, tuple(env.position)))
            # behavior body reads its rng via the decorated module-level name
            # publishing self is what makes that read resolve to *this* obstacle's stream
            token = _current_obstacle.set(self)
            try:
                return self.behavior_fn(env, local_ctx)
            finally:
                _current_obstacle.reset(token)
        return False

act

act(env: PGTGEnv, local_ctx: StepContext) -> bool

Returns True iff processing should stop for this velocity_substep.

Source code in pgtg/obstacles/base.py

def act(self, env: PGTGEnv, local_ctx: StepContext) -> bool:
    """Returns `True` iff processing should stop for this `velocity_substep`."""
    if not env.map.feature_at(*tuple(local_ctx.current_position), self.observation_feature):
        return False

    if self.rng.random() < self.act_probability:
        env.noise_path.append((self.name, tuple(env.position)))
        # behavior body reads its rng via the decorated module-level name
        # publishing self is what makes that read resolve to *this* obstacle's stream
        token = _current_obstacle.set(self)
        try:
            return self.behavior_fn(env, local_ctx)
        finally:
            _current_obstacle.reset(token)
    return False

pgtg.obstacles.set_obstacle_weight

set_obstacle_weight(
    weights: dict[str, float] | str,
    weight: float | None = None,
) -> None

Set the random-map generation weight of already-registered obstacle(s).

Call with either a mapping {name: weight} or a single name, weight pair. The change persists on the process-global registry until set again, so any PGTGEnv built afterwards picks it up automatically: a positive weight makes the obstacle generated and observed; 0 removes it from generation (and from auto-observation). Names are matched loosely ("broken road" == "broken_road").

Example::

pgtg.set_obstacle_weight({"ice": 0.2, "broken road": 0.0})

Source code in pgtg/obstacles/registry.py

def set_obstacle_weight(weights: dict[str, float] | str, weight: float | None = None) -> None:
    """Set the random-map generation weight of already-registered obstacle(s).

    Call with either a mapping `{name: weight}` or a single `name, weight` pair.
    The change persists on the process-global registry until set again, so **any PGTGEnv built
    afterwards picks it up automatically**: a positive weight makes the obstacle generated
    and observed; `0` removes it from generation (and from auto-observation).
    Names are matched loosely (`"broken road"` == `"broken_road"`).

    Example::

        pgtg.set_obstacle_weight({"ice": 0.2, "broken road": 0.0})
    """
    mapping = {weights: weight} if isinstance(weights, str) else dict(weights)
    if isinstance(weights, str) and weight is None:
        raise ValueError("set_obstacle_weight(name, weight): weight is required")
    for name, value in mapping.items():
        if value is None:
            raise ValueError(f"weight for {name!r} must be a number, not None")
        factory = ObstacleRegistry.factory_for(name)
        FeatureRegistry.set_weight(factory.observation_feature, value)

pgtg.obstacles.get_obstacle_weight

get_obstacle_weight(name: str) -> float

Current random-map generation weight of a registered obstacle (see set_obstacle_weight).

Source code in pgtg/obstacles/registry.py

def get_obstacle_weight(name: str) -> float:
    """Current random-map generation weight of a registered obstacle (see set_obstacle_weight)."""
    factory = ObstacleRegistry.factory_for(name)
    return FeatureRegistry.weight_of(factory.observation_feature)

pgtg.contexts.StepContext dataclass

The mutable state of the step currently being processed.

Handed to every obstacle behavior as local_ctx, which may modify it in place to change how the rest of the step plays out. See Custom Obstacles.

Source code in pgtg/contexts.py

@dataclass
class StepContext:
    """The mutable state of the step currently being processed.

    Handed to every obstacle behavior as `local_ctx`, which may modify it in place to
    change how the rest of the step plays out. See
    [Custom Obstacles](https://neuro-mechanistic-modeling.github.io/pgtg/extending/custom-obstacles/).
    """

    current_position: np.ndarray
    """The car's position for this substep."""

    acceleration: np.ndarray
    """The acceleration applied this step, i.e. the agent's action.
    It has already been added to the car's velocity.
    Changing this has no effect on the car's movement.
    """

    velocity_substep: np.ndarray | None = None
    """The unit velocity vector for this substep, or `None` for the final marker substep.
    Reassign it to modify the one cell substep movement of this full step.
    """

    sum_reward: float = 0.0
    """Reward accumulated so far this step. Add to it to reward a behavior."""

    sum_penalty: float = 0.0
    """Penalty accumulated so far this step. Add to it to penalize a behavior."""

    is_success: bool = False
    """Whether the final goal was reached this step. Surfaces as `info["is_success"]`."""

    remaining_substeps: list[np.ndarray | None] = field(default_factory=list)
    """The substeps still to be processed.

    Set it to `[None]` and return `False` to cancel the rest of the velocity while still
    validating the move, which is how sand stops the car safely.

    Changing this changes the future substep walk of this step's processing.
    """

current_position instance-attribute

current_position: ndarray

The car's position for this substep.

acceleration instance-attribute

acceleration: ndarray

The acceleration applied this step, i.e. the agent's action. It has already been added to the car's velocity. Changing this has no effect on the car's movement.

velocity_substep class-attribute instance-attribute

velocity_substep: ndarray | None = None

The unit velocity vector for this substep, or None for the final marker substep. Reassign it to modify the one cell substep movement of this full step.

sum_reward class-attribute instance-attribute

sum_reward: float = 0.0

Reward accumulated so far this step. Add to it to reward a behavior.

sum_penalty class-attribute instance-attribute

sum_penalty: float = 0.0

Penalty accumulated so far this step. Add to it to penalize a behavior.

is_success class-attribute instance-attribute

is_success: bool = False

Whether the final goal was reached this step. Surfaces as info["is_success"].

remaining_substeps class-attribute instance-attribute

remaining_substeps: list[ndarray | None] = field(
    default_factory=list
)

The substeps still to be processed.

Set it to [None] and return False to cancel the rest of the velocity while still validating the move, which is how sand stops the car safely.

Changing this changes the future substep walk of this step's processing.

Driver profiles

pgtg.traffic.drivers.DriverProfile

Bases: Enum

The names of the five profiles PGTG ships with.

Custom profiles defined through a DriverProfileConfig are referred to by plain strings and are not members of this enum.

Source code in pgtg/traffic/drivers.py

class DriverProfile(Enum):
    """The names of the five profiles PGTG ships with.

    Custom profiles defined through a `DriverProfileConfig` are referred to by plain
    strings and are not members of this enum.
    """

    CONSERVATIVE = "conservative"
    """Slightly slow, keeps its distance, and almost always stops at a light. Weight `0.25`."""

    NORMAL = "normal"
    """Full speed and stops at most lights. The most common profile, at weight `0.35`."""

    AGGRESSIVE = "aggressive"
    """Full speed, tailgates, overtakes readily, and runs yellow lights. Weight `0.2`."""

    ELDERLY = "elderly"
    """The slowest and most hesitant: large gaps, frequent delays, never runs a light.

    Weight `0.15`.
    """

    RECKLESS = "reckless"
    """Full speed, overtakes immediately, and runs red lights three times in ten.

    The rarest profile, at weight `0.05`.
    """

CONSERVATIVE class-attribute instance-attribute

CONSERVATIVE = 'conservative'

Slightly slow, keeps its distance, and almost always stops at a light. Weight 0.25.

NORMAL class-attribute instance-attribute

NORMAL = 'normal'

Full speed and stops at most lights. The most common profile, at weight 0.35.

AGGRESSIVE class-attribute instance-attribute

AGGRESSIVE = 'aggressive'

Full speed, tailgates, overtakes readily, and runs yellow lights. Weight 0.2.

ELDERLY class-attribute instance-attribute

ELDERLY = 'elderly'

The slowest and most hesitant: large gaps, frequent delays, never runs a light.

Weight 0.15.

RECKLESS class-attribute instance-attribute

RECKLESS = 'reckless'

Full speed, overtakes immediately, and runs red lights three times in ten.

The rarest profile, at weight 0.05.

pgtg.traffic.drivers.DriverBehavior

Bases: BaseModel

How one driver profile acts. Every traffic car is assigned one.

All probabilities are redrawn every step, so they describe a per-step chance rather than a decision the car commits to.

Source code in pgtg/traffic/drivers.py

class DriverBehavior(BaseModel):
    """How one driver profile acts. Every traffic car is assigned one.

    All probabilities are redrawn every step, so they describe a per-step chance rather
    than a decision the car commits to.
    """

    yellow_light_stop_probability: float
    """Chance of stopping at a yellow light each step. Higher is more cautious."""

    red_light_stop_probability: float
    """Chance of stopping at a red light each step. Higher is more cautious."""

    min_following_distance: int
    """Gap kept from the car ahead, in cells.

    `0` never queues and passes through the car ahead right away,
    anything else queues until `patience_level` runs out before overtaking.
    """

    patience_level: float
    """How long the car queues behind a slower one before overtaking, in `[0, 1]`.

    It waits `patience_level * 10` steps, then overtakes with probability
    `1 - patience_level` per step. `0` overtakes at the first opportunity, `1` never does.
    """

    speed_multiplier: float
    """Chance of attempting to move each step, so effectively the car's speed.

    `1.0` attempts every step and `0.0` never moves. A car covers at most one cell per
    step, so values above `1.0` have no additional effect.
    """

    reaction_delay_probability: float
    """Chance each step of the driver not moving for the next 1 to 3 steps."""

    @field_validator("speed_multiplier")
    @classmethod
    def warn_if_speed_multiplier_has_no_effect(cls, v: float) -> float:
        # A car moves at most one cell per step, so only values in (0, 1) slow it down;
        if v > 1.0:
            PGTGLogger.get_logger().warning(
                f"speed_multiplier={v} is > 1.0, which has no additional effect: a car "
                "already moves at most once per step, and 1.0 already means 'attempt to "
                "move every step'. Values above 1.0 behave identically to 1.0; only values "
                "in (0, 1) slow a car down.",
                stacklevel=2,
            )
        return v

yellow_light_stop_probability instance-attribute

yellow_light_stop_probability: float

Chance of stopping at a yellow light each step. Higher is more cautious.

red_light_stop_probability instance-attribute

red_light_stop_probability: float

Chance of stopping at a red light each step. Higher is more cautious.

min_following_distance instance-attribute

min_following_distance: int

Gap kept from the car ahead, in cells.

0 never queues and passes through the car ahead right away, anything else queues until patience_level runs out before overtaking.

patience_level instance-attribute

patience_level: float

How long the car queues behind a slower one before overtaking, in [0, 1].

It waits patience_level * 10 steps, then overtakes with probability 1 - patience_level per step. 0 overtakes at the first opportunity, 1 never does.

speed_multiplier instance-attribute

speed_multiplier: float

Chance of attempting to move each step, so effectively the car's speed.

1.0 attempts every step and 0.0 never moves. A car covers at most one cell per step, so values above 1.0 have no additional effect.

reaction_delay_probability instance-attribute

reaction_delay_probability: float

Chance each step of the driver not moving for the next 1 to 3 steps.

pgtg.traffic.drivers.DriverProfileConfig

Bases: BaseModel

Configuration for driver behavior profiles, loadable from dict, JSON/YAML path, or instance.

When loaded from a path, behaviors are merged with the package defaults (custom takes precedence), and weights fall back to defaults if the file has none.

Source code in pgtg/traffic/drivers.py

class DriverProfileConfig(BaseModel):
    """Configuration for driver behavior profiles, loadable from dict, JSON/YAML path, or instance.

    When loaded from a path, behaviors are merged with the package defaults
    (custom takes precedence), and weights fall back to defaults if the file
    has none.
    """

    behaviors: dict[str, DriverBehavior]
    """The available profiles, keyed by name. `DriverProfile` members are accepted as keys."""

    weights: dict[str, float] = Field(default_factory=dict)
    """How often each profile is sampled, keyed by name and normalized internally.

    Every name must also appear in `behaviors`. Empty spreads the weight evenly over all
    of them.
    """

    @model_validator(mode="before")
    @classmethod
    def coerce_path_input(cls, data: Any) -> Any:
        if isinstance(data, (str, Path)):
            custom = load_config_file(data)
            defaults = load_config_file(DRIVER_PROFILES_FILE)
            custom_weights = custom.get("weights") or {}
            data = {
                "behaviors": {
                    **defaults.get("behaviors", {}),
                    **custom.get("behaviors", {}),
                },
                "weights": custom_weights or defaults.get("weights", {}),
            }
        return data

    @model_validator(mode="before")
    @classmethod
    def normalize_enum_keys(cls, data: Any) -> Any:
        # Build new dict so the caller's input is never mutated.
        if not isinstance(data, dict):
            return data
        updates = {}
        for field_name in ("behaviors", "weights"):
            field_val = data.get(field_name)
            if isinstance(field_val, dict):
                updates[field_name] = {
                    (k.value if isinstance(k, DriverProfile) else str(k)): v
                    for k, v in field_val.items()
                }
        return {**data, **updates} if updates else data

    @model_validator(mode="after")
    def validate_weights_match_behaviors(self) -> DriverProfileConfig:
        for name in self.weights:
            if name not in self.behaviors:
                raise ValueError(
                    f"Weight defined for unknown profile '{name}'. "
                    f"Known profiles: {list(self.behaviors.keys())}"
                )
        return self

    @property
    def normalized_weights(self) -> dict[str, float]:
        """The sampling weights as a probability distribution over the profile names."""
        if not self.weights:
            n = len(self.behaviors)
            return dict.fromkeys(self.behaviors, 1.0 / n)
        negative = {name: w for name, w in self.weights.items() if w < 0}
        if negative:
            raise ValueError(f"Driver profile weights must be non-negative, got {negative}.")
        total = sum(self.weights.values())
        if total <= 0:
            raise ValueError(
                "Driver profile weights cannot all be zero -- there would be no profile to "
                "sample. Give at least one profile a positive weight, or leave weights empty "
                "to spread them evenly."
            )
        return {k: v / total for k, v in self.weights.items()}

    @classmethod
    def defaults(cls) -> DriverProfileConfig:
        """The five profiles PGTG ships with, and their default weights."""
        return cls.model_validate(load_config_file(DRIVER_PROFILES_FILE))

behaviors instance-attribute

behaviors: dict[str, DriverBehavior]

The available profiles, keyed by name. DriverProfile members are accepted as keys.

weights class-attribute instance-attribute

weights: dict[str, float] = Field(default_factory=dict)

How often each profile is sampled, keyed by name and normalized internally.

Every name must also appear in behaviors. Empty spreads the weight evenly over all of them.

normalized_weights property

normalized_weights: dict[str, float]

The sampling weights as a probability distribution over the profile names.

defaults classmethod

defaults() -> DriverProfileConfig

The five profiles PGTG ships with, and their default weights.

Source code in pgtg/traffic/drivers.py

@classmethod
def defaults(cls) -> DriverProfileConfig:
    """The five profiles PGTG ships with, and their default weights."""
    return cls.model_validate(load_config_file(DRIVER_PROFILES_FILE))

Traffic rules

pgtg.traffic.rules.TrafficRule

Bases: BaseModel

A traffic rule whose action runs when the rule triggers.

All of the trigger conditions below have to hold at once for the action to run. They are re-checked at every substep of the agent's move.

Source code in pgtg/traffic/rules.py

class TrafficRule(BaseModel):
    """A traffic rule whose action runs when the rule triggers.

    All of the trigger conditions below have to hold at once for the action to run. They
    are re-checked at every substep of the agent's move.
    """

    name: str
    """Unique identifier, used to add and remove the rule."""

    tile_type: tuple[int, int, int, int]
    """The tile exit pattern `(N, E, S, W)` the rule applies to.

    `(1, 1, 1, 1)` is a four-way crossing and `(1, 1, 1, 0)` a T-crossing. A four
    character string such as `"1111"` is accepted as well.
    """

    velocity_range: tuple[float, float]
    """Inclusive `[min, max]` range the agent's speed has to fall in."""

    min_traffic: int
    """How many cars have to be in the tile, regardless of where they are heading."""

    min_matching_traffic: int
    """How many of those cars have to be driving one of the directions in `maneuvers`."""

    maneuvers: dict[LD, set[LD]]
    """Which traffic directions matter, per agent heading.

    Keys and values are `LaneDirection` values such as `"west_to_east"`. The agent's
    heading is inferred from the direction to its next subgoal. In a file this is written
    as a list of `{"agent": ..., "traffic": [...]}` entries.
    """

    # Action returns truthy to halt the agent's remaining substep walk this step
    action: Callable[[Any, StepContext], bool]
    """What to run when the rule triggers, as `(env, ctx) -> bool`.

    Returning a truthy value halts the agent's remaining movement this step, which is what
    makes the built-in rules brake. May also be given as a string holding the source of a
    lambda, which keeps a rule picklable and loadable from a file; `numpy` is available as
    `np` in that scope.
    """

    @field_validator("action", mode="before")
    @classmethod
    def evaluate_rule_action(cls, data: Any) -> Callable[[Any, StepContext], Any]:
        if isinstance(data, str):
            # String form so rules survive pickling for multiprocessing and loading.
            # This evaluates the string as Python: a traffic-rules file is executable
            # content, exactly like the module that loads it.
            return eval(data)
        if not callable(data):
            raise ValueError(
                "A rule action must be a callable (env, ctx) -> Any, or a string with the "
                f"source of one; got {type(data).__name__}."
            )
        return data

    @field_validator("tile_type", mode="before")
    @classmethod
    def parse_tile_type(cls, data: Any) -> tuple[int, int, int, int]:
        if isinstance(data, str) and len(data) == 4 and all(c in "01" for c in data):
            return tuple(int(c) for c in data)
        return data

    @field_validator("maneuvers", mode="before")
    @classmethod
    def parse_maneuvers(cls, data: Any) -> dict[LD, set[LD]] | Any:
        if isinstance(data, dict):
            data = [data]

        if isinstance(data, list):
            return {LD(d["agent"]): {LD(t) for t in d["traffic"]} for d in data}

        return data

name instance-attribute

name: str

Unique identifier, used to add and remove the rule.

tile_type instance-attribute

tile_type: tuple[int, int, int, int]

The tile exit pattern (N, E, S, W) the rule applies to.

(1, 1, 1, 1) is a four-way crossing and (1, 1, 1, 0) a T-crossing. A four character string such as "1111" is accepted as well.

velocity_range instance-attribute

velocity_range: tuple[float, float]

Inclusive [min, max] range the agent's speed has to fall in.

min_traffic instance-attribute

min_traffic: int

How many cars have to be in the tile, regardless of where they are heading.

min_matching_traffic instance-attribute

min_matching_traffic: int

How many of those cars have to be driving one of the directions in maneuvers.

maneuvers instance-attribute

maneuvers: dict[LaneDirection, set[LaneDirection]]

Which traffic directions matter, per agent heading.

Keys and values are LaneDirection values such as "west_to_east". The agent's heading is inferred from the direction to its next subgoal. In a file this is written as a list of {"agent": ..., "traffic": [...]} entries.

action instance-attribute

action: Callable[[Any, StepContext], bool]

What to run when the rule triggers, as (env, ctx) -> bool.

Returning a truthy value halts the agent's remaining movement this step, which is what makes the built-in rules brake. May also be given as a string holding the source of a lambda, which keeps a rule picklable and loadable from a file; numpy is available as np in that scope.

pgtg.traffic.rules.TrafficRulesConfig

Bases: BaseModel

Configuration for traffic rules, loadable from dict, JSON/YAML path, or instance. Files use a top-level rules key whose value is a list of rule dicts.

Source code in pgtg/traffic/rules.py

class TrafficRulesConfig(BaseModel):
    """Configuration for traffic rules, loadable from dict, JSON/YAML path, or instance.
    Files use a top-level `rules` key whose value is a list of rule dicts.
    """

    rules: list[TrafficRule]
    """The rules to load, each with a name unique across the defaults and your own.
    Loading a rule does not activate it; the env's `rules_active` decides what rules are active.
    """

    @model_validator(mode="before")
    @classmethod
    def coerce_path_input(cls, data: Any) -> Any:
        if isinstance(data, (str, Path)):
            return load_config_file(data)
        return data

    @classmethod
    def defaults(cls) -> TrafficRulesConfig:
        """The two braking rules PGTG ships with, for four-way and T-crossings."""
        return cls.model_validate(TRAFFIC_RULES_FILE)

rules instance-attribute

rules: list[TrafficRule]

The rules to load, each with a name unique across the defaults and your own. Loading a rule does not activate it; the env's rules_active decides what rules are active.

defaults classmethod

defaults() -> TrafficRulesConfig

The two braking rules PGTG ships with, for four-way and T-crossings.

Source code in pgtg/traffic/rules.py

@classmethod
def defaults(cls) -> TrafficRulesConfig:
    """The two braking rules PGTG ships with, for four-way and T-crossings."""
    return cls.model_validate(TRAFFIC_RULES_FILE)

Map creation

pgtg.map.authoring.create_custom_map

create_custom_map(
    filename, width=3, height=3, seed=42, **rm_kwargs
)

Generate one random map and save it as a JSON.

Extra keyword args are forwarded to RandomMapContext (e.g. connections_percentage, obstacle_probability, start_position, goal_position).

Source code in pgtg/map/authoring.py

def create_custom_map(filename, width=3, height=3, seed=42, **rm_kwargs):
    """Generate one random map and save it as a JSON.

    Extra keyword args are forwarded to [`RandomMapContext`][pgtg.contexts.RandomMapContext]
    (e.g. `connections_percentage`, `obstacle_probability`, `start_position`, `goal_position`).
    """
    rng = np.random.default_rng(seed)
    rm_context = RandomMapContext(width=width, height=height, **rm_kwargs)
    map_plan = generate_map(rm_context, rng)

    map_dict = convert_numpy_types(map_plan.to_dict())

    with open(filename, "w", encoding="utf-8") as f:
        json.dump(map_dict, f, indent=2)

    print(f"Map saved to: {filename}")
    print(f"Start: {map_plan.start}")
    print(f"Goal: {map_plan.goal}")

    return map_dict

Logging

pgtg.core.logger.PGTGLogger

Access to the process-wide "pgtg" logger.

PGTG attaches only a NullHandler and leaves propagate enabled, so the application decides where PGTG's records go. Warnings and errors still reach stderr through Python's standard handler even when nothing is configured.

To get the colorized stderr output PGTG renders itself, opt in with enable_console_handler().

Source code in pgtg/core/logger.py

class PGTGLogger:
    """Access to the process-wide `"pgtg"` logger.

    PGTG  attaches only a `NullHandler` and leaves `propagate` enabled,
    so the *application* decides where PGTG's records go.
    Warnings and errors still reach stderr through Python's
    standard handler even when nothing is configured.

    To get the colorized stderr output PGTG renders itself, opt in with
    [`enable_console_handler()`][pgtg.core.logger.PGTGLogger.enable_console_handler].
    """

    times: ClassVar[dict[str, int]] = defaultdict(int)
    """Accumulated `TimedExecution` durations in nanoseconds, keyed by identifier."""

    _logger: ClassVar[logging.Logger | None] = None
    _console_handler: ClassVar[logging.Handler | None] = None

    _LOGGING_FORMAT = "[%(asctime)s] [%(levelname)s] %(message)s"
    _DATE_FORMAT = "%H:%M:%S"

    class ColorfulFormatter(logging.Formatter):
        NC = "\033[0m"
        BOLD = "\033[1m"
        RED = "\033[91m"
        YELLOW = "\033[93m"
        BLUE = "\033[94m"
        CYAN = "\033[96m"
        GRAY = "\033[90m"

        COLORS: ClassVar[dict[int, str]] = {
            logging.DEBUG: GRAY,
            logging.INFO: CYAN,
            logging.WARNING: YELLOW + BOLD,
            logging.ERROR: RED + BOLD,
            logging.CRITICAL: RED + BOLD,
        }

        _FORMATTERS: ClassVar[dict[int, logging.Formatter]] = {}

        def format(self, record: logging.LogRecord) -> str:
            formatter = self._FORMATTERS.get(record.levelno)
            if formatter is None:
                color = self.COLORS.get(record.levelno, self.GRAY)
                formatter = logging.Formatter(
                    color + PGTGLogger._LOGGING_FORMAT + self.NC,
                    datefmt=PGTGLogger._DATE_FORMAT,
                )
                self._FORMATTERS[record.levelno] = formatter
            return formatter.format(record)

    @staticmethod
    def te_cb(te: TimedExecution) -> None:
        PGTGLogger.times[te.identifier] += te.elapsed_time

    @staticmethod
    def get_logger() -> logging.Logger:
        """Returns the `"pgtg"` logger, creating it on first use."""
        if PGTGLogger._logger is None:
            logger = logging.getLogger(LOGGER_NAME)
            logger.addHandler(logging.NullHandler())
            logger.setLevel(logging.WARNING)
            PGTGLogger._logger = logger
        return PGTGLogger._logger

    @staticmethod
    def enable_console_handler(
        *,
        colorize: bool | None = None,
        level: int | None = None,
    ) -> logging.Logger:
        """Attaches PGTG's own stderr handler to the package logger.

        Opt-in, for scripts and notebooks that want PGTG's formatting without
        configuring logging themselves. Applications that manage their own logging
        should not call this and can rely on propagation instead.

        Args:
            colorize: Emit ANSI colors. `None` (default) enables them only when stderr
                is a TTY, so redirected output and CI logs stay free of escape codes.
            level: Optionally set the package logger's level at the same time. This is
                process-wide, as the logger is shared by every PGTG environment.

        Returns:
            The package logger.
        """
        logger = PGTGLogger.get_logger()

        if colorize is None:
            colorize = sys.stderr is not None and sys.stderr.isatty()

        formatter = (
            PGTGLogger.ColorfulFormatter()
            if colorize
            else logging.Formatter(PGTGLogger._LOGGING_FORMAT, datefmt=PGTGLogger._DATE_FORMAT)
        )

        if PGTGLogger._console_handler is None:
            PGTGLogger._console_handler = logging.StreamHandler()
            logger.addHandler(PGTGLogger._console_handler)
        PGTGLogger._console_handler.setFormatter(formatter)

        if level is not None:
            logger.setLevel(level)

        return logger

    @staticmethod
    def disable_console_handler() -> None:
        """Removes the handler added by `enable_console_handler()`, if any."""
        if PGTGLogger._console_handler is not None:
            PGTGLogger.get_logger().removeHandler(PGTGLogger._console_handler)
            PGTGLogger._console_handler = None

get_logger staticmethod

get_logger() -> logging.Logger

Returns the "pgtg" logger, creating it on first use.

Source code in pgtg/core/logger.py

@staticmethod
def get_logger() -> logging.Logger:
    """Returns the `"pgtg"` logger, creating it on first use."""
    if PGTGLogger._logger is None:
        logger = logging.getLogger(LOGGER_NAME)
        logger.addHandler(logging.NullHandler())
        logger.setLevel(logging.WARNING)
        PGTGLogger._logger = logger
    return PGTGLogger._logger

enable_console_handler staticmethod

enable_console_handler(
    *,
    colorize: bool | None = None,
    level: int | None = None,
) -> logging.Logger

Attaches PGTG's own stderr handler to the package logger.

Opt-in, for scripts and notebooks that want PGTG's formatting without configuring logging themselves. Applications that manage their own logging should not call this and can rely on propagation instead.

Parameters:

Name	Type	Description	Default
colorize	bool | None	Emit ANSI colors. None (default) enables them only when stderr is a TTY, so redirected output and CI logs stay free of escape codes.	None
level	int | None	Optionally set the package logger's level at the same time. This is process-wide, as the logger is shared by every PGTG environment.	None

Returns:

Type	Description
Logger	The package logger.

Source code in pgtg/core/logger.py

@staticmethod
def enable_console_handler(
    *,
    colorize: bool | None = None,
    level: int | None = None,
) -> logging.Logger:
    """Attaches PGTG's own stderr handler to the package logger.

    Opt-in, for scripts and notebooks that want PGTG's formatting without
    configuring logging themselves. Applications that manage their own logging
    should not call this and can rely on propagation instead.

    Args:
        colorize: Emit ANSI colors. `None` (default) enables them only when stderr
            is a TTY, so redirected output and CI logs stay free of escape codes.
        level: Optionally set the package logger's level at the same time. This is
            process-wide, as the logger is shared by every PGTG environment.

    Returns:
        The package logger.
    """
    logger = PGTGLogger.get_logger()

    if colorize is None:
        colorize = sys.stderr is not None and sys.stderr.isatty()

    formatter = (
        PGTGLogger.ColorfulFormatter()
        if colorize
        else logging.Formatter(PGTGLogger._LOGGING_FORMAT, datefmt=PGTGLogger._DATE_FORMAT)
    )

    if PGTGLogger._console_handler is None:
        PGTGLogger._console_handler = logging.StreamHandler()
        logger.addHandler(PGTGLogger._console_handler)
    PGTGLogger._console_handler.setFormatter(formatter)

    if level is not None:
        logger.setLevel(level)

    return logger

disable_console_handler staticmethod

disable_console_handler() -> None

Removes the handler added by enable_console_handler(), if any.

Source code in pgtg/core/logger.py

@staticmethod
def disable_console_handler() -> None:
    """Removes the handler added by `enable_console_handler()`, if any."""
    if PGTGLogger._console_handler is not None:
        PGTGLogger.get_logger().removeHandler(PGTGLogger._console_handler)
        PGTGLogger._console_handler = None