taipy/taipy/gui/gui.py

# Copyright 2021-2025 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.

from __future__ import annotations

import contextlib
import importlib
import json
import math
import os
import re
import sys
import tempfile
import time
import typing as t
import uuid
import warnings
from importlib import metadata, util
from importlib.util import find_spec
from inspect import currentframe, getabsfile, iscoroutinefunction, ismethod, ismodule
from pathlib import Path
from threading import Thread, Timer
from types import FrameType, FunctionType, LambdaType, ModuleType, SimpleNamespace
from urllib.parse import unquote, urlencode, urlparse

import markdown as md_lib
import tzlocal
from flask import (
    Blueprint,
    Flask,
    g,
    has_app_context,
    jsonify,
    request,
    send_file,
    send_from_directory,
)
from werkzeug.utils import secure_filename

import __main__  # noqa: F401
from taipy.common.logger._taipy_logger import _TaipyLogger

if util.find_spec("pyngrok"):
    from pyngrok import ngrok  # type: ignore[reportMissingImports]

from ._default_config import _default_stylekit, default_config
from ._event_context_manager import _EventManager
from ._hook import _Hooks
from ._page import _Page
from ._renderers import _EmptyPage
from ._renderers._markdown import _TaipyMarkdownExtension
from ._renderers.factory import _Factory
from ._renderers.json import _TaipyJsonEncoder
from ._renderers.utils import _get_columns_dict
from ._warnings import TaipyGuiWarning, _warn
from .builder import _ElementApiGenerator
from .config import Config, ConfigParameter, _Config
from .custom import Page as CustomPage
from .custom.utils import get_current_resource_handler, is_in_custom_page_context
from .data.content_accessor import _ContentAccessor
from .data.data_accessor import _DataAccessors
from .data.data_format import _DataFormat
from .data.data_scope import _DataScopes
from .extension.library import Element, ElementLibrary
from .page import Page
from .partial import Partial
from .server import _Server
from .state import State, _AsyncState, _GuiState
from .types import _WsType
from .utils import (
    _delscopeattr,
    _DoNotUpdate,
    _filter_locals,
    _function_name,
    _get_broadcast_var_name,
    _get_client_var_name,
    _get_css_var_value,
    _get_expr_var_name,
    _get_lambda_id,
    _get_module_name_from_frame,
    _get_non_existent_file_path,
    _get_page_from_module,
    _getscopeattr,
    _getscopeattr_drill,
    _hasscopeattr,
    _is_function,
    _is_in_notebook,
    _is_unnamed_function,
    _LocalsContext,
    _MapDict,
    _setscopeattr,
    _setscopeattr_drill,
    _TaipyBase,
    _TaipyContent,
    _TaipyContentHtml,
    _TaipyContentImage,
    _TaipyData,
    _TaipyLov,
    _TaipyLovValue,
    _to_camel_case,
    _variable_decode,
    is_debugging,
)
from .utils._adapter import _Adapter
from .utils._bindings import _Bindings
from .utils._evaluator import _Evaluator
from .utils._variable_directory import _is_moduled_variable, _VariableDirectory
from .utils.chart_config_builder import _build_chart_config
from .utils.table_col_builder import _enhance_columns
from .utils.threads import _invoke_async_callback


