Being able to listen on Core Events and propagate to the state

[FlorianJacta]

Description
I want to listen to Core events to change my charts, inputs, and outputs depending on Core events. This means propagating changes to the state without using Gui Core visual elements.

Expected change
The documentation shows how to propagate a Core event to all or some Gui's states.
This should appear in the "Core events" section as an example.

Acceptance Criteria

• Propose an implementation design
• Create an application as a demo
• Create related issues

[jrobinAV]

It becomes a high priority due to the ticket related to the scenario selector.

[jrobinAV]

@FlorianJacta @FredLL-Avaiga

What are the variables the application needs to access to be able to change a state variable when receiving a core event?

The GUI instance? Anything else?

[FredLL-Avaiga]

a State is associated to a session via a client id ...

[FlorianJacta]

@FlorianJacta @jrobinAV Here is the discussion we had when talking about the custom filter of the Scenario Selector:

Here is an example of a filter criterion covering the most generic cases that a developer would like to provide:

scenarios = [sc for sc in tp.get_scenarios() if sc.config_id == state.config_id]

As you can see in the code above, the filter depends on the Taipy core events and the GUI state. That means this code should be called at least when tp.get_scenarios() or state.config_id changes.
Usage

Updates based on GUI state change

scenarios = [scenario_1, scenario_2, scenario_3]
<| {selected_scenario} | scenario_selector | scenarios = { scenarios } |>

Note that this example is static, but if the list is only updated based on a GUI state modification (and not on Core events), the on_change callback should do the job.

Updates based on Core events and GUI state changes

Taipy exposes a generic on_core_event(state, event) called for each state every time a Core event is published. This is part of the implementation of ticket #1088.

scenarios = [sc for sc in tp.get_scenarios() if sc.config_id == config_id]

def on_core_event(state, event):
    if event.entity_type == EventEntityType.SCENARIO 
        If event.operation == EventOperation.CREATION or event.operation == EventOperation.DELETION:
            state.scenarios = [sc for sc in tp.get_scenarios() if sc.config_id == state.config_id]
    
<| {selected_scenario} | scenario_selector | scenarios = { scenarios } |>

The advantage is that this API is fully generic and easy to understand since it is similar to GUI callbacks (on_init, on_change, on_action, on_exception, on_navigate). Note that even if it is similar, it is still different as this callback is not called for one unique state but for all.
The drawback is that it is called for all the states and events. We should use the next option if the state is not needed in the filter.

Updates based on Core events only

Taipy exposes an on_core_event(event) called every time a Core event is published. This is part of the implementation of ticket #405. We explicitly let the developer be responsible for broadcasting the value to the states if needed. The advantage of this implementation is

scenarios = tp.get_primary_scenarios()

def on_core_event(event):
    if event.EventType == “SCENARIO”:
        if event.operation == EventOperation.CREATION or event.operation == EventOperation.DELETION:
            gui.broadcast(“scenarios”, tp.get_primary_scenarios())

<| {selected_scenario} | selector| lov= { scenarios } |>

Implementing tickets #941 and #1207 will allow us to cover previous cases in which the updates were based on core events and state changes.

scenarios = [sc for sc in tp.get_scenarios() if sc.config_id == config_id]

def filter_scenarios(state, scenarios):
     state.scenarios = [sc for sc in scenarios if sc.config_id == state.config_id]
    
def on_core_event(event):
    if event.entity_type == EventEntityType.SCENARIO 
        If event.operation == EventOperation.CREATION or event.operation == EventOperation.DELETION:
          broadcast_callback(gui, filter_scenarios, [tp.get_scenarios()] )

<| {selected_scenario} | scenario_selector | scenarios = { scenarios } |>

[jrobinAV]

Here is the ideal API solution proposed:

import taipy as tp
from taipy import Config, Core, Gui, Scope
from taipy.gui import notify, get_state_id, invoke_callback
from taipy.core.notification import Notifier, CoreEventConsumerBase, EventOperation, EventEntityType

value = "Default text."

