citylearn.internal.topology module

class citylearn.internal.topology.CityLearnTopologyService(env: CityLearnEnv)

Bases: object

Schema-driven dynamic topology lifecycle and mutation service.

active_chargers_at(time_step: int, member_id: str) → Mapping[str, Charger]

active_deferrable_appliances_at(time_step: int, member_id: str) → Mapping[str, DeferrableAppliance]

property active_ev_ids: Sequence[str]

active_ev_ids_at(time_step: int) → List[str]

property active_member_ids: Sequence[str]

active_member_ids_at(time_step: int) → List[str]

active_storage_at(time_step: int, member_id: str) → Battery | None

apply_events_for_time_step(time_step: int) → bool

Apply schema events due at the current episode-local time step.

Topology-event timestamps belong to the global dataset timeline, while Environment.time_step is local to the current episode. Replaying events before the episode start reconstructs the composition that must be visible at local step zero; events inside the episode are then applied when their global timestamp is reached.

property enabled: bool

property ev_pool: Mapping[str, ElectricVehicle]

property event_log: Sequence[Mapping[str, Any]]

historical_chargers(member_id: str) → Sequence[Charger]

Return every distinct charger instance that was active for a member.

Dynamic remove/reinstall events create a fresh runtime instance from the immutable schema template. Retaining both instances is necessary for end-of-episode KPIs: the current building view alone otherwise drops EV service and energy recorded before the removal.

historical_deferrable_appliances(member_id: str) → Sequence[DeferrableAppliance]

Return every distinct deferrable instance active for a member.

historical_storages(member_id: str) → Sequence[Battery]

Return every distinct stationary-storage instance active for a member.

Storage removal followed by catalogue-based recommissioning creates a new physical/runtime instance. Keeping both objects prevents the KPI layer from silently discarding throughput and degradation accumulated by the instance that was removed.

initialize(buildings: Sequence[Building], electric_vehicles: Sequence[ElectricVehicle])

Initialize static pools and parse deterministic event stream.

property member_lifecycle: Mapping[str, Mapping[str, Any]]

property member_pool: Mapping[str, Building]

reset()

Reset topology and pools for a fresh episode.

property topology_version: int

topology_version_at(time_step: int) → int

class citylearn.internal.topology.TopologyEvent(event_id: str, time_step: int, operation: str, target_member_id: str | None, target_asset_type: str | None, target_asset_id: str | None, source_member_id: str | None, source_asset_id: str | None, overrides: Mapping[str, Any], order: int)

Bases: object

Canonical in-memory topology event.

event_id: str

operation: str

order: int

overrides: Mapping[str, Any]

source_asset_id: str | None

source_member_id: str | None

target_asset_id: str | None

target_asset_type: str | None

target_member_id: str | None

time_step: int