class Gui:
    """Entry point for the Graphical User Interface generation.

    !!! note
        This class belongs to and is documented in the `taipy.gui` package, but it is
        accessible from the top `taipy` package to simplify its access, allowing to
        use:
        ```py
        from taipy import Gui
        ```
    """

    __root_page_name = "TaiPy_root_page"
    __env_filename = "taipy.gui.env"
    __UI_BLOCK_NAME = "TaipyUiBlockVar"
    __MESSAGE_GROUPING_NAME = "TaipyMessageGrouping"
    __ON_INIT_NAME = "TaipyOnInit"
    __ARG_CLIENT_ID = "client_id"
    __INIT_URL = "taipy-init"
    __JSX_URL = "taipy-jsx"
    __CONTENT_ROOT = "taipy-content"
    __UPLOAD_URL = "taipy-uploads"
    _EXTENSION_ROOT = "taipy-extension"
    __USER_CONTENT_URL = "taipy-user-content"
    __BROADCAST_G_ID = "taipy_broadcasting"
    __BRDCST_CALLBACK_G_ID = "taipy_brdcst_callback"
    __SELF_VAR = "__gui"
    __DO_NOT_UPDATE_VALUE = _DoNotUpdate()
    _HTML_CONTENT_KEY = "__taipy_html_content"
    __USER_CONTENT_CB = "custom_user_content_cb"
    __ROBOTO_FONT = "https://fonts.googleapis.com/css?family=Roboto:300,400,500,700&display=swap"
    __DOWNLOAD_ACTION = "__Taipy__download_csv"
    __DOWNLOAD_DELETE_ACTION = "__Taipy__download_delete_csv"
    __DEFAULT_FAVICON_URL = "https://raw.githubusercontent.com/Avaiga/taipy-assets/develop/favicon.png"

    __RE_HTML = re.compile(r"(.*?)\.html$")
    __RE_MD = re.compile(r"(.*?)\.md$")
    __RE_PY = re.compile(r"(.*?)\.py$")
    __RE_PAGE_NAME = re.compile(r"^[\w\-\/]+$")

    __reserved_routes: t.List[str] = [
        __INIT_URL,
        __JSX_URL,
        __CONTENT_ROOT,
        __UPLOAD_URL,
        _EXTENSION_ROOT,
        __USER_CONTENT_URL,
    ]

    __LOCAL_TZ = str(tzlocal.get_localzone())

    __extensions: t.Dict[str, t.List[ElementLibrary]] = {}

    __shared_variables: t.List[str] = []

    __content_providers: t.Dict[type, t.Callable[..., str]] = {}

    # See set_unsupported_data_converter()
    __unsupported_data_converter: t.Optional[t.Callable] = None

    def __init__(
        self,
        page: t.Optional[t.Union[str, Page]] = None,
        pages: t.Optional[dict] = None,
        css_file: t.Optional[str] = None,
        path_mapping: t.Optional[dict] = None,
        env_filename: t.Optional[str] = None,
        libraries: t.Optional[t.List[ElementLibrary]] = None,
        flask: t.Optional[Flask] = None,
        script_paths: t.Optional[t.Union[str, Path, t.List[t.Union[str, Path]]]] = None,
    ):
        """Initialize a new Gui instance.

        Arguments:
            page (Optional[Union[str, Page^]]): An optional `Page^` instance that is used
                when there is a single page in this interface, referenced as the *root*
                page (located at `/`).<br/>
                If *page* is a raw string and if it holds a path to a readable file then
                a `Markdown^` page is built from the content of that file.<br/>
                If *page* is a string that does not indicate a path to readable file then
                a `Markdown^` page is built from that string.<br/>
                Note that if *pages* is provided, those pages are added as well.
            pages (Optional[dict]): Used if you want to initialize this instance with a set
                of pages.<br/>
                The method `(Gui.)add_pages(pages)^` is called if *pages* is not None.
                You can find details on the possible values of this argument in the
                documentation for this method.
            css_file (Optional[str]): A pathname to a CSS file that gets used as a style sheet in
                all the pages.<br/>
                The default value is a file that has the same base name as the Python
                file defining the `main` function, sitting next to this Python file,
                with the `.css` extension.
            path_mapping (Optional[dict]): A dictionary that associates a URL prefix to
                a path in the server file system.<br/>
                If the assets of your application are located in */home/me/app_assets* and
                you want to access them using only '*assets*' in your application, you can
                set *path_mapping={"assets": "/home/me/app_assets"}*. If your application
                then requests the file *"/assets/images/logo.png"*, the server searches
                for the file  *"/home/me/app_assets/images/logo.png"*.<br/>
                If empty or not defined, access through the browser to all resources under the directory
                of the main Python file is allowed.
            env_filename (Optional[str]): An optional file from which to load application
                configuration variables (see the
                [Configuration](../../../../../userman/advanced_features/configuration/gui-config.md#configuring-the-gui-instance)
                section of the User Manual for details.)<br/>
                The default value is "taipy.gui.env"
            libraries (Optional[List[ElementLibrary]]): An optional list of extension library
                instances that pages can reference.<br/>
                Using this argument is equivalent to calling `(Gui.)add_library()^` for each
                list's elements.
            flask (Optional[Flask]): An optional instance of a Flask application object.<br/>
                If this argument is set, this `Gui` instance will use the value of this argument
                as the underlying server. If omitted or set to None, this `Gui` will create its
                own Flask application instance and use it to serve the pages.
            script_paths (Optional[Union[str, Path, List[Union[str, Path]]]]):
                Specifies the path(s) to the JavaScript files or external resources used by the application.
                It can be a single URL or path, or a list containing multiple URLs and/or paths.
        """
        # store suspected local containing frame
        self.__frame = t.cast(FrameType, t.cast(FrameType, currentframe()).f_back)
        self.__default_module_name = _get_module_name_from_frame(self.__frame)
        self._set_css_file(css_file)

        # Set the assets URL path
        self.__script_files: list[str] = []
        self.__load_scripts(script_paths)

        # Preserve server config for server initialization
        if path_mapping is None:
            path_mapping = {}
        self._path_mapping = path_mapping
        self._flask = flask

        self._config = _Config()
        self.__content_accessor = None
        self.__accessors: t.Optional[_DataAccessors] = None
        self.__state: t.Optional[State] = None
        self.__bindings = _Bindings(self)
        self.__locals_context = _LocalsContext()
        self.__var_dir = _VariableDirectory(self.__locals_context)

        self.__evaluator: _Evaluator = None  # type: ignore[assignment]
        self.__adapter = _Adapter()
        self.__directory_name_of_pages: t.List[str] = []
        self.__favicon: t.Optional[t.Union[str, Path]] = None

        # default actions
        self.on_action: t.Optional[t.Callable] = None
        """The function called when a control triggers an action, as the result of an end-user's interaction.

        It defaults to the `on_action()` global function defined in the Python
        application. If there is no such function, actions will not trigger anything.<br/>
        The signature of the *on_action* callback function must be:

        - *state*: the `State^` instance of the caller.
        - *id* (optional): a string representing the identifier of the caller.
        - *payload* (optional): an optional payload from the caller.
        """
        self.on_change: t.Optional[t.Callable] = None
        """The function called when a control modifies bound variables, as the result of an end-user's interaction.

        It defaults to the `on_change()` global function defined in the Python
        application. If there is no such function, user interactions will not trigger
        anything.<br/>
        The signature of the *on_change* callback function must be:

        - *state*: the `State^` instance of the caller.
        - *var_name* (str): The name of the variable that triggered this callback.
        - *var_value* (any): The new value for this variable.
        """
        self.on_init: t.Optional[t.Callable] = None
        """The function that is called on the first connection of a new client.

        It defaults to the `on_init()` global function defined in the Python
        application. If there is no such function, the first connection will not trigger
        anything.<br/>

        The signature of the *on_init* callback function must be:

        - *state*: the `State^` instance of the caller.
        """
        self.on_page_load: t.Optional[t.Callable] = None
        """This callback is invoked just before the page content is sent to the front-end.

        It defaults to the `on_page_load()` global function defined in the Python
        application. If there is no such function, page loads will not trigger
        anything.<br/>

        The signature of the *on_page_load* callback function must be:
        - *state*: the `State^` instance of the caller.
        - *page_name*: the name of the page that is being loaded.
        """
        self.on_navigate: t.Optional[t.Callable] = None
        """The function that is called when a page is requested.

        It defaults to the `on_navigate()` global function defined in the Python
        application. If there is no such function, page requests will not trigger
        anything.<br/>

        The signature of the *on_navigate* callback function must be:

        - *state*: the `State^` instance of the caller.
        - *page_name*: the name of the page the user is navigating to.
        - *params* (Optional): the query parameters provided in the URL.

        The *on_navigate* callback function must return the name of the page the user should be
        directed to.
        """
        self.on_exception: t.Optional[t.Callable] = None
        """The function that is called an exception occurs on user code.

        It defaults to the `on_exception()` global function defined in the Python
        application. If there is no such function, exceptions will not trigger
        anything.<br/>

        The signature of the *on_exception* callback function must be:

        - *state*: the `State^` instance of the caller.
        - *function_name*: the name of the function that raised the exception.
        - *exception*: the exception object that was raised.
        """
        self.on_status: t.Optional[t.Callable] = None
        """The function that is called when the status page is shown.

        It defaults to the `on_status()` global function defined in the Python
        application. If there is no such function, status page content shows only the status of the
        server.<br/>

        The signature of the *on_status* callback function must be:

        - *state*: the `State^` instance of the caller.

        It must return raw and valid HTML content as a string.
        """
        self.on_user_content: t.Optional[t.Callable] = None
        """The function that is called when a specific URL (generated by `get_user_content_url()^`) is requested.

        This callback function must return the raw HTML content of the page to be displayed on
        the browser.

        This attribute defaults to the `on_user_content()` global function defined in the Python
        application. If there is no such function, those specific URLs will not trigger
        anything.<br/>
        The signature of the *on_user_content* callback function must be:

        - *state*: the `State^` instance of the caller.
        - *path*: the path provided to the `get_user_content_url()^` to build the URL.
        - *parameters*: An optional dictionary as defined in the `get_user_content_url()^` call.

        The returned HTML content can therefore use both the variables stored in the *state*
        and the parameters provided in the call to `get_user_content_url()^`.
        """

        # sid from client_id
        self.__client_id_2_sid: t.Dict[str, t.Set[str]] = {}

        # Load default config
        self._flask_blueprint: t.List[Blueprint] = []
        self._config._load(default_config)

        # get taipy version
        try:
            gui_file = Path(__file__ or ".").resolve()
            with open(gui_file.parent / "version.json") as version_file:
                self.__version = json.load(version_file)
        except Exception as e:  # pragma: no cover
            _warn("Cannot retrieve version.json file", e)
            self.__version = {}

        # Load Markdown extension
        # NOTE: Make sure, if you change this extension list, that the User Manual gets updated.
        # There's a section that explicitly lists these extensions in
        #      docs/gui/pages.md#markdown-specifics
        self._markdown = md_lib.Markdown(
            extensions=[
                "fenced_code",
                "meta",
                "admonition",
                "sane_lists",
                "tables",
                "attr_list",
                "md_in_html",
                _TaipyMarkdownExtension(gui=self),
            ]
        )

        self.__event_manager = _EventManager()

        # Init Gui Hooks
        _Hooks()._init(self)

        if page:
            self.add_page(name=Gui.__root_page_name, page=page)
        if pages is not None:
            self.add_pages(pages)
        if env_filename is not None:
            self.__env_filename = env_filename
        if libraries is not None:
            for library in libraries:
                Gui.add_library(library)

    def __load_scripts(self, script_paths: t.Optional[t.Union[str, Path, t.List[t.Union[str, Path]]]]):
        if script_paths is None:
            return
        else:
            if isinstance(script_paths, (str, Path)):
                script_paths = [script_paths]

            for script_path in script_paths:
                if isinstance(script_path, str):
                    parsed_url = urlparse(script_path)
                    if parsed_url.netloc:
                        self.__script_files.append(script_path)
                        continue
                    script_path = Path(script_path)

                if (
                    isinstance(script_path, Path)
                    and script_path.exists()
                    and script_path.is_file()
                    and script_path.suffix == ".js"
                ):
                    self.__script_files.append(str(script_path))
                else:
                    _warn(f"Script path '{script_path}' does not exist, is not a file, or is not a JavaScript file.")

            if not self.__script_files:
                _warn("No JavaScript files found in the specified script paths.")

    @staticmethod
    def add_library(library: ElementLibrary) -> None:
        """Add a custom visual element library.

        This application will be able to use custom visual elements defined in this library.

        Arguments:
            library: The custom visual element library to add to this application.

        Multiple libraries with the same name can be added. This allows to split multiple custom visual
        elements in several `ElementLibrary^` instances, but still refer to these elements with the same
        prefix in the page definitions.
        """
        if isinstance(library, ElementLibrary):
            _Factory.set_library(library)
            library_name = library.get_name()
            if library_name.isidentifier():
                libs = Gui.__extensions.get(library_name)
                if libs is None:
                    Gui.__extensions[library_name] = [library]
                else:
                    libs.append(library)
                _ElementApiGenerator().add_library(library)
            else:
                raise NameError(f"ElementLibrary passed to add_library() has an invalid name: '{library_name}'")
        else:  # pragma: no cover
            raise RuntimeError(
                f"add_library() argument should be a subclass of ElementLibrary instead of '{type(library)}'"
            )

    @staticmethod
    def register_content_provider(content_type: type, content_provider: t.Callable[..., str]) -> None:
        """Add a custom content provider.

        The application can use custom content for the `part` block when its *content* property
        is set to an object with type *type*.

        Arguments:
            content_type: The type of the content that triggers the content provider.
            content_provider: The function that converts content of type *type* into an HTML string.
        """  # noqa: E501
        if Gui.__content_providers.get(content_type):
            _warn(f"The type {content_type} is already associated with a provider.")
            return
        if not _is_function(content_provider):
            _warn(f"The provider for {content_type} must be a function.")
            return
        Gui.__content_providers[content_type] = content_provider

    def __process_content_provider(self, state: State, path: str, query: t.Dict[str, str]):
        variable_name = query.get("variable_name")
        content = None
        if variable_name:
            content = _getscopeattr(self, variable_name)
            if isinstance(content, _TaipyContentHtml):
                content = content.get()
            provider_fn = Gui.__content_providers.get(type(content))
            if provider_fn is None:
                # try plotly
                if find_spec("plotly") and find_spec("plotly.graph_objs"):
                    from plotly.graph_objs import Figure as PlotlyFigure  # type: ignore[reportMissingImports]

                    if isinstance(content, PlotlyFigure):

                        def get_plotly_content(figure: PlotlyFigure):
                            return figure.to_html()

                        Gui.register_content_provider(PlotlyFigure, get_plotly_content)
                        provider_fn = get_plotly_content
            if provider_fn is None:
                # try matplotlib
                if find_spec("matplotlib") and find_spec("matplotlib.figure"):
                    from matplotlib.figure import Figure as MatplotlibFigure

                    if isinstance(content, MatplotlibFigure):

                        def get_matplotlib_content(figure: MatplotlibFigure):
                            import base64
                            from io import BytesIO

                            buf = BytesIO()
                            figure.savefig(buf, format="png")
                            data = base64.b64encode(buf.getbuffer()).decode("ascii")
                            return f'<img src="data:image/png;base64,{data}"/>'

                        Gui.register_content_provider(MatplotlibFigure, get_matplotlib_content)
                        provider_fn = get_matplotlib_content

            if _is_function(provider_fn):
                try:
                    return t.cast(t.Callable, provider_fn)(t.cast(t.Any, content))
                except Exception as e:
                    _warn(f"Error in content provider for type {str(type(content))}", e)
        return (
            '<div style="background:white;color:red;">'
            + (f"No valid provider for type {type(content).__name__}" if content else "Wrong context.")
            + "</div>"
        )

    @staticmethod
    def add_shared_variable(*names: str) -> None:
        """Add shared variables.

        The variables will be synchronized between all clients when updated.
        Note that only variables from the main module will be registered.

        This is a synonym for `(Gui.)add_shared_variables()^`.

        Arguments:
            names: The names of the variables that become shared, as a list argument.
        """
        for name in names:
            if name not in Gui.__shared_variables:
                Gui.__shared_variables.append(name)

    @staticmethod
    def add_shared_variables(*names: str) -> None:
        """Add shared variables.

        The variables will be synchronized between all clients when updated.
        Note that only variables from the main module will be registered.

        This is a synonym for `(Gui.)add_shared_variable()^`.

        Arguments:
            names: The names of the variables that become shared, as a list argument.
        """
        Gui.add_shared_variable(*names)

    def _get_shared_variables(self) -> t.List[str]:
        return self.__evaluator.get_shared_variables()

    @staticmethod
    def _clear_shared_variable() -> None:
        Gui.__shared_variables.clear()

    def __get_content_accessor(self):
        if self.__content_accessor is None:
            self.__content_accessor = _ContentAccessor(self._get_config("data_url_max_size", 50 * 1024))
        return self.__content_accessor

    def _bindings(self):
        return self.__bindings

    def _get_data_scope(self) -> SimpleNamespace:
        return self.__bindings._get_data_scope()

    def _get_data_scope_metadata(self) -> t.Dict[str, t.Any]:
        return self.__bindings._get_data_scope_metadata()

    def _get_all_data_scopes(self) -> t.Dict[str, SimpleNamespace]:
        return self.__bindings._get_all_scopes()

    def _get_config(self, name: ConfigParameter, default_value: t.Any) -> t.Any:
        return self._config._get_config(name, default_value)

    def _get_themes(self) -> t.Optional[t.Dict[str, t.Any]]:
        theme = self._get_config("theme", None)
        dark_theme = self._get_config("dark_theme", None)
        light_theme = self._get_config("light_theme", None)
        res = {}
        if theme:
            res["base"] = theme
        if dark_theme:
            res["dark"] = dark_theme
        if light_theme:
            res["light"] = light_theme
        return res if theme or dark_theme or light_theme else None

    def _bind(self, name: str, value: t.Any) -> None:
        self._bindings()._bind(name, value)

    def __get_state(self):
        return self.__state

    def _get_client_id(self) -> str:
        return (
            _DataScopes._GLOBAL_ID
            if self._bindings()._is_single_client()
            else getattr(g, Gui.__ARG_CLIENT_ID, "unknown id")
        )

    def __set_client_id_in_context(self, client_id: t.Optional[str] = None, force=False):
        if not client_id and request:
            client_id = request.args.get(Gui.__ARG_CLIENT_ID, "")
        if not client_id and (ws_client_id := getattr(g, "ws_client_id", None)):
            client_id = ws_client_id
        if not client_id and force:
            res = self._bindings()._get_or_create_scope("")
            client_id = res[0] if res[1] else None
        if client_id and request:
            if sid := getattr(request, "sid", None):
                sids = self.__client_id_2_sid.get(client_id, None)
                if sids is None:
                    sids = set()
                    self.__client_id_2_sid[client_id] = sids
                sids.add(sid)
        g.client_id = client_id

    def __is_var_modified_in_context(self, var_name: str, derived_vars: t.Set[str]) -> bool:
        modified_vars: t.Optional[t.Set[str]] = getattr(g, "modified_vars", None)
        der_vars: t.Optional[t.Set[str]] = getattr(g, "derived_vars", None)
        setattr(g, "update_count", getattr(g, "update_count", 0) + 1)  # noqa: B010
        if modified_vars is None:
            modified_vars = set()
            g.modified_vars = modified_vars
        if der_vars is None:
            g.derived_vars = derived_vars
        else:
            der_vars.update(derived_vars)
        if var_name in modified_vars:
            return True
        modified_vars.add(var_name)
        return False

    def __clean_vars_on_exit(self) -> t.Optional[t.Set[str]]:
        update_count = getattr(g, "update_count", 0) - 1
        if update_count < 1:
            derived_vars: t.Set[str] = getattr(g, "derived_vars", set())
            delattr(g, "update_count")
            delattr(g, "modified_vars")
            delattr(g, "derived_vars")
            return derived_vars
        else:
            setattr(g, "update_count", update_count)  # noqa: B010
            return None

    def _handle_connect(self):
        _Hooks().handle_connect(self)

    def _handle_disconnect(self):
        _Hooks()._handle_disconnect(self)
        if (sid := getattr(request, "sid", None)) and (st_to := self._get_config("state_retention_period", 0)) > 0:
            for cl_id, sids in self.__client_id_2_sid.items():
                if sid in sids:
                    if len(sids) == 1:
                        Timer(st_to, self._remove_state, [cl_id]).start()
                    else:
                        sids.remove(sid)
                    return

    def _remove_state(self, client_id: str):
        if (sids := self.__client_id_2_sid.get(client_id, None)) and len(sids) == 1:
            try:
                del self.__client_id_2_sid[client_id]
                self._bindings()._delete_scope(client_id)
            except Exception as e:
                _warn(f"Unexpected error removing state {client_id}", e)

    def _manage_message(self, msg_type: _WsType, message: dict) -> None:
        try:
            client_id = None
            if msg_type == _WsType.CLIENT_ID.value:
                res = self._bindings()._get_or_create_scope(message.get("payload", ""))
                client_id = res[0] if res[1] else None
            expected_client_id = client_id or message.get(Gui.__ARG_CLIENT_ID)
            self.__set_client_id_in_context(expected_client_id)
            g.ws_client_id = expected_client_id
            with self._set_locals_context(message.get("module_context") or None):
                with self._get_authorization():
                    payload = message.get("payload", {})
                    if msg_type == _WsType.UPDATE.value:
                        self.__front_end_update(
                            str(message.get("name")),
                            payload.get("value"),
                            message.get("propagate", True),
                            payload.get("relvar"),
                            payload.get("on_change"),
                        )
                    elif msg_type == _WsType.ACTION.value:
                        self.__on_action(message.get("name"), message.get("payload"))
                    elif msg_type == _WsType.DATA_UPDATE.value:
                        self.__request_data_update(str(message.get("name")), message.get("payload"))
                    elif msg_type == _WsType.REQUEST_UPDATE.value:
                        self.__request_var_update(message.get("payload"))
                    elif msg_type == _WsType.GET_MODULE_CONTEXT.value:
                        self.__handle_ws_get_module_context(payload)
                    elif msg_type == _WsType.GET_DATA_TREE.value:
                        self.__handle_ws_get_data_tree()
                    elif msg_type == _WsType.APP_ID.value:
                        self.__handle_ws_app_id(message)
                    elif msg_type == _WsType.GET_ROUTES.value:
                        self.__handle_ws_get_routes()
                    else:
                        self._manage_external_message(msg_type, message)
                self.__send_ack(message.get("ack_id"))
        except Exception as e:  # pragma: no cover
            if isinstance(e, AttributeError) and (name := message.get("name")):
                try:
                    names = self._get_real_var_name(name)
                    var_name = names[0] if isinstance(names, tuple) else names
                    var_context = names[1] if isinstance(names, tuple) else None
                    if var_name.startswith("tpec_"):
                        var_name = var_name[5:]
                    if var_name.startswith("TpExPr_"):
                        var_name = var_name[7:]
                    _warn(
                        f"A problem occurred while resolving variable '{var_name}'"
                        + (f" in module '{var_context}'." if var_context else ".")
                    )
                except Exception as e1:
                    _warn(f"Resolving  name '{name}' failed", e1)
            else:
                _warn(f"Decoding Message has failed: {message}", e)

    # To be expanded by inheriting classes
    # this will be used to handle ws messages that is not handled by the base Gui class
    def _manage_external_message(self, msg_type: _WsType, message: dict) -> None:
        _Hooks()._manage_external_message(self, msg_type, message)

    def __front_end_update(
        self,
        var_name: str,
        value: t.Any,
        propagate=True,
        rel_var: t.Optional[str] = None,
        on_change: t.Optional[str] = None,
    ) -> None:
        if not var_name:
            return
        # Check if Variable is a managed type
        current_value = _getscopeattr_drill(self, self.__evaluator.get_hash_from_expr(var_name))
        if isinstance(current_value, _TaipyData):
            return
        elif rel_var and isinstance(current_value, _TaipyLovValue):  # pragma: no cover
            lov_holder = _getscopeattr_drill(self, self.__evaluator.get_hash_from_expr(rel_var))
            if isinstance(lov_holder, _TaipyLov):
                if isinstance(value, (str, list)):
                    val = value if isinstance(value, list) else [value]
                    elt_4_ids = self.__adapter._get_elt_per_ids(lov_holder.get_name(), lov_holder.get())
                    ret_val = [elt_4_ids.get(x, x) for x in val]
                    if isinstance(value, list):
                        value = ret_val
                    elif ret_val:
                        value = ret_val[0]
        elif isinstance(current_value, _TaipyBase):
            value = current_value.cast_value(value)
        self._update_var(
            var_name, value, propagate, current_value if isinstance(current_value, _TaipyBase) else None, on_change
        )

    def _update_var(
        self,
        var_name: str,
        value: t.Any,
        propagate=True,
        holder: t.Optional[_TaipyBase] = None,
        on_change: t.Optional[str] = None,
        forward: t.Optional[bool] = True,
    ) -> None:
        if holder:
            var_name = holder.get_name()
        hash_expr = self.__evaluator.get_hash_from_expr(var_name)
        derived_vars = {hash_expr}
        # set to broadcast mode if hash_expr is in shared_variable
        if hash_expr in self._get_shared_variables():
            self._set_broadcast()
        # Use custom attrsetter function to allow value binding for _MapDict
        if propagate:
            _setscopeattr_drill(self, hash_expr, value)
            # In case expression == hash (which is when there is only a single variable in expression)
            if var_name == hash_expr or hash_expr.startswith("tpec_"):
                derived_vars.update(self._re_evaluate_expr(var_name))
        elif holder:
            derived_vars.update(self._evaluate_holders(hash_expr))
        if forward:
            # if the variable has been evaluated then skip updating to prevent infinite loop
            var_modified = self.__is_var_modified_in_context(hash_expr, derived_vars)
            if not var_modified:
                self._call_on_change(
                    var_name,
                    value.get()
                    if isinstance(value, _TaipyBase)
                    else value._dict
                    if isinstance(value, _MapDict)
                    else value,
                    on_change,
                )
            derived_modified = self.__clean_vars_on_exit()
            if derived_modified is not None:
                self.__send_var_list_update(list(derived_modified), var_name)

    def _get_real_var_name(self, var_name: str) -> t.Tuple[str, str]:
        if not var_name:
            return (var_name, var_name)
        # Handle holder prefix if needed
        if var_name.startswith(_TaipyBase._HOLDER_PREFIX):
            for hp in _TaipyBase._get_holder_prefixes():
                if var_name.startswith(hp):
                    var_name = var_name[len(hp) :]
                    break
        suffix_var_name = ""
        if "." in var_name:
            first_dot_index = var_name.index(".")
            suffix_var_name = var_name[first_dot_index + 1 :]
            var_name = var_name[:first_dot_index]
        var_name_decode, module_name = _variable_decode(self._get_expr_from_hash(var_name))
        current_context = self._get_locals_context()
        # #583: allow module resolution for var_name in current_context root_page context
        if (
            module_name
            and self._config.root_page
            and self._config.root_page._renderer
            and self._config.root_page._renderer._get_module_name() == module_name
        ):
            return f"{var_name_decode}.{suffix_var_name}" if suffix_var_name else var_name_decode, module_name
        if module_name == current_context:
            var_name = var_name_decode
        # only strict checking for cross-context linked variable when the context has been properly set
        elif self._has_set_context():
            if var_name not in self.__var_dir._var_head:
                raise NameError(f"Can't find matching variable for {var_name} on context: {current_context}")
            _found = False
            for k, v in self.__var_dir._var_head[var_name]:
                if v == current_context:
                    var_name = k
                    _found = True
                    break
            if not _found:  # pragma: no cover
                raise NameError(f"Can't find matching variable for {var_name} on context: {current_context}")
        return f"{var_name}.{suffix_var_name}" if suffix_var_name else var_name, current_context

    def _call_on_change(self, var_name: str, value: t.Any, on_change: t.Optional[str] = None):
        try:
            var_name, current_context = self._get_real_var_name(var_name)
        except Exception as e:  # pragma: no cover
            _warn("", e)
            return
        on_change_fn = self._get_user_function(on_change) if on_change else None
        if not _is_function(on_change_fn):
            on_change_fn = self._get_user_function("on_change")
        if _is_function(on_change_fn):
            try:
                self._call_function_with_state(t.cast(t.Callable, on_change_fn), [var_name, value, current_context])
            except Exception as e:  # pragma: no cover
                if not self._call_on_exception(on_change or "on_change", e):
                    _warn(f"{on_change or 'on_change'}(): callback function raised an exception", e)

    def _get_content(self, var_name: str, value: t.Any, image: bool) -> t.Any:
        ret_value = self.__get_content_accessor().get_info(var_name, value, image)
        return f"/{Gui.__CONTENT_ROOT}/{ret_value[0]}" if isinstance(ret_value, tuple) else ret_value

    def __serve_content(self, path: str) -> t.Any:
        self.__set_client_id_in_context()
        parts = path.split("/")
        if len(parts) > 1:
            file_name = parts[-1]
            (dir_path, as_attachment) = self.__get_content_accessor().get_content_path(
                path[: -len(file_name) - 1], file_name, request.args.get("bypass")
            )
            if dir_path:
                return send_from_directory(str(dir_path), file_name, as_attachment=as_attachment)
        return ("", 404)

    def _get_user_content_url(
        self, path: t.Optional[str] = None, query_args: t.Optional[t.Dict[str, str]] = None
    ) -> t.Optional[str]:
        q_args = query_args or {}
        q_args.update({Gui.__ARG_CLIENT_ID: self._get_client_id()})
        return f"/{Gui.__USER_CONTENT_URL}/{path or 'TaIpY'}?{urlencode(q_args)}"

    def __serve_user_content(self, path: str) -> t.Any:
        self.__set_client_id_in_context()
        q_args: t.Dict[str, str] = {}
        q_args.update(request.args)
        q_args.pop(Gui.__ARG_CLIENT_ID, None)
        cb_function: t.Optional[t.Union[t.Callable, str]] = None
        cb_function_name = None
        if q_args.get(Gui._HTML_CONTENT_KEY):
            cb_function = self.__process_content_provider
            cb_function_name = cb_function.__name__
        else:
            cb_function_name = q_args.get(Gui.__USER_CONTENT_CB)
            if cb_function_name:
                cb_function = self._get_user_function(cb_function_name)
                if not _is_function(cb_function):
                    parts = cb_function_name.split(".", 1)
                    if len(parts) > 1:
                        base = _getscopeattr(self, parts[0], None)
                        if base and (meth := getattr(base, parts[1], None)):
                            cb_function = meth
                        else:
                            base = self.__evaluator._get_instance_in_context(parts[0])
                            if base and (meth := getattr(base, parts[1], None)):
                                cb_function = meth
                if not _is_function(cb_function):
                    _warn(f"{cb_function_name}() callback function has not been defined.")
                    cb_function = None
        if cb_function is None:
            cb_function_name = "on_user_content"
            if hasattr(self, cb_function_name) and _is_function(self.on_user_content):
                cb_function = self.on_user_content
            else:
                _warn("on_user_content() callback function has not been defined.")
        if _is_function(cb_function):
            try:
                args: t.List[t.Any] = []
                if path:
                    args.append(path)
                if len(q_args):
                    args.append(q_args)
                ret = self._call_function_with_state(t.cast(t.Callable, cb_function), args)
                if ret is None:
                    _warn(f"{cb_function_name}() callback function must return a value.")
                else:
                    return (ret, 200)
            except Exception as e:  # pragma: no cover
                if not self._call_on_exception(cb_function_name, e):
                    _warn(f"{cb_function_name}() callback function raised an exception", e)
        return ("", 404)

    def __serve_extension(self, path: str) -> t.Any:
        parts = path.split("/")
        last_error = ""
        resource_name = None
        if len(parts) > 1:
            libs = Gui.__extensions.get(parts[0], [])
            for library in libs:
                try:
                    resource_name = library.get_resource("/".join(parts[1:]))
                    if resource_name:
                        return send_file(resource_name)
                except Exception as e:
                    last_error = f"\n{e}"  # Check if the resource is served by another library with the same name
        _warn(f"Resource '{resource_name or path}' not accessible for library '{parts[0]}'{last_error}")
        return ("", 404)

    def __get_version(self) -> str:
        return f'{self.__version.get("major", 0)}.{self.__version.get("minor", 0)}.{self.__version.get("patch", 0)}'

    def __append_libraries_to_status(self, status: t.Dict[str, t.Any]):
        libraries: t.Dict[str, t.Any] = {}
        for libs_list in self.__extensions.values():
            for lib in libs_list:
                if not isinstance(lib, ElementLibrary):
                    continue
                libs = libraries.get(lib.get_name())
                if libs is None:
                    libs = []
                    libraries[lib.get_name()] = libs
                elements: t.List[t.Dict[str, str]] = []
                libs.append({"js module": lib.get_js_module_name(), "elements": elements})
                for element_name, elt in lib.get_elements().items():
                    if not isinstance(elt, Element):
                        continue
                    elt_dict = {"name": element_name}
                    if hasattr(elt, "_render_xhtml"):
                        elt_dict["render function"] = elt._render_xhtml.__code__.co_name
                    else:
                        elt_dict["react name"] = elt._get_js_name(element_name)
                    elements.append(elt_dict)
        status.update({"libraries": libraries})

    def _serve_status(self, template: Path) -> t.Dict[str, t.Dict[str, str]]:
        base_json: t.Dict[str, t.Any] = {"user_status": str(self.__call_on_status() or "")}
        if self._get_config("extended_status", False):
            base_json.update(
                {
                    "flask_version": str(metadata.version("flask") or ""),
                    "backend_version": self.__get_version(),
                    "host": f'{self._get_config("host", "localhost")}:{self._get_config("port", "default")}',
                    "python_version": sys.version,
                }
            )
            self.__append_libraries_to_status(base_json)
            try:
                base_json.update(json.loads(template.read_text()))
            except Exception as e:  # pragma: no cover
                _warn(f"Exception raised reading JSON in '{template}'", e)
        return {"gui": base_json}

    def __upload_files(self):
        self.__set_client_id_in_context()
        on_upload_action = request.form.get("on_action", None)
        var_name = t.cast(str, request.form.get("var_name", None))
        if not var_name and not on_upload_action:
            _warn("upload files: No var name")
            return ("upload files: No var name", 400)
        context = request.form.get("context", None)
        upload_data = request.form.get("upload_data", None)
        multiple = "multiple" in request.form and request.form["multiple"] == "True"

        # File parsing and checks
        file = request.files.get("blob", None)
        if not file:
            _warn("upload files: No file part")
            return ("upload files: No file part", 400)
        # If the user does not select a file, the browser submits an
        # empty file without a filename.
        if file.filename == "":
            _warn("upload files: No selected file")
            return ("upload files: No selected file", 400)

        # Path parsing and checks
        path = request.form.get("path", "")
        suffix = ""
        complete = True
        part = 0

        if "total" in request.form:
            total = int(request.form["total"])
            if total > 1 and "part" in request.form:
                part = int(request.form["part"])
                suffix = f".part.{part}"
                complete = part == total - 1

        # Extract upload path (when single file is selected, path="" does not change the path)
        upload_root = os.path.abspath(self._get_config("upload_folder", tempfile.gettempdir()))
        upload_path = os.path.abspath(os.path.join(upload_root, os.path.dirname(path)))
        if upload_path.startswith(upload_root):
            upload_path = Path(upload_path).resolve()
            os.makedirs(upload_path, exist_ok=True)
            # Save file into upload_path directory
            file_path = _get_non_existent_file_path(upload_path, secure_filename(file.filename))
            file.save(os.path.join(upload_path, (file_path.name + suffix)))
        else:
            _warn(f"upload files: Path {path} points outside of upload root.")
            return ("upload files: Path part points outside of upload root.", 400)

        if complete:
            if part > 0:
                try:
                    with open(file_path, "wb") as grouped_file:
                        for nb in range(part + 1):
                            part_file_path = upload_path / f"{file_path.name}.part.{nb}"
                            with open(part_file_path, "rb") as part_file:
                                grouped_file.write(part_file.read())
                            # remove file_path after it is merged
                            part_file_path.unlink()
                except EnvironmentError as ee:  # pragma: no cover
                    _warn(f"Cannot group file after chunk upload for {file.filename}", ee)
                    return (f"Cannot group file after chunk upload for {file.filename}", 500)
            # notify the file is uploaded
            newvalue = str(file_path)
            if multiple and var_name:
                value = _getscopeattr(self, var_name)
                if not isinstance(value, t.List):
                    value = [] if value is None else [value]
                value.append(newvalue)
                newvalue = value
            with self._set_locals_context(context):
                if on_upload_action:
                    data = {}
                    if upload_data:
                        try:
                            data = json.loads(upload_data)
                        except Exception:
                            pass
                    data["path"] = file_path
                    file_fn = self._get_user_function(on_upload_action)
                    if not _is_function(file_fn):
                        file_fn = _getscopeattr(self, on_upload_action)
                    if _is_function(file_fn):
                        self._call_function_with_state(t.cast(t.Callable, file_fn), ["file_upload", {"args": [data]}])
                else:
                    setattr(self._bindings(), var_name, newvalue)
        return ("", 200)

    def __send_var_list_update(  # noqa C901
        self,
        modified_vars: t.List[str],
        front_var: t.Optional[str] = None,
    ):
        ws_dict = {}
        is_custom_page = is_in_custom_page_context()
        values = {v: _getscopeattr_drill(self, v) for v in modified_vars if is_custom_page or _is_moduled_variable(v)}
        if not values:
            return
        for k, v in values.items():
            if isinstance(v, (_TaipyData, _TaipyContentHtml)) and v.get_name() in modified_vars:
                modified_vars.remove(v.get_name())
            elif isinstance(v, _DoNotUpdate):
                modified_vars.remove(k)
        for _var in modified_vars:
            newvalue = values.get(_var)
            resource_handler = get_current_resource_handler()
            custom_page_filtered_types = resource_handler.data_layer_supported_types if resource_handler else ()
            if isinstance(newvalue, (_TaipyData)) or isinstance(newvalue, custom_page_filtered_types):
                newvalue = {"__taipy_refresh": True}
            else:
                if isinstance(newvalue, (_TaipyContent, _TaipyContentImage)):
                    ret_value = self.__get_content_accessor().get_info(
                        t.cast(str, front_var), newvalue.get(), isinstance(newvalue, _TaipyContentImage)
                    )
                    if isinstance(ret_value, tuple):
                        newvalue = f"/{Gui.__CONTENT_ROOT}/{ret_value[0]}"
                    else:
                        newvalue = ret_value
                elif isinstance(newvalue, _TaipyContentHtml):
                    newvalue = self._get_user_content_url(
                        None, {"variable_name": str(_var), Gui._HTML_CONTENT_KEY: str(time.time())}
                    )
                elif isinstance(newvalue, (_TaipyLov, _TaipyLovValue)):
                    newvalue = self.__adapter.run(
                        newvalue.get_name(), newvalue.get(), id_only=isinstance(newvalue, _TaipyLovValue)
                    )
                elif isinstance(newvalue, _TaipyBase):
                    newvalue = newvalue.get()
                # Skip in taipy-gui, available in custom frontend
                if isinstance(newvalue, (dict, _MapDict)) and not is_in_custom_page_context():
                    continue
                if isinstance(newvalue, float) and math.isnan(newvalue):
                    # do not let NaN go through json, it is not handle well (dies silently through websocket)
                    newvalue = None
                if newvalue is not None and not isinstance(newvalue, str):
                    debug_warnings: t.List[warnings.WarningMessage] = []
                    with warnings.catch_warnings(record=True) as warns:
                        warnings.resetwarnings()
                        json.dumps(newvalue, cls=_TaipyJsonEncoder)
                        if len(warns):
                            keep_value = True
                            for w in warns:
                                if is_debugging():
                                    debug_warnings.append(w)
                                if w.category is not DeprecationWarning and w.category is not PendingDeprecationWarning:
                                    keep_value = False
                                    break
                            if not keep_value:
                                # do not send data that is not serializable
                                continue
                    for w in debug_warnings:
                        warnings.warn(w.message, w.category)  # noqa: B028
            ws_dict[_var] = newvalue
        # TODO: What if value == newvalue?
        self.__send_ws_update_with_dict(ws_dict)

    def __update_state_context(self, payload: dict):
        # apply state context if any
        state_context = payload.get("state_context")
        if isinstance(state_context, dict):
            for var, val in state_context.items():
                self._update_var(var, val, True, forward=False)

    @staticmethod
    def set_unsupported_data_converter(converter: t.Optional[t.Callable[[t.Any], t.Any]]) -> None:
        """Set a custom converter for unsupported data types.

        This function allows specifying a custom conversion function for data types that cannot
        be serialized automatically. When Taipy GUI encounters an unsupported type in the data
        being serialized, it will invoke the provided *converter* function. The returned value
        from this function will then be serialized if possible.

        Arguments:
            converter: A function that converts a value with an unsupported data type (the only
                parameter to the function) into data with a supported data type (the returned value
                from the function).</br>
                If set to `None`, it removes any existing converter.
        """
        Gui.__unsupported_data_converter = converter

    @staticmethod
    def _convert_unsupported_data(value: t.Any) -> t.Optional[t.Any]:
        """
        Handles unsupported data by invoking the converter if there is one.

        Arguments:
            value: The unsupported data encountered.

        Returns:
            The transformed data or None if no transformation is possible.
        """
        try:
            return Gui.__unsupported_data_converter(value) if _is_function(Gui.__unsupported_data_converter) else None  # type: ignore
        except Exception as e:
            _warn(f"Error transforming data: {str(e)}")
            return None

    def __request_data_update(self, var_name: str, payload: t.Any) -> None:
        # Use custom attrgetter function to allow value binding for _MapDict
        newvalue = _getscopeattr_drill(self, var_name)
        resource_handler = get_current_resource_handler()
        custom_page_filtered_types = resource_handler.data_layer_supported_types if resource_handler else ()
        if not isinstance(newvalue, _TaipyData) and isinstance(newvalue, custom_page_filtered_types):
            newvalue = _TaipyData(newvalue, "")
        if isinstance(newvalue, _TaipyData):
            ret_payload = None
            if isinstance(payload, dict):
                self.__update_state_context(payload)
                lib_name = payload.get("library")
                if isinstance(lib_name, str):
                    libs = self.__extensions.get(lib_name, [])
                    for lib in libs:
                        user_var_name = var_name
                        try:
                            with contextlib.suppress(NameError):
                                # ignore name error and keep var_name
                                user_var_name = self._get_real_var_name(var_name)[0]
                            ret_payload = lib.get_data(lib_name, payload, user_var_name, newvalue)
                            if ret_payload:
                                break
                        except Exception as e:  # pragma: no cover
                            _warn(
                                f"Exception raised in '{lib_name}.get_data({lib_name}, payload, {user_var_name}, value)'",  # noqa: E501
                                e,
                            )
            if not isinstance(ret_payload, dict):
                ret_payload = self._get_accessor().get_data(var_name, newvalue, payload)
            self.__send_ws_update_with_dict({var_name: ret_payload})

    def __request_var_update(self, payload: t.Any):
        if isinstance(payload, dict) and isinstance(payload.get("names"), list):
            self.__update_state_context(payload)
            if payload.get("refresh", False):
                # refresh vars
                for _var in t.cast(list, payload.get("names")):
                    val = _getscopeattr_drill(self, _var)
                    self._refresh_expr(
                        val.get_name() if isinstance(val, _TaipyBase) else _var,
                        val if isinstance(val, _TaipyBase) else None,
                    )
            self.__send_var_list_update(payload["names"])

    def __handle_ws_get_module_context(self, payload: t.Any):
        if isinstance(payload, dict):
            page_path = str(payload.get("path"))
            if page_path in {"/", ""}:
                page_path = Gui.__root_page_name
            # Get Module Context
            if mc := self._get_page_context(page_path):
                page_renderer = t.cast(_Page, self._get_page(page_path))._renderer
                self._bind_custom_page_variables(t.cast(t.Any, page_renderer), self._get_client_id())
                # get metadata if there is one
                metadata: t.Dict[str, t.Any] = {}
                if hasattr(page_renderer, "_metadata"):
                    metadata = getattr(page_renderer, "_metadata", {})
                meta_return = json.dumps(metadata, cls=_TaipyJsonEncoder) if metadata else None
                self.__send_ws(
                    {
                        "type": _WsType.GET_MODULE_CONTEXT.value,
                        "payload": {"context": mc, "metadata": meta_return},
                    },
                    send_back_only=True,
                )

    def __get_variable_tree(self, data: t.Dict[str, t.Any]):
        # Module Context -> Variable -> Variable data (name, type, initial_value)
        variable_tree: t.Dict[str, t.Dict[str, t.Dict[str, t.Any]]] = {}
        # Types of data to be handled by the data layer and filtered out here
        resource_handler = get_current_resource_handler()
        filtered_value_types = resource_handler.data_layer_supported_types if resource_handler else ()
        for k, v in data.items():
            if isinstance(v, _TaipyBase):
                data[k] = v.get()
            var_name, var_module_name = _variable_decode(k)
            if var_module_name == "" or var_module_name is None:
                var_module_name = "__main__"
            if var_module_name not in variable_tree:
                variable_tree[var_module_name] = {}
            data_update = isinstance(v, filtered_value_types)
            value = None if data_update else data[k]
            # if _is_moduled_variable(k):
            variable_tree[var_module_name][var_name] = {
                "type": type(v).__name__,
                "value": value,
                "encoded_name": k,
                "data_update": data_update,
            }
        return variable_tree

    def __handle_ws_get_data_tree(self):
        # Get Variables
        self.__pre_render_pages()
        data = {
            k: v
            for k, v in vars(self._get_data_scope()).items()
            if not k.startswith("_")
            and not callable(v)
            and "TpExPr" not in k
            and not isinstance(v, (ModuleType, FunctionType, LambdaType, type, Page))
        }
        function_data = {
            k: v
            for k, v in vars(self._get_data_scope()).items()
            if not k.startswith("_") and "TpExPr" not in k and isinstance(v, (FunctionType, LambdaType))
        }
        self.__send_ws(
            {
                "type": _WsType.GET_DATA_TREE.value,
                "payload": {
                    "variable": self.__get_variable_tree(data),
                    "function": self.__get_variable_tree(function_data),
                },
            },
            send_back_only=True,
        )

    def __handle_ws_app_id(self, message: t.Any):
        if not isinstance(message, dict):
            return
        name = message.get("name", "")
        payload = message.get("payload", "")
        app_id = id(self)
        if payload == app_id:
            return
        self.__send_ws(
            {
                "type": _WsType.APP_ID.value,
                "payload": {"name": name, "id": app_id},
            },
            send_back_only=True,
        )

    def __handle_ws_get_routes(self):
        routes = (
            [[self._config.root_page._route, t.cast(t.Any, self._config.root_page._renderer).page_type]]
            if self._config.root_page
            else []
        )
        routes += [
            [page._route, t.cast(t.Any, page._renderer).page_type]
            for page in self._config.pages
            if page._route != Gui.__root_page_name
        ]
        self.__send_ws(
            {
                "type": _WsType.GET_ROUTES.value,
                "payload": routes,
            },
            send_back_only=True,
        )

    def __send_ws(self, payload: dict, allow_grouping=True, send_back_only=False) -> None:
        grouping_message = self.__get_message_grouping() if allow_grouping else None
        if grouping_message is None:
            try:
                self._server._ws.emit(
                    "message",
                    payload,
                    to=t.cast(str, self.__get_ws_receiver(send_back_only)),
                )
                time.sleep(0.001)
            except Exception as e:  # pragma: no cover
                _warn(f"Exception raised in WebSocket communication in '{self.__frame.f_code.co_name}'", e)
        else:
            grouping_message.append(payload)

    def __broadcast_ws(self, payload: dict, client_id: t.Optional[str] = None):
        try:
            to = list(self.__get_sids(client_id)) if client_id else []
            self._server._ws.emit("message", payload, to=t.cast(str, to) if to else None, include_self=True)
            time.sleep(0.001)
        except Exception as e:  # pragma: no cover
            _warn(f"Exception raised in WebSocket communication in '{self.__frame.f_code.co_name}'", e)

    def __send_ack(self, ack_id: t.Optional[str]) -> None:
        if ack_id:
            try:
                self._server._ws.emit(
                    "message",
                    {"type": _WsType.ACKNOWLEDGEMENT.value, "id": ack_id},
                    to=t.cast(str, self.__get_ws_receiver(True)),
                )
                time.sleep(0.001)
            except Exception as e:  # pragma: no cover
                _warn(f"Exception raised in WebSocket communication (send ack) in '{self.__frame.f_code.co_name}'", e)

    def _send_ws_id(self, id: str) -> None:
        self.__send_ws(
            {
                "type": _WsType.CLIENT_ID.value,
                "id": id,
            },
            allow_grouping=False,
        )

    def __send_ws_download(self, content: str, name: str, on_action: str) -> None:
        self.__send_ws(
            {"type": _WsType.DOWNLOAD_FILE.value, "content": content, "name": name, "onAction": on_action},
            send_back_only=True,
        )

    def __send_ws_alert(
        self, type: str, message: str, system_notification: bool, duration: int, notification_id: t.Optional[str] = None
    ) -> None:
        payload = {
            "type": _WsType.ALERT.value,
            "atype": type,
            "message": message,
            "system": system_notification,
            "duration": duration,
        }

        if notification_id:
            payload["notificationId"] = notification_id

        self.__send_ws(
            payload,
        )

    def __send_ws_partial(self, partial: str):
        self.__send_ws(
            {
                "type": _WsType.PARTIAL.value,
                "name": partial,
            }
        )

    def __send_ws_block(
        self,
        action: t.Optional[str] = None,
        message: t.Optional[str] = None,
        close: t.Optional[bool] = False,
        cancel: t.Optional[bool] = False,
    ):
        self.__send_ws(
            {
                "type": _WsType.BLOCK.value,
                "action": action,
                "close": close,
                "message": message,
                "noCancel": not cancel,
            }
        )

    def __send_ws_navigate(
        self,
        to: str,
        params: t.Optional[t.Dict[str, str]],
        tab: t.Optional[str],
        force: bool,
    ):
        self.__send_ws({"type": _WsType.NAVIGATE.value, "to": to, "params": params, "tab": tab, "force": force})

    def __send_ws_update_with_dict(self, modified_values: dict) -> None:
        payload = [
            {"name": _get_client_var_name(k), "payload": v if isinstance(v, dict) and "value" in v else {"value": v}}
            for k, v in modified_values.items()
        ]
        if self._is_broadcasting():
            self.__broadcast_ws({"type": _WsType.MULTIPLE_UPDATE.value, "payload": payload})
            self._set_broadcast(False)
        else:
            self.__send_ws({"type": _WsType.MULTIPLE_UPDATE.value, "payload": payload})

    def __send_ws_broadcast(
        self,
        var_name: str,
        var_value: t.Any,
        client_id: t.Optional[str] = None,
        message_type: t.Optional[_WsType] = None,
    ):
        self.__broadcast_ws(
            {
                "type": _WsType.BROADCAST.value if message_type is None else message_type.value,
                "name": _get_broadcast_var_name(var_name),
                "payload": {"value": var_value},
            },
            client_id,
        )

    def __get_ws_receiver(self, send_back_only=False) -> t.Union[t.List[str], t.Any, None]:
        if self._bindings()._is_single_client():
            return None
        sid = getattr(request, "sid", None) if request else None
        sids = self.__get_sids(self._get_client_id())
        if sid:
            sids.add(sid)
            if send_back_only:
                return sid
        return list(sids)

    def __get_sids(self, client_id: str) -> t.Set[str]:
        return self.__client_id_2_sid.get(client_id, set())

    def __get_message_grouping(self):
        return (
            _getscopeattr(self, Gui.__MESSAGE_GROUPING_NAME)
            if _hasscopeattr(self, Gui.__MESSAGE_GROUPING_NAME)
            else None
        )

    def __enter__(self):
        self.__hold_messages()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            self.__send_messages()
        except Exception as e:  # pragma: no cover
            _warn("Exception raised while sending messages", e)
        if exc_value:  # pragma: no cover
            _warn(f"An {exc_type or 'Exception'} was raised", exc_value)
        return True

    def __hold_messages(self):
        grouping_message = self.__get_message_grouping()
        if grouping_message is None:
            self._bind_var_val(Gui.__MESSAGE_GROUPING_NAME, [])

    def __send_messages(self):
        grouping_message = self.__get_message_grouping()
        if grouping_message is not None:
            _delscopeattr(self, Gui.__MESSAGE_GROUPING_NAME)
            if len(grouping_message):
                self.__send_ws({"type": _WsType.MULTIPLE_MESSAGE.value, "payload": grouping_message})

    def _get_user_function(self, func_name: str) -> t.Union[t.Callable, str]:
        func = (
            getattr(self, func_name.split(".", 2)[1], func_name) if func_name.startswith(f"{Gui.__SELF_VAR}.") else None
        )
        if not _is_function(func):
            func = _getscopeattr(self, func_name, None)
        if not _is_function(func):
            func = self._get_locals_bind().get(func_name)
        if not _is_function(func):
            func = self.__locals_context.get_default().get(func_name)
        return t.cast(t.Callable, func) if _is_function(func) else func_name

    def _get_user_instance(self, class_name: str, class_type: type) -> t.Union[object, str]:
        cls = _getscopeattr(self, class_name, None)
        if not isinstance(cls, class_type):
            cls = self._get_locals_bind().get(class_name)
        if not isinstance(cls, class_type):
            cls = self.__locals_context.get_default().get(class_name)
        return cls if isinstance(cls, class_type) else class_name

    def __download_csv(self, state: State, var_name: str, payload: dict):
        holder_name = t.cast(str, payload.get("var_name"))
        try:
            csv_path = self._get_accessor().to_csv(
                holder_name,
                _getscopeattr(self, holder_name, None),
            )
            if csv_path:
                self._download(csv_path, "data.csv", Gui.__DOWNLOAD_DELETE_ACTION)
        except Exception as e:  # pragma: no cover
            if not self._call_on_exception("download_csv", e):
                _warn("download_csv(): Exception raised", e)

    def __delete_csv(self, state: State, var_name: str, payload: dict):
        try:
            (Path(tempfile.gettempdir()) / t.cast(str, payload.get("args", [])[-1]).split("/")[-1]).unlink(True)
        except Exception:
            pass

    def __on_action(self, id: t.Optional[str], payload: t.Any) -> None:
        if isinstance(payload, dict):
            action = payload.get("action")
        else:
            action = str(payload)
            payload = {"action": action}
        if action:
            action_fn: t.Union[t.Callable, str]
            if Gui.__DOWNLOAD_ACTION == action:
                action_fn = self.__download_csv
                payload["var_name"] = id
            elif Gui.__DOWNLOAD_DELETE_ACTION == action:
                action_fn = self.__delete_csv
            else:
                action_fn = self._get_user_function(action)
            if self.__call_function_with_args(action_function=action_fn, id=id, payload=payload):
                return
            else:  # pragma: no cover
                _warn(f"on_action(): '{action}' is not a valid function.")
        if getattr(self, "on_action", None) is not None:
            self.__call_function_with_args(action_function=self.on_action, id=id, payload=payload)

    def __call_function_with_args(self, **kwargs):
        action_function = kwargs.get("action_function")
        id = t.cast(str, kwargs.get("id"))
        payload = kwargs.get("payload")

        if _is_function(action_function):
            try:
                try:
                    args = [self._get_real_var_name(id)[0], payload]
                except Exception:
                    args = [id, payload]
                self._call_function_with_state(t.cast(t.Callable, action_function), args)
                return True
            except Exception as e:  # pragma: no cover
                if not self._call_on_exception(action_function, e):
                    _warn(f"on_action(): Exception raised in '{_function_name(action_function)}()'", e)
        return False

    def _call_function_with_state(self, user_function: t.Callable, args: t.Optional[t.List[t.Any]] = None) -> t.Any:
        cp_args = [] if args is None else args.copy()
        cp_args.insert(
            0,
            _AsyncState(t.cast(_GuiState, self.__get_state()))
            if iscoroutinefunction(user_function)
            else self.__get_state(),
        )
        argcount = user_function.__code__.co_argcount
        if argcount > 0 and ismethod(user_function):
            argcount -= 1
        if argcount > len(cp_args):
            cp_args += (argcount - len(cp_args)) * [None]
        else:
            cp_args = cp_args[:argcount]
        with self.__event_manager:
            if iscoroutinefunction(user_function):
                return _invoke_async_callback(user_function, cp_args)
            else:
                return user_function(*cp_args)

    def _set_module_context(self, module_context: t.Optional[str]) -> t.ContextManager[None]:
        return self._set_locals_context(module_context) if module_context is not None else contextlib.nullcontext()

    def invoke_callback(
        self,
        state_id: str,
        callback: t.Union[str, t.Callable],
        args: t.Optional[t.Sequence[t.Any]] = None,
        module_context: t.Optional[str] = None,
    ) -> t.Any:
        """Invoke a user callback for a given state.

        See the [section on Long Running Callbacks in a Thread](../../../../../userman/gui/callbacks.md#long-running-callbacks-in-a-thread)
        in the User Manual for details on when and how this function can be used.

        Arguments:
            state_id: The identifier of the state to use, as returned by `get_state_id()^`.
            callback (Union[str, Callable[[State^, ...], None]]): The user-defined function that is invoked.<br/>
                The first parameter of this function **must** be a `State^`.
            args (Optional[Sequence]): The remaining arguments, as a List or a Tuple.
            module_context (Optional[str]): The name of the module that will be used.
        """  # noqa: E501
        this_sid = None
        if request:
            # avoid messing with the client_id => Set(ws id)
            this_sid = getattr(request, "sid", None)
            request.sid = None  # type: ignore[attr-defined]
        try:
            with self.get_flask_app().app_context():
                setattr(g, Gui.__ARG_CLIENT_ID, state_id)
                with self._set_module_context(module_context):
                    if not _is_function(callback):
                        callback = self._get_user_function(t.cast(str, callback))
                    if not _is_function(callback):
                        _warn(f"invoke_callback(): {callback} is not callable.")
                        return None
                    return self._call_function_with_state(t.cast(t.Callable, callback), list(args) if args else None)
        except Exception as e:  # pragma: no cover
            if not self._call_on_exception(callback, e):
                _warn(
                    f"Gui.invoke_callback(): Exception raised in {_function_name(callback)}",
                    e,
                )
        finally:
            if this_sid:
                request.sid = this_sid  # type: ignore[attr-defined]
        return None

    def broadcast_callback(
        self,
        callback: t.Callable,
        args: t.Optional[t.Sequence[t.Any]] = None,
        module_context: t.Optional[str] = None,
    ) -> t.Dict[str, t.Any]:
        """Invoke a callback for every client.

        This callback gets invoked for every client connected to the application with the appropriate
        `State^` instance. You can then perform client-specific tasks, such as updating the state
        variable reflected in the user interface.

        Arguments:
            callback: The user-defined function to be invoked.<br/>
                The first parameter of this function must be a `State^` object representing the
                client for which it is invoked.<br/>
                The other parameters should reflect the ones provided in the *args* collection.
            args: The parameters to send to *callback*, if any.
        """
        # Iterate over all the scopes
        res = {}
        for id in [id for id in self.__bindings._get_all_scopes() if id != _DataScopes._GLOBAL_ID]:
            ret = self.invoke_callback(id, callback, args, module_context)
            res[id] = ret
        return res

    def broadcast_change(self, var_name: str, value: t.Any):
        """Propagates a new value for a given variable to all states.

        This callback gets invoked for every client connected to the application to update the value
        of the variable called *var_name* to the new value *value*, in their specific `State^`
        instance. All user interfaces reflect the change.

        Arguments:
            var_name: The name of the variable to change.
            value: The new value for the variable.
        """
        self.broadcast_callback(lambda s, n, v: s.assign(n, v), [var_name, value])

    @staticmethod
    def __broadcast_changes_fn(state: State, values: dict[str, t.Any]) -> None:
        with state:
            for n, v in values.items():
                state.assign(n, v)

    def broadcast_changes(self, values: t.Optional[dict[str, t.Any]] = None, **kwargs):
        """Propagates new values for several variables to all states.

        This callback gets invoked for every client connected to the application to update the value
        of all the variables that are keys in *values*, in their specific `State^` instance. All
        user interfaces reflect the change.

        Arguments:
            values: An optional dictionary where each key is the name of a variable to change, and
                where the associated value is the new value to set for that variable, in each state
                for the application.
            **kwargs (dict[str, any]): A collection of variable name-value pairs that are updated
                for each state of the application. Name-value pairs overload the ones in *values*
                if the name exists as a key in the dictionary.
        """
        if kwargs:
            values = values.copy() if values else {}
            for n, v in kwargs.items():
                values[n] = v
        self.broadcast_callback(Gui.__broadcast_changes_fn, [values])

    def _is_in_brdcst_callback(self):
        try:
            return getattr(g, Gui.__BRDCST_CALLBACK_G_ID, False)
        except RuntimeError:
            return False

    # Proxy methods for Evaluator
    def _evaluate_expr(
        self, expr: str, lazy_declare: t.Optional[bool] = False, lambda_expr: t.Optional[bool] = False
    ) -> t.Any:
        return self.__evaluator.evaluate_expr(self, expr, lazy_declare, lambda_expr)

    def _re_evaluate_expr(self, var_name: str) -> t.Set[str]:
        return self.__evaluator.re_evaluate_expr(self, var_name)

    def _refresh_expr(self, var_name: str, holder: t.Optional[_TaipyBase]):
        return self.__evaluator.refresh_expr(self, var_name, holder)

    def _get_expr_from_hash(self, hash_val: str) -> str:
        return self.__evaluator.get_expr_from_hash(hash_val)

    def _evaluate_bind_holder(self, holder: t.Type[_TaipyBase], expr: str) -> str:
        return self.__evaluator.evaluate_bind_holder(self, holder, expr)

    def _evaluate_holders(self, expr: str) -> t.List[str]:
        return self.__evaluator.evaluate_holders(self, expr)

    def _is_expression(self, expr: str) -> bool:
        if self.__evaluator is None:
            return False
        return self.__evaluator._is_expression(expr)

    # make components resettable
    def _set_building(self, building: bool):
        self._building = building

    def __is_building(self):
        return hasattr(self, "_building") and self._building

    def _get_call_method_name(self, name: str):
        return f"{Gui.__SELF_VAR}.{name}"

    def __get_attributes(self, attr_json: str, hash_json: str, args_dict: t.Dict[str, t.Any]):
        attributes: t.Dict[str, t.Any] = json.loads(unquote(attr_json))
        hashes: t.Dict[str, str] = json.loads(unquote(hash_json))
        attributes.update({k: args_dict.get(v) for k, v in hashes.items()})
        return attributes, hashes

    def _compare_data(self, *data):
        return data[0]

    def _get_adapted_lov(self, lov: list, var_type: str):
        return self.__adapter._get_adapted_lov(lov, var_type)

    def table_on_edit(self, state: State, var_name: str, payload: t.Dict[str, t.Any]):
        """Default implementation of the `on_edit` callback for tables.

           This function sets the value of a specific cell in the tabular dataset stored in
           *var_name*, typically bound to the *data* property of a table control.

        Arguments:
            state: the state instance received in the callback.
            var_name: the name of the variable bound to the table's *data* property.
            payload: the payload dictionary received from the `on_edit` callback.<br/>
                This dictionary has the following keys:

                - *"index"*: The row index of the cell to be modified.
                - *"col"*: Specifies the name of the column of the cell to be modified.
                - *"value"*: Specifies the new value to be assigned to the cell.
                - *"user_value"*: Contains the text entered by the user.
                - *"tz"*: Specifies the timezone to be used, if applicable.
        """
        try:
            setattr(state, var_name, self._get_accessor().on_edit(getattr(state, var_name), payload))
        except Exception as e:
            _warn("Gui.table_on_edit() failed potentially from a table's on_edit callback.", e)

    def table_on_add(
        self, state: State, var_name: str, payload: t.Dict[str, t.Any], new_row: t.Optional[t.List[t.Any]] = None
    ):
        """Default implementation of the `on_add` callback for tables.

        This function creates a new row in the tabular dataset stored in *var_name*.<br/>
        The row is added at the index specified in *payload["index"]*.

        Arguments:
            state: The state instance received from the callback.
            var_name: The name of the variable bound to the table's *data* property.
            payload: The payload dictionary received from the `on_add` callback.
            new_row: The initial values for the new row.<br/>
                If this parameter is not specified, the new row is initialized with all values set
                to 0, with the exact meaning depending on the column data type.
        """
        try:
            setattr(state, var_name, self._get_accessor().on_add(getattr(state, var_name), payload, new_row))
        except Exception as e:
            _warn("Gui.table_on_add() failed potentially from a table's on_add callback.", e)

    def table_on_delete(self, state: State, var_name: str, payload: t.Dict[str, t.Any]):
        """Default implementation of the `on_delete` callback for tables.

        This function removes a row from the tabular dataset stored in *var_name*.<br/>
        The row to be removed is located at the index specified in *payload["index"]*.

        Arguments:
            state: The state instance received in the callback.
            var_name: The name of the variable bound to the table's *data* property.
            payload: The payload dictionary received from the `on_delete` callback.
        """
        try:
            setattr(state, var_name, self._get_accessor().on_delete(getattr(state, var_name), payload))
        except Exception as e:
            _warn("Gui.table_on_delete() failed potentially from a table's on_delete callback.", e)

    def _tbl_cols(
        self, rebuild: bool, rebuild_val: t.Optional[bool], attr_json: str, hash_json: str, **kwargs
    ) -> t.Union[str, _DoNotUpdate]:
        if not self.__is_building():
            try:
                rebuild = rebuild_val if rebuild_val is not None else rebuild
                if rebuild:
                    attributes, hashes = self.__get_attributes(attr_json, hash_json, kwargs)
                    data_hash = hashes.get("data", "")
                    data = kwargs.get(data_hash)
                    col_dict = _get_columns_dict(
                        data,
                        attributes.get("columns", {}),
                        self._get_accessor().get_col_types(data_hash, _TaipyData(data, data_hash)),
                        attributes.get("date_format"),
                        attributes.get("number_format"),
                    )
                    _enhance_columns(attributes, hashes, t.cast(dict, col_dict), "table(cols)")

                    return json.dumps(col_dict, cls=_TaipyJsonEncoder)
            except Exception as e:  # pragma: no cover
                _warn("Exception while rebuilding table columns", e)
        return Gui.__DO_NOT_UPDATE_VALUE

    def _chart_conf(
        self, rebuild: bool, rebuild_val: t.Optional[bool], attr_json: str, hash_json: str, **kwargs
    ) -> t.Union[str, _DoNotUpdate]:
        if not self.__is_building():
            try:
                rebuild = rebuild_val if rebuild_val is not None else rebuild
                if rebuild:
                    attributes, hashes = self.__get_attributes(attr_json, hash_json, kwargs)
                    data_hash = hashes.get("data", "")
                    config = _build_chart_config(
                        self,
                        attributes,
                        self._get_accessor().get_col_types(data_hash, _TaipyData(kwargs.get(data_hash), data_hash)),
                    )

                    return json.dumps(config, cls=_TaipyJsonEncoder)
            except Exception as e:  # pragma: no cover
                _warn("Exception while rebuilding chart config", e)
        return Gui.__DO_NOT_UPDATE_VALUE

    # Proxy methods for Adapter

    def _add_adapter_for_type(self, type_name: str, adapter: t.Callable) -> None:
        self.__adapter._add_for_type(type_name, adapter)

    def _add_type_for_var(self, var_name: str, type_name: str) -> None:
        self.__adapter._add_type_for_var(var_name, type_name)

    def _get_adapter_for_type(self, type_name: str) -> t.Optional[t.Callable]:
        return self.__adapter._get_for_type(type_name)

    def _get_unique_type_adapter(self, type_name: str) -> str:
        return self.__adapter._get_unique_type(type_name)

    def _run_adapter(
        self, adapter: t.Optional[t.Callable], value: t.Any, var_name: str, id_only=False
    ) -> t.Union[t.Tuple[str, ...], str, None]:
        return self.__adapter._run(adapter, value, var_name, id_only)

    def _get_valid_adapter_result(self, value: t.Any, id_only=False) -> t.Union[t.Tuple[str, ...], str, None]:
        return self.__adapter._get_valid_result(value, id_only)

    def _is_ui_blocked(self):
        return _getscopeattr(self, Gui.__UI_BLOCK_NAME, False)

    def __get_on_cancel_block_ui(self, callback: t.Optional[str]):
        def _taipy_on_cancel_block_ui(guiApp, id: t.Optional[str], payload: t.Any):
            if _hasscopeattr(guiApp, Gui.__UI_BLOCK_NAME):
                _setscopeattr(guiApp, Gui.__UI_BLOCK_NAME, False)
            guiApp.__on_action(id, {"action": callback})

        return _taipy_on_cancel_block_ui

    def __add_pages_in_folder(self, folder_name: str, folder_path: str):
        from ._renderers import Html, Markdown

        list_of_files = os.listdir(folder_path)
        for file_name in list_of_files:
            if file_name.startswith("__"):
                continue
            if (re_match := Gui.__RE_HTML.match(file_name)) and f"{re_match.group(1)}.py" not in list_of_files:
                _renderers = Html(os.path.join(folder_path, file_name), frame=None)
                _renderers.modify_taipy_base_url(folder_name)
                self.add_page(name=f"{folder_name}/{re_match.group(1)}", page=_renderers)
            elif (re_match := Gui.__RE_MD.match(file_name)) and f"{re_match.group(1)}.py" not in list_of_files:
                _renderers_md = Markdown(os.path.join(folder_path, file_name), frame=None)
                self.add_page(name=f"{folder_name}/{re_match.group(1)}", page=_renderers_md)
            elif re_match := Gui.__RE_PY.match(file_name):
                module_name = re_match.group(1)
                module_path = os.path.join(folder_name, module_name).replace(os.path.sep, ".")
                try:
                    module = importlib.import_module(module_path)
                    page_instance = _get_page_from_module(module)
                    if page_instance is not None:
                        self.add_page(name=f"{folder_name}/{module_name}", page=page_instance)
                except Exception as e:
                    _warn(f"Error while importing module '{module_path}'", e)
            elif os.path.isdir(child_dir_path := os.path.join(folder_path, file_name)):
                child_dir_name = f"{folder_name}/{file_name}"
                self.__add_pages_in_folder(child_dir_name, child_dir_path)

    # Proxy methods for LocalsContext
    def _get_locals_context_obj(self) -> _LocalsContext:
        return self.__locals_context

    def _get_variable_directory_obj(self) -> _VariableDirectory:
        return self.__var_dir

    def _get_locals_bind(self) -> t.Dict[str, t.Any]:
        return self.__locals_context.get_locals()

    def _get_default_locals_bind(self) -> t.Dict[str, t.Any]:
        return self.__locals_context.get_default()

    def _get_locals_bind_from_context(self, context: t.Optional[str]) -> t.Dict[str, t.Any]:
        return self.__locals_context._get_locals_bind_from_context(context)

    def _get_locals_context(self) -> str:
        current_context = self.__locals_context.get_context()
        return current_context if current_context is not None else t.cast(str, self.__default_module_name)

    def _set_locals_context(self, context: t.Optional[str]) -> t.ContextManager[None]:
        return self.__locals_context.set_locals_context(context)

    def _has_set_context(self):
        return self.__locals_context.get_context() is not None

    def _get_page_context(self, page_name: str) -> str | None:
        if page_name not in self._config.routes:
            return None
        page = None
        for p in self._config.pages:
            if p._route == page_name:
                page = p
        if page is None:
            return None
        return (
            (page._renderer._get_module_name() or self.__default_module_name)
            if page._renderer is not None
            else self.__default_module_name
        )

    @staticmethod
    def _get_root_page_name():
        return Gui.__root_page_name

    def _set_flask(self, flask: Flask):
        self._flask = flask

    def _get_default_module_name(self):
        return self.__default_module_name

    @staticmethod
    def _get_timezone() -> str:
        return Gui.__LOCAL_TZ

    @staticmethod
    def _set_timezone(tz: str):
        Gui.__LOCAL_TZ = tz

    # Public methods
    def add_page(
        self,
        name: str,
        page: t.Union[str, Page],
        style: t.Optional[str] = "",
    ) -> None:
        """Add a page to the Graphical User Interface.

        Arguments:
            name: The name of the page.
            page (Union[str, Page^]): The content of the page.<br/>
                It can be an instance of `Markdown^` or `Html^`.<br/>
                If *page* is a string, then:

                - If *page* is set to the pathname of a readable file, the page
                  content is read as Markdown input text.
                - If it is not, the page content is read from this string as
                  Markdown text.
            style (Optional[str]): Additional CSS style to apply to this page.

                - If there is style associated with a page, it is used at a global level
                - If there is no style associated with the page, the style is cleared at a global level
                - If the page is embedded in a block control, the style is ignored

        Note that page names cannot start with the slash ('/') character and that each
        page must have a unique name.
        """
        # Validate name
        if name is None:  # pragma: no cover
            raise Exception("name is required for add_page() function.")
        if not Gui.__RE_PAGE_NAME.match(name):  # pragma: no cover
            raise SyntaxError(
                f'Page name "{name}" is invalid. It must only contain letters, digits, dash (-), underscore (_), and forward slash (/) characters.'  # noqa: E501
            )
        if name.startswith("/"):  # pragma: no cover
            raise SyntaxError(f'Page name "{name}" cannot start with forward slash (/) character.')
        if name in self._config.routes:  # pragma: no cover
            raise Exception(f'Page name "{name if name != Gui.__root_page_name else "/"}" is already defined.')
        if isinstance(page, str):
            from ._renderers import Markdown

            page = Markdown(page, frame=None)
        elif not isinstance(page, Page):  # pragma: no cover
            raise Exception(
                f'Parameter "page" is invalid for page name "{name if name != Gui.__root_page_name else "/"}.'
            )
        # Init a new page
        new_page = _Page()
        new_page._route = name
        new_page._renderer = page
        new_page._style = style or page._get_style()
        new_page._script_paths = page._get_script_paths()
        # Append page to _config
        self._config.pages.append(new_page)
        self._config.routes.append(name)
        # set root page
        if name == Gui.__root_page_name:
            self._config.root_page = new_page
        # Validate Page
        _Hooks().validate_page(self, page)
        # Update locals context
        self.__locals_context.add(page._get_module_name(), page._get_locals())
        # Update variable directory
        if not page._is_class_module():
            self.__var_dir.add_frame(page._frame)
        # Special case needed for page to access gui to trigger reload in notebook
        if _is_in_notebook():
            page._notebook_gui = self
            page._notebook_page = new_page
        # add page to hook
        _Hooks().add_page(self, page)

    def add_pages(self, pages: t.Optional[t.Union[t.Mapping[str, t.Union[str, Page]], str]] = None) -> None:
        """Add several pages to the Graphical User Interface.

        Arguments:
            pages (Union[dict[str, Union[str, Page^]], str]): The pages to add.<br/>
                If *pages* is a dictionary, a page is added to this `Gui` instance
                for each of the entries in *pages*:

                - The entry key is used as the page name.
                - The entry value is used as the page content:
                    - The value can can be an instance of `Markdown^` or `Html^`, then
                      it is used as the page definition.
                    - If entry value is a string, then:
                        - If it is set to the pathname of a readable file, the page
                          content is read as Markdown input text.
                        - If it is not, the page content is read from this string as
                          Markdown text.

        !!! note "Reading pages from a directory"
            If *pages* is a string that holds the path to a readable directory, then
            this directory is traversed, recursively, to find files that Taipy can build
            pages from.

            For every new directory that is traversed, a new hierarchical level
            for pages is created.

            For every file that is found:

            - If the filename extension is *.md*, it is read as Markdown content and
              a new page is created with the base name of this filename.
            - If the filename extension is *.html*, it is read as HTML content and
              a new page is created with the base name of this filename.

            For example, say you have the following directory structure:
            ```
            reports
            ├── home.html
            ├── budget/
            │   ├── expenses/
            │   │   ├── marketing.md
            │   │   └── production.md
            │   └── revenue/
            │       ├── EMAE.md
            │       ├── USA.md
            │       └── ASIA.md
            └── cashflow/
                ├── weekly.md
                ├── monthly.md
                └── yearly.md
            ```

            Calling `gui.add_pages('reports')` is equivalent to calling:
            ```py
            gui.add_pages({
                            "reports/home", Html("reports/home.html"),
                            "reports/budget/expenses/marketing", Markdown("reports/budget/expenses/marketing.md"),
                            "reports/budget/expenses/production", Markdown("reports/budget/expenses/production.md"),
                            "reports/budget/revenue/EMAE", Markdown("reports/budget/revenue/EMAE.md"),
                            "reports/budget/revenue/USA", Markdown("reports/budget/revenue/USA.md"),
                            "reports/budget/revenue/ASIA", Markdown("reports/budget/revenue/ASIA.md"),
                            "reports/cashflow/weekly", Markdown("reports/cashflow/weekly.md"),
                            "reports/cashflow/monthly", Markdown("reports/cashflow/monthly.md"),
                            "reports/cashflow/yearly", Markdown("reports/cashflow/yearly.md")
            })
            ```
        """
        if isinstance(pages, dict):
            for k, v in pages.items():
                if k == "/":
                    k = Gui.__root_page_name
                self.add_page(name=k, page=v)
        elif isinstance(folder_name := pages, str):
            if not hasattr(self, "_root_dir"):
                self._root_dir = os.path.dirname(getabsfile(self.__frame))
            folder_path = folder_name if os.path.isabs(folder_name) else os.path.join(self._root_dir, folder_name)
            folder_name = os.path.basename(folder_path)
            if not os.path.isdir(folder_path):  # pragma: no cover
                raise RuntimeError(f"Path {folder_path} is not a valid directory")
            if folder_name in self.__directory_name_of_pages:  # pragma: no cover
                raise Exception(f"Base directory name {folder_name} of path {folder_path} is not unique")
            if folder_name in Gui.__reserved_routes:  # pragma: no cover
                raise Exception(f"Invalid directory. Directory {folder_name} is a reserved route")
            self.__directory_name_of_pages.append(folder_name)
            self.__add_pages_in_folder(folder_name, folder_path)

    # partials

    def add_partial(
        self,
        page: t.Union[str, Page],
    ) -> Partial:
        """Create a new `Partial^`.

        The [User Manual section on Partials](../../../../../userman/gui/pages/partial/index.md)
        gives details on when and how to use this class.

        Arguments:
            page (Union[str, Page^]): The page to create a new Partial from.<br/>
                It can be an instance of `Markdown^` or `Html^`.<br/>
                If *page* is a string, then:

                - If *page* is set to the pathname of a readable file, the content of
                  the new `Partial` is read as Markdown input text.
                - If it is not, the content of the new `Partial` is read from this string
                  as Markdown text.

        Returns:
            The new `Partial` object defined by *page*.
        """
        new_partial = Partial()
        # Validate name
        if (
            new_partial._route in self._config.partial_routes or new_partial._route in self._config.routes
        ):  # pragma: no cover
            _warn(f'Partial name "{new_partial._route}" is already defined.')
        if isinstance(page, str):
            from ._renderers import Markdown

            page = Markdown(page, frame=None)
        elif not isinstance(page, Page):  # pragma: no cover
            raise Exception(f'Partial name "{new_partial._route}" has an invalid Page.')
        new_partial._renderer = page
        # Append partial to _config
        self._config.partials.append(new_partial)
        self._config.partial_routes.append(str(new_partial._route))
        # Update locals context
        self.__locals_context.add(page._get_module_name(), page._get_locals())
        # Update variable directory
        self.__var_dir.add_frame(page._frame)
        return new_partial

    def _update_partial(self, partial: Partial):
        partials = _getscopeattr(self, Partial._PARTIALS, {})
        partials[partial._route] = partial
        _setscopeattr(self, Partial._PARTIALS, partials)
        self.__send_ws_partial(str(partial._route))

    def _get_partial(self, route: str) -> t.Optional[Partial]:
        partials = _getscopeattr(self, Partial._PARTIALS, {})
        partial = partials.get(route)
        if partial is None:
            partial = next((p for p in self._config.partials if p._route == route), None)
        return partial

    # Main binding method (bind in markdown declaration)
    def _bind_var(self, var_name: str) -> str:
        bind_context = None
        if var_name in self._get_locals_bind().keys():
            bind_context = self._get_locals_context()
        if bind_context is None:
            encoded_var_name = self.__var_dir.add_var(var_name, self._get_locals_context(), var_name)
        else:
            encoded_var_name = self.__var_dir.add_var(var_name, bind_context)
        if not hasattr(self._bindings(), encoded_var_name):
            bind_locals = self._get_locals_bind_from_context(bind_context)
            if var_name in bind_locals.keys():
                self._bind(encoded_var_name, bind_locals[var_name])
            else:
                _warn(
                    f"Variable '{var_name}' is not available in either the '{self._get_locals_context()}' or '__main__' modules."  # noqa: E501
                )
        return encoded_var_name

    def _bind_var_val(self, var_name: str, value: t.Any) -> bool:
        if not _is_moduled_variable(var_name):
            var_name = self.__var_dir.add_var(var_name, self._get_locals_context())
        if not hasattr(self._bindings(), var_name):
            self._bind(var_name, value)
            return True
        return False

    def __bind_local_func(self, name: str):
        func = getattr(self, name, None)
        if func is not None and not _is_function(func):  # pragma: no cover
            _warn(f"{self.__class__.__name__}.{name}: {func} should be a function; looking for {name} in the script.")
            func = None
        if func is None:
            func = self._get_locals_bind().get(name)
        if func is not None:
            if _is_function(func):
                setattr(self, name, func)
            else:  # pragma: no cover
                _warn(f"{name}: {func} should be a function.")

    def load_config(self, config: Config) -> None:
        self._config._load(config)

    def _broadcast(
        self,
        name: str,
        value: t.Any,
        client_id: t.Optional[str] = None,
        message_type: t.Optional[_WsType] = None,
    ):
        """NOT DOCUMENTED
        Send the new value of a variable to all connected clients.

        Arguments:
            name: The name of the variable to update or create.
            value: The value (must be serializable to the JSON format).
            client_id: The client id (broadcast to all client if None)
        """
        self.__send_ws_broadcast(name, value, client_id, message_type)

    def _broadcast_all_clients(self, name: str, value: t.Any):
        try:
            self._set_broadcast()
            self._update_var(name, value)
        finally:
            self._set_broadcast(False)

    def _set_broadcast(self, broadcast: bool = True):
        with contextlib.suppress(RuntimeError):
            setattr(g, Gui.__BROADCAST_G_ID, broadcast)

    def _is_broadcasting(self) -> bool:
        try:
            return getattr(g, Gui.__BROADCAST_G_ID, False)
        except RuntimeError:
            return False

    def _download(
        self, content: t.Any, name: t.Optional[str] = "", on_action: t.Optional[t.Union[str, t.Callable]] = ""
    ):
        if _is_function(on_action):
            on_action_name = (
                _get_lambda_id(t.cast(LambdaType, on_action))
                if _is_unnamed_function(on_action)
                else _get_expr_var_name(t.cast(t.Callable, on_action).__name__)
            )
            if on_action_name:
                self._bind_var_val(on_action_name, on_action)
                on_action = on_action_name
            else:
                _warn("download() on_action is invalid.")
        content_str = self._get_content("Gui.download", content, False)
        self.__send_ws_download(content_str, str(name), str(on_action) if on_action is not None else "")

    def _notify(
        self,
        notification_type: str = "I",
        message: str = "",
        system_notification: t.Optional[bool] = None,
        duration: t.Optional[int] = None,
        notification_id: t.Optional[str] = None,
    ):
        if not notification_id:
            notification_id = str(uuid.uuid4())

        self.__send_ws_alert(
            notification_type,
            message,
            self._get_config("system_notification", False) if system_notification is None else system_notification,
            self._get_config("notification_duration", 3000) if duration is None else duration,
            notification_id,
        )
        return notification_id

    def _close_notification(
        self,
        notification_id: str,
    ):
        if notification_id:
            self.__send_ws_alert(
                type="",  # Since you're closing, set type to an empty string or a predefined "close" type
                message="",  # No need for a message when closing
                system_notification=False,  # System notification not needed for closing
                duration=0,  # No duration since it's an immediate close
                notification_id=notification_id,
            )

    def _hold_actions(
        self,
        callback: t.Optional[t.Union[str, t.Callable]] = None,
        message: t.Optional[str] = "Work in Progress...",
    ):  # pragma: no cover
        action_name = (
            callback
            if isinstance(callback, str)
            else _get_lambda_id(t.cast(LambdaType, callback))
            if _is_unnamed_function(callback)
            else callback.__name__
            if callback is not None
            else None
        )
        func = self.__get_on_cancel_block_ui(action_name)
        def_action_name = func.__name__
        _setscopeattr(self, def_action_name, func)

        if _hasscopeattr(self, Gui.__UI_BLOCK_NAME):
            _setscopeattr(self, Gui.__UI_BLOCK_NAME, True)
        else:
            self._bind(Gui.__UI_BLOCK_NAME, True)
        self.__send_ws_block(action=def_action_name, message=message, cancel=bool(action_name))

    def _resume_actions(self):  # pragma: no cover
        if _hasscopeattr(self, Gui.__UI_BLOCK_NAME):
            _setscopeattr(self, Gui.__UI_BLOCK_NAME, False)
        self.__send_ws_block(close=True)

    def _navigate(
        self,
        to: t.Optional[str] = "",
        params: t.Optional[t.Dict[str, str]] = None,
        tab: t.Optional[str] = None,
        force: t.Optional[bool] = False,
    ):
        to = to or Gui.__root_page_name
        if not to.startswith("/") and to not in self._config.routes and not urlparse(to).netloc:
            _warn(f'Cannot navigate to "{to if to != Gui.__root_page_name else "/"}": unknown page.')
            return False
        self.__send_ws_navigate(to if to != Gui.__root_page_name else "/", params, tab, force or False)
        return True

    def __init_libs(self):
        for name, libs in self.__extensions.items():
            for lib in libs:
                if not isinstance(lib, ElementLibrary):
                    continue
                try:
                    self._call_function_with_state(lib.on_user_init)
                except Exception as e:  # pragma: no cover
                    if not self._call_on_exception(f"{name}.on_user_init", e):
                        _warn(f"Exception raised in {name}.on_user_init()", e)

    def __init_route(self):
        self.__set_client_id_in_context(force=True)
        if not _hasscopeattr(self, Gui.__ON_INIT_NAME):
            _setscopeattr(self, Gui.__ON_INIT_NAME, True)
            self.__pre_render_pages()
            self.__init_libs()
            if hasattr(self, "on_init") and _is_function(self.on_init):
                try:
                    self._call_function_with_state(t.cast(t.Callable, self.on_init))
                except Exception as e:  # pragma: no cover
                    if not self._call_on_exception("on_init", e):
                        _warn("Exception raised in on_init()", e)
        return self._render_route()

    def _call_on_exception(self, function: t.Any, exception: Exception) -> bool:
        if hasattr(self, "on_exception") and _is_function(self.on_exception):
            function_name = _function_name(function) if callable(function) else str(function)
            try:
                self._call_function_with_state(t.cast(t.Callable, self.on_exception), [function_name, exception])
            except Exception as e:  # pragma: no cover
                _warn("Exception raised in on_exception()", e)
            return True
        return False

    def __call_on_status(self) -> t.Optional[str]:
        if hasattr(self, "on_status") and _is_function(self.on_status):
            try:
                return self._call_function_with_state(t.cast(t.Callable, self.on_status))
            except Exception as e:  # pragma: no cover
                if not self._call_on_exception("on_status", e):
                    _warn("Exception raised in on_status", e)
        return None

    def __pre_render_pages(self) -> None:
        """Pre-render all pages to have a proper initialization of all variables"""
        self.__set_client_id_in_context()
        scope_metadata = self._get_data_scope_metadata()
        if scope_metadata[_DataScopes._META_PRE_RENDER]:
            return
        for page in self._config.pages:
            if page is not None:
                with contextlib.suppress(Exception):
                    if isinstance(page._renderer, CustomPage):
                        self._bind_custom_page_variables(page._renderer, self._get_client_id())
                    else:
                        page.render(self, silent=True)
        scope_metadata[_DataScopes._META_PRE_RENDER] = True

    def _get_navigated_page(self, page_name: str) -> t.Any:
        nav_page = page_name
        if hasattr(self, "on_navigate") and _is_function(self.on_navigate):
            try:
                params = request.args.to_dict() if hasattr(request, "args") else {}
                params.pop("client_id", None)
                params.pop("v", None)
                nav_page = self._call_function_with_state(t.cast(t.Callable, self.on_navigate), [page_name, params])
                if nav_page != page_name:
                    if isinstance(nav_page, str):
                        if self._navigate(nav_page):
                            return ("Root page cannot be re-routed by on_navigate().", 302)
                    else:
                        _warn(f"on_navigate() returned an invalid page name '{nav_page}'.")
                    nav_page = page_name
            except Exception as e:  # pragma: no cover
                if not self._call_on_exception("on_navigate", e):
                    _warn("Exception raised in on_navigate()", e)
        return nav_page

    def _call_on_page_load(self, page_name: str) -> None:
        if page_name == Gui.__root_page_name:
            page_name = "/"
        on_page_load_fn = self._get_user_function("on_page_load")
        if not _is_function(on_page_load_fn):
            return
        try:
            self._call_function_with_state(t.cast(t.Callable, on_page_load_fn), [page_name])
        except Exception as e:
            if not self._call_on_exception("on_page_load", e):
                _warn("Exception raised in on_page_load()", e)

    def _get_page(self, page_name: str):
        return next((page_i for page_i in self._config.pages if page_i._route == page_name), None)

    def _bind_custom_page_variables(self, page: CustomPage, client_id: t.Optional[str]):
        """Handle the bindings of custom page variables"""
        if not isinstance(page, CustomPage):
            return
        with self.get_flask_app().app_context() if has_app_context() else contextlib.nullcontext():  # type: ignore[attr-defined]
            self.__set_client_id_in_context(client_id)
            with self._set_locals_context(page._get_module_name()):
                for k, v in self._get_locals_bind().items():
                    if (
                        (not page._binding_variables or k in page._binding_variables)
                        and not k.startswith("_")
                        and not isinstance(v, ModuleType)
                    ):
                        self._bind_var(k)

    def __render_page(self, page_name: str) -> t.Any:
        self.__set_client_id_in_context()
        nav_page = self._get_navigated_page(page_name)
        if not isinstance(nav_page, str):
            return nav_page
        page = self._get_page(nav_page)
        # Try partials
        if page is None:
            page = self._get_partial(nav_page)
        # Make sure that there is a page instance found
        if page is None:
            return (
                jsonify({"error": f"Page '{nav_page}' doesn't exist."}),
                400,
                {"Content-Type": "application/json; charset=utf-8"},
            )
        # Handle custom pages
        if (pr := page._renderer) is not None and isinstance(pr, CustomPage):
            if self._navigate(
                to=page_name,
                params={
                    _Server._RESOURCE_HANDLER_ARG: pr._resource_handler.get_id(),
                },
            ):
                # Proactively handle the bindings of custom page variables
                self._bind_custom_page_variables(pr, self._get_client_id())
                return ("Successfully redirect to custom resource handler", 200)
            return ("Failed to navigate to custom resource handler", 500)
        # Handle page rendering
        context = page.render(self)
        if (
            nav_page == Gui.__root_page_name
            and page._rendered_jsx is not None
            and "<PageContent" not in page._rendered_jsx
        ):
            page._rendered_jsx += "<PageContent />"
        # Return jsx page
        if page._rendered_jsx is not None:
            with self._set_locals_context(context):
                self._call_on_page_load(nav_page)
            return self._server._render(
                page._rendered_jsx, page._script_paths if page._script_paths is not None else [], page._style if page._style is not None else "", page._head, context # noqa: E501
            )
        else:
            return ("No page template", 404)

    def _render_route(self) -> t.Any:
        return self._server._direct_render_json(
            {
                "locations": {
                    "/" if route == Gui.__root_page_name else f"/{route}": f"/{route}" for route in self._config.routes
                },
                "blockUI": self._is_ui_blocked(),
            }
        )

    def get_flask_app(self) -> Flask:
        """Get the internal Flask application.

        This method must be called **after** `(Gui.)run()^` was invoked.

        Returns:
            The Flask instance used.
        """
        if hasattr(self, "_server"):
            return t.cast(Flask, self._server.get_flask())
        raise RuntimeError("get_flask_app() cannot be invoked before run() has been called.")

    def _get_port(self) -> int:
        return self._server.get_port()

    def _set_frame(self, frame: t.Optional[FrameType]):
        if not isinstance(frame, FrameType):  # pragma: no cover
            raise RuntimeError("frame must be a FrameType where Gui can collect the local variables.")
        self.__frame = frame
        self.__default_module_name = _get_module_name_from_frame(self.__frame)

    def _set_css_file(self, css_file: t.Optional[str] = None):
        script_file = Path(self.__frame.f_code.co_filename or ".").resolve()
        if css_file is None:
            if script_file.with_suffix(".css").exists():
                css_file = f"{script_file.stem}.css"
            elif script_file.is_dir() and (script_file / "taipy.css").exists():
                css_file = "taipy.css"
        if css_file is None:
            script_file = script_file.with_name("taipy").with_suffix(".css")
            if script_file.exists():
                css_file = f"{script_file.stem}.css"
        self.__css_file = css_file

    def _set_state(self, state: State):
        if isinstance(state, State):
            self.__state = state

    def _get_webapp_path(self):
        _conf_webapp_path = (
            Path(self._get_config("webapp_path", None)) if self._get_config("webapp_path", None) else None
        )
        _webapp_path = str((Path(__file__).parent / "webapp").resolve())
        if _conf_webapp_path:
            if _conf_webapp_path.is_dir():
                _webapp_path = str(_conf_webapp_path.resolve())
                _warn(f"Using webapp_path: '{_conf_webapp_path}'.")
            else:  # pragma: no cover
                _warn(
                    f"webapp_path: '{_conf_webapp_path}' is not a valid directory. Falling back to '{_webapp_path}'."  # noqa: E501
                )
        return _webapp_path

    def __get_client_config(self) -> t.Dict[str, t.Any]:
        config = {
            "timeZone": self._config.get_time_zone(),
            "darkMode": self._get_config("dark_mode", True),
            "baseURL": self._config._get_config("base_url", "/"),
        }
        if themes := self._get_themes():
            config["themes"] = themes
        if len(self.__extensions):
            config["extensions"] = {}
            for libs in self.__extensions.values():
                for lib in libs:
                    config["extensions"][f"./{Gui._EXTENSION_ROOT}/{lib.get_js_module_name()}"] = [
                        e._get_js_name(n)
                        for n, e in lib.get_elements().items()
                        if isinstance(e, Element) and not e._is_server_only()
                    ]
        if stylekit := self._get_config("stylekit", _default_stylekit):
            config["stylekit"] = {_to_camel_case(k): v for k, v in stylekit.items()}
        return config

    def __get_css_vars(self) -> str:
        css_vars = []
        if stylekit := self._get_config("stylekit", _default_stylekit):
            for k, v in stylekit.items():
                css_vars.append(f'--{k.replace("_", "-")}:{_get_css_var_value(v)};')
        return " ".join(css_vars)

    def __init_server(self):
        app_config = self._config.config
        # Init server if there is no server
        if not hasattr(self, "_server"):
            self._server = _Server(
                self,
                path_mapping=self._path_mapping,
                flask=self._flask,
                async_mode=app_config.get("async_mode"),
                allow_upgrades=not app_config.get("notebook_proxy"),
                server_config=app_config.get("server_config"),
            )

        # Stop and reinitialize the server if it is still running as a thread
        if (_is_in_notebook() or app_config.get("run_in_thread")) and hasattr(self._server, "_thread"):
            self.stop()
            self._flask_blueprint = []
            self._server = _Server(
                self,
                path_mapping=self._path_mapping,
                flask=self._flask,
                async_mode=app_config.get("async_mode"),
                allow_upgrades=not app_config.get("notebook_proxy"),
                server_config=app_config.get("server_config"),
            )
            self._bindings()._new_scopes()

    def __init_ngrok(self):
        app_config = self._config.config
        if hasattr(self, "_ngrok"):
            # Keep the ngrok instance if token has not changed
            if app_config.get("ngrok_token") == self._ngrok[1]:
                _TaipyLogger._get_logger().info(f" * NGROK Public Url: {self._ngrok[0].public_url}")
                return
            # Close the old tunnel so new tunnel can open for new token
            ngrok.disconnect(self._ngrok[0].public_url)  # type: ignore[reportPossiblyUnboundVariable]
        if app_config.get("run_server") and (token := app_config.get("ngrok_token")):  # pragma: no cover
            if not util.find_spec("pyngrok"):
                raise RuntimeError("Cannot use ngrok as pyngrok package is not installed.")
            ngrok.set_auth_token(token)  # type: ignore[reportPossiblyUnboundVariable]
            self._ngrok = (ngrok.connect(app_config.get("port"), "http"), token)  # type: ignore[reportPossiblyUnboundVariable]
            _TaipyLogger._get_logger().info(f" * NGROK Public Url: {self._ngrok[0].public_url}")

    def __bind_default_function(self):
        with self.get_flask_app().app_context():
            self.__var_dir.process_imported_var()
            # bind on_* function if available
            self.__bind_local_func("on_init")
            self.__bind_local_func("on_change")
            self.__bind_local_func("on_action")
            self.__bind_local_func("on_page_load")
            self.__bind_local_func("on_navigate")
            self.__bind_local_func("on_exception")
            self.__bind_local_func("on_status")
            self.__bind_local_func("on_user_content")

    def __register_blueprint(self):
        # add en empty main page if it is not defined
        if Gui.__root_page_name not in self._config.routes:
            new_page = _Page()
            new_page._route = Gui.__root_page_name
            new_page._renderer = _EmptyPage()
            self._config.pages.append(new_page)
            self._config.routes.append(Gui.__root_page_name)

        pages_bp = Blueprint("taipy_pages", __name__)
        self._flask_blueprint.append(pages_bp)

        # server URL Rule for taipy images
        images_bp = Blueprint("taipy_images", __name__)
        images_bp.add_url_rule(f"/{Gui.__CONTENT_ROOT}/<path:path>", view_func=self.__serve_content)
        self._flask_blueprint.append(images_bp)

        # server URL for uploaded files
        upload_bp = Blueprint("taipy_upload", __name__)
        upload_bp.add_url_rule(f"/{Gui.__UPLOAD_URL}", view_func=self.__upload_files, methods=["POST"])
        self._flask_blueprint.append(upload_bp)

        # server URL for user content
        user_content_bp = Blueprint("taipy_user_content", __name__)
        user_content_bp.add_url_rule(f"/{Gui.__USER_CONTENT_URL}/<path:path>", view_func=self.__serve_user_content)
        self._flask_blueprint.append(user_content_bp)

        # server URL for extension resources
        extension_bp = Blueprint("taipy_extensions", __name__)
        extension_bp.add_url_rule(f"/{Gui._EXTENSION_ROOT}/<path:path>", view_func=self.__serve_extension)
        scripts = [
            s if bool(urlparse(s).netloc) else f"{Gui._EXTENSION_ROOT}/{name}/{s}{lib.get_query(s)}"
            for name, libs in Gui.__extensions.items()
            for lib in libs
            for s in (lib._do_get_relative_paths(lib.get_scripts()))
        ]
        styles = [
            s if bool(urlparse(s).netloc) else f"{Gui._EXTENSION_ROOT}/{name}/{s}{lib.get_query(s)}"
            for name, libs in Gui.__extensions.items()
            for lib in libs
            for s in (lib._do_get_relative_paths(lib.get_styles()))
        ]
        if self._get_config("stylekit", True):
            styles.append("stylekit/stylekit.css")
        else:
            styles.append(Gui.__ROBOTO_FONT)
        if self.__css_file:
            styles.append(f"{self.__css_file}")

        if self.__script_files:
            scripts.extend(self.__script_files)

        self._flask_blueprint.append(extension_bp)

        _webapp_path = self._get_webapp_path()

        self._flask_blueprint.append(
            self._server._get_default_blueprint(
                static_folder=_webapp_path,
                template_folder=_webapp_path,
                title=self._get_config("title", "Taipy App"),
                favicon=self._get_config("favicon", Gui.__DEFAULT_FAVICON_URL),
                root_margin=self._get_config("margin", None),
                scripts=scripts,
                styles=styles,
                version=self.__get_version(),
                client_config=self.__get_client_config(),
                watermark=self._get_config("watermark", None),
                css_vars=self.__get_css_vars(),
                base_url=self._get_config("base_url", "/"),
            )
        )

        # Run parse markdown to force variables binding at runtime
        pages_bp.add_url_rule(f"/{Gui.__JSX_URL}/<path:page_name>", view_func=self.__render_page)

        # server URL Rule for flask rendered react-router
        pages_bp.add_url_rule(f"/{Gui.__INIT_URL}", view_func=self.__init_route)

        _Hooks()._add_external_blueprint(self, __name__)

        # Register Flask Blueprint if available
        for bp in self._flask_blueprint:
            t.cast(Flask, self._server.get_flask()).register_blueprint(bp)

    def _get_accessor(self):
        if self.__accessors is None:
            self.__accessors = _DataAccessors(self)
        return self.__accessors

    def run(
        self,
        run_server: bool = True,
        run_in_thread: bool = False,
        async_mode: str = "gevent",
        **kwargs,
    ) -> t.Optional[Flask]:
        """Start the server that delivers pages to web clients.

        Once you enter `run()`, users can run web browsers and point to the web server
        URL that `Gui` serves. The default is to listen to the *localhost* address
        (127.0.0.1) on the port number 5000. However, the configuration of this `Gui`
        object may impact that (see the
        [Configuration](../../../../../userman/advanced_features/configuration/gui-config.md#configuring-the-gui-instance)
        section of the User Manual for details).

        Arguments:
            run_server (bool): Whether or not to run a web server locally.
                If set to *False*, a web server is *not* created and started.
            run_in_thread (bool): Whether or not to run a web server in a separated thread.
                If set to *True*, the web server runs is a separated thread.<br/>
                Note that if you are running in an IPython notebook context, the web
                server always runs in a separate thread.
            async_mode (str): The asynchronous model to use for the Flask-SocketIO.
                Valid values are:<br/>

                - "gevent": Use a [gevent](https://www.gevent.org/servers.html) server.
                - "threading": Use the Flask Development Server. This allows the application to use
                  the Flask reloader (the *use_reloader* option) and Debug mode (the *debug* option).
                - "eventlet": Use an [*eventlet*](https://flask.palletsprojects.com/en/2.2.x/deploying/eventlet/)
                  event-driven WSGI server.

                The default value is "gevent"<br/>
                Note that only the "threading" value provides support for the development reloader
                functionality (*use_reloader* option). Any other value makes the *use_reloader* configuration parameter
                ignored.<br/>
                Also note that setting the *debug* argument to True forces *async_mode* to "threading".
            **kwargs (dict[str, any]): Additional keyword arguments that configure how this `Gui` is run.
                Please refer to the gui config section
                [page](../../../../../userman/advanced_features/configuration/gui-config.md#configuring-the-gui-instance)
                of the User Manual for more information.

        Returns:
            The Flask instance if *run_server* is False else None.
        """
        # --------------------------------------------------------------------------------
        # The ssl_context argument was removed just after 1.1. It was defined as:
        # t.Optional[t.Union[ssl.SSLContext, t.Tuple[str, t.Optional[str]], t.Literal["adhoc"]]] = None
        #
        # With the doc:
        #     ssl_context (Optional[Union[ssl.SSLContext, Tuple[str, Optional[str]], t.Literal['adhoc']]]):
        #         Configures TLS to serve over HTTPS. This value can be:
        #
        #         - An `ssl.SSLContext` object
        #         - A `(cert_file, key_file)` tuple to create a typical context
        #         - The string "adhoc" to generate a temporary self-signed certificate.
        #
        #         The default value is None.
        # --------------------------------------------------------------------------------
        app_config = self._config.config

        run_root_dir = os.path.dirname(getabsfile(self.__frame))

        # Register _root_dir for abs path
        if not hasattr(self, "_root_dir"):
            self._root_dir = run_root_dir

        is_reloading = kwargs.pop("_reload", False)

        if not is_reloading:
            self.__run_kwargs = kwargs = {
                **kwargs,
                "run_server": run_server,
                "run_in_thread": run_in_thread,
                "async_mode": async_mode,
            }

        # Load application config from multiple sources (env files, kwargs, command line)
        self._config._build_config(run_root_dir, self.__env_filename, kwargs)

        self._config.resolve()
        TaipyGuiWarning.set_debug_mode(self._get_config("debug", False))

        # setup run function with gui hooks
        _Hooks().run(self, **kwargs)

        self.__init_server()

        self.__init_ngrok()

        locals_bind = _filter_locals(self.__frame.f_locals)

        self.__locals_context.set_default(locals_bind, t.cast(str, self.__default_module_name))

        self.__var_dir.set_default(self.__frame)

        if self.__state is None or is_reloading:
            self.__state = _GuiState(
                self, self.__locals_context.get_all_keys(), self.__locals_context.get_all_context()
            )

        if _is_in_notebook():
            # Allow gui.state.x in notebook mode
            self.state = self.__state
            """Only defined when running in an IPython notebook context.

            The unique instance of State that you can use to change bound variables directly,
            potentially impacting the user interface in real-time.
            """

        self.__bind_default_function()

        # Base global ctx is TaipyHolder classes + script modules and callables
        glob_ctx: t.Dict[str, t.Any] = {t.__name__: t for t in _TaipyBase.__subclasses__()}
        glob_ctx[Gui.__SELF_VAR] = self
        glob_ctx["state"] = self.__state
        glob_ctx.update({k: v for k, v in locals_bind.items() if ismodule(v) or callable(v)})

        # Call on_init on each library
        for name, libs in self.__extensions.items():
            for lib in libs:
                if not isinstance(lib, ElementLibrary):
                    continue
                try:
                    lib_context = lib.on_init(self)
                    if (
                        isinstance(lib_context, tuple)
                        and len(lib_context) > 1
                        and isinstance(lib_context[0], str)
                        and lib_context[0].isidentifier()
                    ):
                        if lib_context[0] in glob_ctx:
                            _warn(f"Method {name}.on_init() returned a name already defined '{lib_context[0]}'.")
                        else:
                            glob_ctx[lib_context[0]] = lib_context[1]
                    elif lib_context:
                        _warn(
                            f"Method {name}.on_init() should return a Tuple[str, Any] where the first element must be a valid Python identifier."  # noqa: E501
                        )
                except Exception as e:  # pragma: no cover
                    if not self._call_on_exception(f"{name}.on_init", e):
                        _warn(f"Method {name}.on_init() raised an exception", e)

        # Initiate the Evaluator with the right context
        self.__evaluator = _Evaluator(glob_ctx, self.__shared_variables)

        self.__register_blueprint()

        # Register data accessor communication data format (JSON, Apache Arrow)
        self._get_accessor().set_data_format(
            _DataFormat.APACHE_ARROW if app_config.get("use_arrow") else _DataFormat.JSON
        )

        # Use multi user or not
        self._bindings()._set_single_client(bool(app_config.get("single_client")))

        # Start Flask Server
        if not run_server:
            return self.get_flask_app()

        return self._server.run(
            host=app_config.get("host"),
            port=app_config.get("port"),
            client_url=app_config.get("client_url"),
            debug=app_config.get("debug"),
            use_reloader=app_config.get("use_reloader"),
            flask_log=app_config.get("flask_log"),
            run_in_thread=app_config.get("run_in_thread"),
            allow_unsafe_werkzeug=app_config.get("allow_unsafe_werkzeug"),
            notebook_proxy=app_config.get("notebook_proxy"),
            port_auto_ranges=app_config.get("port_auto_ranges"),
        )

    def reload(self):  # pragma: no cover
        """Reload the web server.

        This function reloads the underlying web server only in the situation where
        it was run in a separated thread: the *run_in_thread* parameter to the
        `(Gui.)run^` method was set to True, or you are running in an IPython notebook
        context.
        """
        if hasattr(self, "_server") and hasattr(self._server, "_thread") and self._server._is_running:
            self._server.stop_thread()
            self.run(**self.__run_kwargs, _reload=True)
            _TaipyLogger._get_logger().info("Gui server has been reloaded.")

    def stop(self):
        """Stop the web server.

        This function stops the underlying web server only in the situation where
        it was run in a separated thread: the *run_in_thread* parameter to the
        `(Gui.)run()^` method was set to True, or you are running in an IPython notebook
        context.
        """
        if hasattr(self, "_server") and hasattr(self._server, "_thread") and self._server._is_running:
            self._server.stop_thread()
            _TaipyLogger._get_logger().info("Gui server has been stopped.")

    def _get_authorization(self, client_id: t.Optional[str] = None, system: t.Optional[bool] = False):
        try:
            return _Hooks()._get_authorization(self, client_id, system) or contextlib.nullcontext()
        except Exception as e:
            _warn("Hooks:", e)
            return contextlib.nullcontext()

    def set_favicon(self, favicon_path: t.Union[str, Path], state: t.Optional[State] = None):
        """Change the favicon for all clients.

        This function dynamically changes the favicon (the icon associated with the application's
        pages) of Taipy GUI pages for a single or all connected clients.
        Note that the *favicon* parameter to `(Gui.)run()^` can also be used to change
        the favicon when the application starts.

        Arguments:
            favicon_path: The path to the image file to use.<br/>
                This can be expressed as a path name or a URL (relative or not).
            state: The state to apply the change to.<br/>
                If no set or set to None, all the application's clients are impacted.
        """
        if state or self.__favicon != favicon_path:
            if not state:
                self.__favicon = favicon_path
            url = self._get_content("__taipy_favicon", favicon_path, True)
            self._broadcast(
                "taipy_favicon", url, self._get_client_id() if state else None, message_type=_WsType.FAVICON
            )

    @staticmethod
    def _add_event_listener(
        event_name: str,
        listener: t.Union[
            t.Callable[[str, t.Dict[str, t.Any]], None], t.Callable[[State, str, t.Dict[str, t.Any]], None]
        ],
        with_state: t.Optional[bool] = False,
    ):
        _Hooks()._add_event_listener(event_name, listener, with_state)

    def _fire_event(
        self, event_name: str, client_id: t.Optional[str] = None, payload: t.Optional[t.Dict[str, t.Any]] = None
    ):
        # the event manager will take care of starting the thread
        # once the current callback (or the next one) is finished
        self.__event_manager._add_thread(
            Thread(
                target=self.__do_fire_event,
                args=(event_name, client_id, payload),
            )
        )

    def __do_fire_event(
        self, event_name: str, client_id: t.Optional[str] = None, payload: t.Optional[t.Dict[str, t.Any]] = None
    ):
        this_sid = None
        if request:
            # avoid messing with the client_id => Set(ws id)
            this_sid = getattr(request, "sid", None)
            request.sid = None  # type: ignore[attr-defined]

        try:
            with self.get_flask_app().app_context(), self.__event_manager:
                if client_id:
                    setattr(g, Gui.__ARG_CLIENT_ID, client_id)
                _Hooks()._fire_event(event_name, client_id, payload)
        finally:
            if this_sid:
                request.sid = this_sid  # type: ignore[attr-defined]