citylearn.internal.robustness module

class citylearn.internal.robustness.CityLearnRobustnessService(env: CityLearnEnv)

Bases: object

Dataset-driven robustness perturbations for observations, actions and asset availability.

CONTROL_FEATURES = {'*', 'both', 'control'}

DEFAULT_MISSING_REPLACEMENT_VALUE = -9999.0

KPI_KEYS = ('robustness_events_count', 'robustness_active_time_step_count', 'robustness_observation_corruption_count', 'robustness_forecast_corruption_count', 'robustness_action_corruption_count', 'robustness_asset_unavailable_time_step_count', 'robustness_missing_observation_count', 'robustness_action_dropout_count')

MODULE_ALIASES = {'action': 'actions', 'asset': 'assets', 'forecast': 'forecasts', 'observation': 'observations'}

TELEMETRY_FEATURES = {'*', 'both', 'telemetry'}

VALID_EVENT_DOMAINS = {'ACTUATOR_CHANNEL', 'ASSET_AVAILABILITY', 'ASSET_CONNECTION', 'COMMUNICATION_LINK', 'SENSOR_CHANNEL', 'VALUE_QUALITY'}

VALID_MODES = {'action': {'bias', 'clip', 'delay', 'dropout', 'noise', 'stuck'}, 'asset': {'unavailable'}, 'forecast': {'bias', 'clip', 'missing', 'noise', 'stuck'}, 'observation': {'bias', 'clip', 'missing', 'noise', 'stuck'}}

VALID_TARGET_TYPES = {'building', 'charger', 'deferrable_appliance', 'district', 'ev', 'pv', 'storage'}

adjust_observation_space(observation_space: Any) → Any

Conservatively expand observation spaces for sentinel/clip values.

apply_actions(parsed_actions: Sequence[Mapping[str, Any]]) → Sequence[Mapping[str, Any]]

Apply action-channel and asset-control events to parsed action dictionaries.

apply_observations(observations: Any) → Any

Apply observation/forecast/telemetry events to the agent-facing payload.

property events: Sequence[RobustnessEvent]

get_metadata() → Mapping[str, Any]

property history: Mapping[int, Mapping[str, Any]]

initialize()

Load event definitions after env.root_directory is available.

observation_meta(*, current_step: int | None = None) → Mapping[str, Any]

observation_values() → Mapping[str, float]

Return entity robustness bundle values for the district table.

reset()

Reset episode-local state without reloading event definitions.

runtime_status(*, current_step: int | None = None) → Mapping[str, Any]

Return raw runtime evidence without deriving TI-MARL health.

fault_mode is deliberately preserved as the simulator cause. The consumer decides whether the same evidence is healthy, degraded, stale, missing or failed for its own semantic contract.

summarize(building_names: Sequence[str]) → Tuple[Mapping[str, Mapping[str, float]], Mapping[str, float]]

validate_targets()

Validate enabled events against the current flat/entity layout.

class citylearn.internal.robustness.RobustnessEvent(event_id: str, module: str, target_type: str, target_id: str, target_feature: str, start_time_step: int, end_time_step: int, mode: str, event_domain: str, value: float | None, std: float | None, min_value: float | None, max_value: float | None, replacement_value: float | None, delay_steps: int, order: int)

Bases: object

Canonical robustness event loaded from a dataset file.

delay_steps: int

end_time_step: int

event_domain: str

event_id: str

max_value: float | None

min_value: float | None

mode: str

module: str

order: int

replacement_value: float | None

start_time_step: int

std: float | None

target_feature: str

target_id: str

target_type: str

value: float | None