# Copyright 2023 Avaiga Private Limited # # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the # specific language governing permissions and limitations under the License. import typing as t from dateutil import parser import taipy as tp from taipy.core import Cycle, Scenario from taipy.core.notification import CoreEventConsumerBase, EventEntityType from taipy.core.notification.event import Event from taipy.core.notification.notifier import Notifier from taipy.gui import Gui, State from taipy.gui.extension import Element, ElementLibrary, ElementProperty, PropertyType from taipy.gui.utils import _TaipyBase class GuiCoreScenarioAdapter(_TaipyBase): __INNER_PROPS = ["name"] def get(self): data = super().get() if isinstance(data, Scenario): return [ data.id, data.is_primary, data.config_id, data.creation_date, data.get_simple_label(), list(data.tags), [(k, v) for k, v in data.properties.items() if k not in GuiCoreScenarioAdapter.__INNER_PROPS], [(p.id, p.get_simple_label()) for p in data.pipelines.values()], list(data.properties.get("authorized_tags", set())), ] return data @staticmethod def get_hash(): return _TaipyBase._HOLDER_PREFIX + "Sc" class GuiCoreContext(CoreEventConsumerBase): __PROP_SCENARIO_ID = "id" __PROP_SCENARIO_CONFIG_ID = "config" __PROP_SCENARIO_DATE = "date" __PROP_SCENARIO_NAME = "name" __SCENARIO_PROPS = (__PROP_SCENARIO_CONFIG_ID, __PROP_SCENARIO_DATE, __PROP_SCENARIO_NAME) _CORE_CHANGED_NAME = "core_changed" _ERROR_VAR = "gui_core_error" _SCENARIO_SELECTOR_ID_VAR = "gui_core_sc_id" def __init__(self, gui: Gui) -> None: self.gui = gui self.cycles_scenarios: t.Optional[t.List[t.Union[Cycle, Scenario]]] = None self.scenario_configs: t.Optional[t.List[t.Tuple[str, str]]] = None # register to taipy core notification reg_id, reg_queue = Notifier.register() super().__init__(reg_id, reg_queue) self.start() def process_event(self, event: Event): if event.entity_type == EventEntityType.SCENARIO or event.entity_type == EventEntityType.CYCLE: self.cycles_scenarios = None self.gui.broadcast(GuiCoreContext._CORE_CHANGED_NAME, {"scenario": True}) @staticmethod def scenario_adapter(data): if isinstance(data, Cycle): return (data.id, data.name, tp.get_scenarios(data), 0, False) elif isinstance(data, Scenario): return (data.id, data.name, None, 1, data.is_primary) return data def get_scenarios(self): if self.cycles_scenarios is None: self.cycles_scenarios = [] for cycle, scenarios in tp.get_cycles_scenarios().items(): if cycle is None: self.cycles_scenarios.extend(scenarios) else: self.cycles_scenarios.append(cycle) return self.cycles_scenarios def select_scenario(self, state: State, id: str, action: str, payload: t.Dict[str, str]): args = payload.get("args") if args is None or not isinstance(args, list) or len(args) == 0: return scenario_id = args[0] state.assign(GuiCoreContext._SCENARIO_SELECTOR_ID_VAR, scenario_id) def get_scenario_by_id(self, id: str) -> t.Optional[Scenario]: if not id: return None try: return tp.get(id) except Exception: return None def get_scenario_configs(self): if self.scenario_configs is None: configs = tp.Config.scenarios if isinstance(configs, dict): self.scenario_configs = [(id, f"{c.id}") for id, c in configs.items()] return self.scenario_configs def crud_scenario(self, state: State, id: str, action: str, payload: t.Dict[str, str]): args = payload.get("args") if ( args is None or not isinstance(args, list) or len(args) < 3 or not isinstance(args[0], bool) or not isinstance(args[1], bool) or not isinstance(args[2], dict) ): return update = args[0] delete = args[1] data = args[2] name = data.get(GuiCoreContext.__PROP_SCENARIO_NAME) if update: scenario_id = data.get(GuiCoreContext.__PROP_SCENARIO_ID) if delete: try: tp.delete(scenario_id) except Exception as e: state.assign(GuiCoreContext._ERROR_VAR, f"Error deleting Scenario. {e}") else: scenario = tp.get(scenario_id) else: config_id = data.get(GuiCoreContext.__PROP_SCENARIO_CONFIG_ID) scenario_config = tp.Config.scenarios.get(config_id) if scenario_config is None: state.assign(GuiCoreContext._ERROR_VAR, f"Invalid configuration id ({config_id})") return date_str = data.get(GuiCoreContext.__PROP_SCENARIO_DATE) try: date = parser.parse(date_str) if isinstance(date_str, str) else None except Exception as e: state.assign(GuiCoreContext._ERROR_VAR, f"Invalid date ({date_str}).{e}") return try: scenario = tp.create_scenario(scenario_config, date, name) except Exception as e: state.assign(GuiCoreContext._ERROR_VAR, f"Error creating Scenario. {e}") if scenario: with scenario as sc: sc._properties[GuiCoreContext.__PROP_SCENARIO_NAME] = name if props := data.get("properties"): try: for prop in props: key = prop.get("key") if key and key not in GuiCoreContext.__SCENARIO_PROPS: sc._properties[key] = prop.get("value") state.assign(GuiCoreContext._ERROR_VAR, "") except Exception as e: state.assign(GuiCoreContext._ERROR_VAR, f"Error creating Scenario. {e}") def broadcast_core_changed(self): self.gui.broadcast(GuiCoreContext._CORE_CHANGED_NAME, "") class GuiCore(ElementLibrary): __LIB_NAME = "taipy_gui_core" __CTX_VAR_NAME = f"__{__LIB_NAME}_Ctx" __elts = { "scenario_selector": Element( "value", { "show_add_button": ElementProperty(PropertyType.dynamic_boolean, True), "display_cycles": ElementProperty(PropertyType.dynamic_boolean, True), "show_primary_flag": ElementProperty(PropertyType.dynamic_boolean, True), "value": ElementProperty(PropertyType.lov_value), "on_change": ElementProperty(PropertyType.function), }, inner_properties={ "scenarios": ElementProperty(PropertyType.lov, f"{{{__CTX_VAR_NAME}.get_scenarios()}}"), "on_scenario_crud": ElementProperty(PropertyType.function, f"{{{__CTX_VAR_NAME}.crud_scenario}}"), "configs": ElementProperty(PropertyType.react, f"{{{__CTX_VAR_NAME}.get_scenario_configs()}}"), "core_changed": ElementProperty(PropertyType.broadcast, GuiCoreContext._CORE_CHANGED_NAME), "error": ElementProperty(PropertyType.react, f"{{{GuiCoreContext._ERROR_VAR}}}"), "type": ElementProperty(PropertyType.inner, Scenario), "adapter": ElementProperty(PropertyType.inner, GuiCoreContext.scenario_adapter), "scenario_edit": ElementProperty( GuiCoreScenarioAdapter, f"{{{__CTX_VAR_NAME}.get_scenario_by_id({GuiCoreContext._SCENARIO_SELECTOR_ID_VAR})}}", ), "on_scenario_select": ElementProperty(PropertyType.function, f"{{{__CTX_VAR_NAME}.select_scenario}}"), }, ) } def get_name(self) -> str: return GuiCore.__LIB_NAME def get_elements(self) -> t.Dict[str, Element]: return GuiCore.__elts def get_scripts(self) -> t.List[str]: return ["lib/taipy-gui-core.js"] def on_init(self, gui: Gui) -> t.Optional[t.Tuple[str, t.Any]]: return GuiCore.__CTX_VAR_NAME, GuiCoreContext(gui) def on_user_init(self, state: State): state._add_attribute(GuiCoreContext._ERROR_VAR) state._gui._bind_var_val(GuiCoreContext._ERROR_VAR, "") state._add_attribute(GuiCoreContext._SCENARIO_SELECTOR_ID_VAR) state._gui._bind_var_val(GuiCoreContext._SCENARIO_SELECTOR_ID_VAR, "")