Spaces:
Running
Running
File size: 3,644 Bytes
0643cf5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | # Copyright (c) 2025-2026, RTE (https://www.rte-france.com)
# This Source Code Form is subject to the terms of the Mozilla Public License, version 2.0.
# If a copy of the Mozilla Public License, version 2.0 was not distributed with this file,
# you can obtain one at http://mozilla.org/MPL/2.0/.
# SPDX-License-Identifier: MPL-2.0
"""Canonical example: a baseline random recommender.
Does NOT need the overflow-graph step. Samples uniformly from the
operator's action dictionary, augmented at runtime with simple
reconnection / load-shedding / curtailment actions derived from the
post-fault observation. Useful as a sanity-check baseline against
the expert system.
"""
from __future__ import annotations
import logging
import random
from typing import Any, Dict, List
from expert_op4grid_recommender.models.base import (
ParamSpec,
RecommenderInputs,
RecommenderModel,
RecommenderOutput,
)
from expert_backend.recommenders.network_existence import (
filter_to_existing_network_elements,
)
from expert_backend.recommenders.synthetic_actions import (
build_curtailment_actions,
build_load_shedding_actions,
build_reconnection_actions,
)
logger = logging.getLogger(__name__)
class RandomRecommender(RecommenderModel):
name = "random"
label = "Random"
requires_overflow_graph = False
@classmethod
def params_spec(cls) -> List[ParamSpec]:
return [
ParamSpec(
"n_prioritized_actions",
"N Prioritized Actions",
"int",
default=5,
min=1,
max=50,
description="Total number of actions sampled uniformly.",
),
]
def recommend(self, inputs: RecommenderInputs, params: dict) -> RecommenderOutput:
n = int(params.get("n_prioritized_actions", 5))
env = inputs.env
obs = inputs.obs_defaut
# Drop dict entries whose target VL / line isn't on the loaded
# network (e.g. dict shipped for a larger grid). Same defensive
# check as RandomOverflow — the expert pipeline doesn't run for
# this model so without it bad entries leak through to the UI.
dict_ids = list((inputs.dict_action or {}).keys())
dict_ids = filter_to_existing_network_elements(
dict_ids, inputs.dict_action, inputs.network,
)
pool: Dict[str, Any] = {}
# Materialise dict_action entries into grid2op/pypowsybl actions.
# Entries without a usable `content` field are skipped.
for action_id in dict_ids:
desc = (inputs.dict_action or {}).get(action_id)
content = desc.get("content") if isinstance(desc, dict) else None
if content is None:
continue
try:
pool[action_id] = env.action_space(content)
except Exception as e:
logger.debug("Skipping dict action %s: %s", action_id, e)
# Augment with synthetic reconnection / shedding / curtailment.
pool.update(build_reconnection_actions(env, inputs.non_connected_reconnectable_lines))
pool.update(build_load_shedding_actions(env, obs))
pool.update(build_curtailment_actions(env, obs))
if not pool:
logger.warning("RandomRecommender: empty action pool, returning {}")
return RecommenderOutput(prioritized_actions={})
chosen_ids = random.sample(list(pool.keys()), min(n, len(pool)))
return RecommenderOutput(
prioritized_actions={aid: pool[aid] for aid in chosen_ids},
action_scores={},
)
|