class SpecificCoreConsumer(CoreEventConsumerBase):

    def __init__(self, gui):
        self.gui = gui
        reg_id, queue = Notifier.register()  # Adapt the registration to the events you want to listen to
        super().__init__(reg_id, queue)


    def process_event(self, event):
        if event.operation == EventOperation.CREATION:
            if event.entity_type == EventEntityType.DATA_NODE:
                self.gui.broadcast("value", "Propagated text!")
                # Alternative API
                args = [] # other args to pass to the callback
                self.gui.broadcast_callback((lambda state, args: state.value="Propagated text!"), args)

def create_global_dn(state):
    tp.create_global_data_node(Config.data_nodes["dataset"])

if __name__ == "__main__":
    Config.configure_data_node("dataset", scope=Scope.GLOBAL, default_data=42)
    core = Core()
    gui = Gui(page="""
<|{value}|text|>

<|Press me!|button|on_action=create_global_dn|>
""")
    core.run()
    SpecificCoreConsumer(gui).start()
    gui.run()

As we can see, we propose that the developer instantiate a CoreEventConsumerBase to listen to core events.
The consumer should be started after running the Core service.
From this consumer, the developer can register only for the events he/she needs and freely process them. In particular, since the GUI instance can be stored as a class attribute, the developer should be able to broadcast a variable change or a callback to all the states.

Today, sending a callback to all the states from the GUI instance (and not from a particular state) has not yet been implemented. A proof of concept using broadcasting a shared variable from a state has been implemented, but does not meet the requirements:

import taipy as tp
from taipy import Config, Core, Gui, Scope
from taipy.gui import notify, get_state_id, invoke_callback
from taipy.core.notification import Notifier, CoreEventConsumerBase, EventOperation, EventEntityType

value = "Default text"

class SpecificCoreConsumer(CoreEventConsumerBase):

    def __init__(self, gui):
        self.gui = gui
        reg_id, queue = Notifier.register()  # Adapt the registration to the events you want to listen to
        super().__init__(reg_id, queue)


    def process_event(self, event):
        if event.operation == EventOperation.CREATION:
            if event.entity_type == EventEntityType.DATA_NODE:
                global state_id
                invoke_callback(self.gui, state_id, lambda s: s.broadcast("value", "Propagated text !!"), [])

state_id = None

def create_global_dn(state):
    global state_id
    state_id = get_state_id(state)
    tp.create_global_data_node(Config.data_nodes["dataset"])


if __name__ == "__main__":
    Config.configure_data_node("dataset", scope=Scope.GLOBAL, default_data=42)
    core = Core()
    gui = Gui(page="""
<|{value}|text|>

<|Press me!|button|on_action=create_global_dn|>
""")
    gui.add_shared_variable("value")
    core.run()
    SpecificCoreConsumer(gui).start()
    gui.run()

• The variable must be declared as a shared variable. In our use case, we want to be able to broadcast any variable.
• The broadcast must be done from a state. So we need to get a state or state id somehow. It can be any state. We do it from a button callback by storing the state id in a global variable. Then, we call the invoke callback with some specific parameters.

[FlorianJacta]

LGTM

[jrobinAV]

The Gui is now able to broadcast a callback to all clients, so the only remaining part is on taipy-doc.

#416

[github-actions]

This issue has been labelled as "🥶Waiting for contributor" because it has been inactive for more than 14 days. If you would like to continue working on this issue, please add another comment or create a PR that links to this issue. If a PR has already been created which refers to this issue, then you should explicitly mention this issue in the relevant PR. Otherwise, you will be unassigned in 14 days. For more information please refer to the contributing guidelines.

[jrobinAV]

A Pull request is opened. It remains to write some real examples.

[FlorianJacta]

Revision of JR example on how to notify the user of a Core event:

import taipy as tp
from taipy import Config, Core, Gui, Scope
from taipy.gui import notify
import taipy.gui.builder as tgb
from taipy.core.notification import (
    Notifier,
    CoreEventConsumerBase,
    EventOperation,
    EventEntityType,
)
from taipy.core import SubmissionStatus


