Spaces:
Sleeping
Sleeping
| # 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 | |
| # This file is part of Co-Study4Grid a Power Grid Study tool Assistant Interface to help solve contigencies for a grid state under study. | |
| """Type-only declaration of the shared ``RecommenderService`` surface. | |
| ``RecommenderService`` composes ``DiagramMixin`` / ``AnalysisMixin`` / | |
| ``SimulationMixin``; the state lives in ``recommender_service.py`` and the | |
| mixins reach into it through ``self`` (see ``expert_backend/CLAUDE.md`` — | |
| "treat the mixins as one class split across files"). mypy checks each class | |
| in isolation, so every cross-mixin ``self._x`` looked like ``attr-defined``. | |
| This class declares that shared surface ONCE. Each mixin inherits it **only | |
| under** ``TYPE_CHECKING`` (``_Base = object`` at runtime — no behaviour | |
| change, no MRO change), which lets mypy resolve the composition. It is never | |
| instantiated. | |
| Maintenance: when a mixin starts calling a new ``self._x`` / ``self._method`` | |
| that is defined on the *service* or a *sibling* mixin, declare it here. | |
| Attribute value-types are intentionally ``Any`` and methods are permissive | |
| (``*args, **kwargs -> Any``); this kills the composition false-positives | |
| without pretending the service is fully typed. Tighten incrementally. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any | |
| class RecommenderState: | |
| # --- shared instance state (set in RecommenderService.__init__) --- | |
| _analysis_context: Any | |
| _base_network: Any | |
| _cached_env_context: Any | |
| _cached_obs_n: Any | |
| _cached_obs_n1: Any | |
| _cached_obs_n1_elements: tuple[str, ...] | None | |
| _cached_obs_n1_id: Any | |
| _cached_obs_n_id: Any | |
| _dict_action: Any | |
| _generator: Any | |
| _initial_pst_taps: Any | |
| _is_running: Any | |
| _last_action_path: Any | |
| _last_disconnected_elements: Any | |
| _last_result: Any | |
| _last_step1_time: Any | |
| _last_step2_context: Any | |
| _last_step2_signature: Any | |
| _layout_cache: Any | |
| _lf_status_by_variant: Any | |
| _n_state_currents: Any | |
| _overflow_layout_cache: Any | |
| _overflow_layout_mode: Any | |
| _prefetched_base_nad: Any | |
| _prefetched_base_nad_error: Any | |
| _prefetched_base_nad_event: Any | |
| _prefetched_base_nad_thread: Any | |
| _saved_computed_pairs: Any | |
| _simulation_env: Any | |
| # --- concurrency ownership (D3, set in RecommenderService.__init__) --- | |
| _network_lock: Any | |
| _study_gate: Any | |
| _prefetch_generation: Any | |
| _contingency_variant_lru: Any | |
| # --- methods inherited from ModelSelectionMixin (explicit composition | |
| # on RecommenderService; declared here so mixin-internal call sites | |
| # like AnalysisMixin.run_analysis_step2 type-check in isolation) --- | |
| def get_active_model_name(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_compute_overflow_graph(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _apply_model_settings(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _reset_model_settings(self, *args: Any, **kwargs: Any) -> Any: ... | |
| # --- shared methods (defined on the service or a sibling mixin) --- | |
| def _apply_contingency_element(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _apply_target_mw_updates(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _apply_target_tap_updates(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _apply_threshold(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _attach_convergence_from_obs(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _attach_flow_deltas_vs_base(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _augment_combined_actions_with_target_max_rho(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _augment_superposition_result(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _build_action_entry_from_topology(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _build_combined_action_object(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _build_contingency_patch_payload(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _cache_obs_for_variant(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _can_reuse_step2_graph(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _canonicalize_id(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _capture_initial_pst_taps(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_asset_deltas(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_curtailment_details(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_deltas(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_load_shedding_details(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_mw_start_for_scores(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_pst_details(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_redispatch_details(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _compute_vl_topology_diff(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _contingency_variant_id(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_actions_if_needed(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_curtailment(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_load_shedding(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_pst(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_reconnection(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _create_dynamic_redispatch(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _default_nad_parameters(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _diff_action_switches_vs_contingency(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _diff_switches(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _drain_pending_base_nad_prefetch(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _enrich_actions(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _ensure_contingency_state_ready(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _ensure_n_state_ready(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _ensure_pair_simulated(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _extract_sld_svg_and_metadata(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _extract_vl_subtrees_with_edges(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _fetch_n_and_contingency_observations(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _generate_diagram(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_action_mw_start(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_asset_flows(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_base_network(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_contingency_flows(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_contingency_variant(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_disconnected_branches_from_snapshot(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_element_max_currents(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_latest_pdf_path(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_lines_we_care_about(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_monitoring_parameters(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_n_variant(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_network_flows(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_overloaded_lines(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_pst_tap_start(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _get_simulation_env(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _identify_elements_with_pst_fallback(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _inject_action_content_entries(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _is_renewable_gen(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _lf_status_for_variant(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _load_layout(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _load_layout_coords(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _log_dict_action_snapshot(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _log_per_line_rho(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _mw_start_curtailment(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _mw_start_load_shedding(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _mw_start_open_coupling(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _narrow_context_to_selected_overloads(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _network_service(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _normalize_contingency_elements(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _obs_n1_from_context(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _promote_recent_actions_to_dict(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _pst_tap_info_fn(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _register_action_result(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _render_topological_sld(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _require_action(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _resolve_action_description_and_content(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _run_ac_with_fallback(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _select_terminal_for_branch(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _snapshot_contingency_state(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _step2_graph_signature(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _superposition_lines_overloaded(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _terminal_aware_delta(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _upstream_step1_supports_prebuilt_obs(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def _worker(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def compute_superposition(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_action_variant_diagram(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_action_variant_diagram_patch(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_action_variant_sld(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_all_action_ids(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_contingency_diagram(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_contingency_diagram_patch(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_contingency_sld(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_n_sld(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_network_diagram(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_prefetched_base_nad(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_saved_computed_pairs(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def get_topology_preview_sld(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def network_lock(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def try_begin_study_mutation(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def end_study_mutation(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def prefetch_base_nad_async(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def regenerate_overflow_graph(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def reset(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def restore_analysis_context(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def run_analysis(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def run_analysis_step1(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def run_analysis_step2(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def simulate_manual_action(self, *args: Any, **kwargs: Any) -> Any: ... | |
| def update_config(self, *args: Any, **kwargs: Any) -> Any: ... | |