##### Configuration and Functions #####
from taipy import Config


def build_message(name: str):
    return f"Hello {name}!"


name_data_node_cfg = Config.configure_data_node(id="input_name", default_data="Florian")
message_data_node_cfg = Config.configure_data_node(id="message")
build_msg_task_cfg = Config.configure_task(
    "build_msg", build_message, name_data_node_cfg, message_data_node_cfg
)
scenario_cfg = Config.configure_scenario("scenario", task_configs=[build_msg_task_cfg])
#### Listen on Core Events ####


value = "Default text"


class SpecificCoreConsumer(CoreEventConsumerBase):
    def __init__(self, gui):
        self.gui = gui
        reg_id, queue = (
            Notifier.register()
        )  # Adapt the registration to the events you want to listen to
        super().__init__(reg_id, queue)

    def process_event(self, event):
        if event.operation == EventOperation.CREATION:
            if event.entity_type == EventEntityType.SCENARIO:
                self.gui.broadcast_callback(notify_users_of_creation)
        elif event.operation == EventOperation.UPDATE:
            if event.entity_type == EventEntityType.SUBMISSION:
                print(event)
                if event.attribute_value == SubmissionStatus.COMPLETED:
                    scenario_id = event.metadata["origin_entity_id"]
                    scenario = tp.get(scenario_id)
                    new_value_of_dn = scenario.message.read()
                    self.gui.broadcast_callback(
                        notify_users_of_update, [new_value_of_dn]
                    )
        else:
            pass


#### Notification function to be called ####


def notify_users_of_creation(state):
    state.value = "Scenario created and submitted"
    notify(state, "s", "Scenario Created")


def notify_users_of_update(state, new_value_of_dn):
    print("Value of Data Node:", new_value_of_dn)
    state.value = f"Data Node updated with value: {new_value_of_dn}"
    notify(state, "i", "Data Node Updated")


#### Normal callbacks ####


def create_and_submit_scenario(state):
    scenario = tp.create_scenario(config=scenario_cfg)
    tp.submit(scenario)


#### Page ####

with tgb.Page() as page:
    tgb.text("{value}")
    tgb.button("Press me!", on_action=create_and_submit_scenario)


if __name__ == "__main__":
    core = Core()
    gui = Gui(page)
    core.run()
    SpecificCoreConsumer(gui).start()
    gui.run()

[2024-08-22 17:52:03.010][Taipy][INFO] job JOB_build_msg_56045147-372b-4968-85b5-c0a70d38cf6e is completed.

Value of Data Node: Hello Florian

[github-actions]

This issue has been labelled as "🥶Waiting for contributor" because it has been inactive for more than 14 days. If you would like to continue working on this issue, please add another comment or create a PR that links to this issue. If a PR has already been created which refers to this issue, then you should explicitly mention this issue in the relevant PR. Otherwise, you will be unassigned in 14 days. For more information please refer to the contributing guidelines.

[jrobinAV]

We don't have an example in the doc where only part of the gui states are updated when a core event is processed. @FlorianJacta Do you know how to implement that?

[FlorianJacta]

Implementing it is just adding a if statement inside the callback:

def notify_users_of_creation(state):
    if ...:
        state.value = "Scenario created and submitted"
        notify(state, "s", "Scenario Created")

[jrobinAV]

OK. What about adding the following example to the doc?

We have a scenario variable in the state that contains a selected scenario.
We also have an email variable containing the end user's email.
Whenever a data node update event is processed, we get the data node's parent scenarios.

1. We notify the states where the scenario value is among the parents.
2. We send an email to all other states.

What do you think? @FlorianJacta, @toan-quach ?

[toan-quach]

@jrobinAV I think the example where we added a notification saying a scenario or task is updated is sufficient though?

[jrobinAV]

@FlorianJacta, you know better than I do. What do you think? If it is sufficient, Can we close this issue?

[FlorianJacta]

Let's close it for now as a first version, I think.