API Reference

def serialize(self) -> SerializedObject:
    """
    Serializes the instance into a dictionary, preserving the structure of the
    instance.

    For each attribute, method, and property, the method includes its name, type,
    value, readonly status, and documentation if any in the resulting dictionary.
    Attributes and methods starting with an underscore are ignored.

    For nested DataService instances, the method serializes recursively.
    For attributes of type list, each item in the list is serialized individually.
    If an item in the list is an instance of DataService, it is serialized
    recursively.

    Returns:
        dict: The serialized instance.
    """
    return Serializer.serialize_object(self)

def add_notification_callback(
    self, callback: Callable[[str, Any, SerializedObject], None]
) -> None:
    """
    Registers a callback function to be invoked upon attribute changes in the
    observed object.

    This method allows for the addition of custom callback functions that will be
    executed whenever there is a change in the value of an observed attribute. The
    callback function is called with detailed information about the change, enabling
    external logic to respond to specific state changes within the observable
    object.

    Args:
        callback:
            The callback function to be registered. The function should have the
            following signature:

            - full_access_path (str): The full dot-notation access path of the
              changed attribute. This path indicates the location of the changed
              attribute within the observable object's structure.
            - value (Any): The new value of the changed attribute.
            - cached_value_dict (dict[str, Any]): A dictionary representing the
              cached state of the attribute prior to the change. This can be useful
              for understanding the nature of the change and for historical
              comparison.
    """
    self._notification_callbacks.append(callback)

def __is_loadable_state_attribute(self, full_access_path: str) -> bool:
    """Checks if an attribute defined by a dot-separated path should be loaded from
    storage.

    For properties, it verifies the presence of the '@load_state' decorator. Regular
    attributes default to being loadable.
    """

    path_parts = parse_full_access_path(full_access_path)
    parent_object = get_object_by_path_parts(self.service, path_parts[:-1])

    if is_property_attribute(parent_object, path_parts[-1]):
        prop = getattr(type(parent_object), path_parts[-1])
        has_decorator = has_load_state_decorator(prop)
        if not has_decorator:
            logger.debug(
                "Property '%s' has no '@load_state' decorator. "
                "Ignoring value from JSON file...",
                path_parts[-1],
            )
        return has_decorator

    try:
        cached_serialization_dict = self.cache_manager.get_value_dict_from_cache(
            full_access_path
        )

        if cached_serialization_dict["value"] == "method":
            return False

        # nested objects cannot be loaded
        return not serialized_dict_is_nested_object(cached_serialization_dict)
    except SerializationPathError:
        logger.debug(
            "Path %a could not be loaded. It does not correspond to an attribute of"
            " the class. Ignoring value from JSON file...",
            path_parts[-1],
        )
        return False

async def autosave(self) -> None:
    """Periodically saves the current service state to the configured file.

    This coroutine is automatically started by the [`pydase.Server`][pydase.Server]
    when a filename is provided. It runs in the background and writes the latest
    known state of the service to disk every `autosave_interval` seconds.

    If `autosave_interval` is set to `None`, autosaving is disabled and this
    coroutine exits immediately.
    """

    if self.autosave_interval is None:
        return

    while True:
        try:
            if self.filename is not None:
                self.save_state()
            await asyncio.sleep(self.autosave_interval)
        except Exception as e:
            logger.exception(e)

def load_state(self) -> None:
    """Loads the DataService's state from a JSON file defined by `self.filename`.
    Updates the service's attributes, respecting type and read-only constraints.
    """

    # Traverse the serialized representation and set the attributes of the class
    json_dict = self._get_state_dict_from_json_file()
    if json_dict == {}:
        logger.debug("Could not load the service state.")
        return

    for path in generate_serialized_data_paths(json_dict):
        if self.__is_loadable_state_attribute(path):
            nested_json_dict = get_nested_dict_by_path(json_dict, path)
            try:
                nested_class_dict = self.cache_manager.get_value_dict_from_cache(
                    path
                )
            except (SerializationPathError, KeyError):
                nested_class_dict = {
                    "full_access_path": path,
                    "value": None,
                    "type": "None",
                    "doc": None,
                    "readonly": False,
                }

            value_type = nested_json_dict["type"]
            class_attr_value_type = nested_class_dict.get("type", None)

            if class_attr_value_type == value_type:
                self.set_service_attribute_value_by_path(path, nested_json_dict)
            else:
                logger.info(
                    "Attribute type of '%s' changed from '%s' to "
                    "'%s'. Ignoring value from JSON file...",
                    path,
                    value_type,
                    class_attr_value_type,
                )

def save_state(self) -> None:
    """Saves the DataService's current state to a JSON file defined by
    `self.filename`.
    """

    if self.filename is not None:
        with open(self.filename, "w") as f:
            json.dump(self.cache_value, f, indent=4)
    else:
        logger.debug(
            "State manager was not initialised with a filename. Skipping "
            "'save_state'..."
        )

def set_service_attribute_value_by_path(
    self,
    path: str,
    serialized_value: SerializedObject,
) -> None:
    """Sets the value of an attribute in the service managed by the `StateManager`
    given its path as a dot-separated string.

    This method updates the attribute specified by 'path' with 'value' only if the
    attribute is not read-only and the new value differs from the current one.
    It also handles type-specific conversions for the new value before setting it.

    Args:
        path:
            A dot-separated string indicating the hierarchical path to the
            attribute.
        serialized_value:
            The serialized representation of the new value to set for the attribute.
    """

    try:
        current_value_dict = self.cache_manager.get_value_dict_from_cache(path)
    except (SerializationPathError, KeyError):
        current_value_dict = {
            "full_access_path": path,
            "value": None,
            "type": "None",
            "doc": None,
            "readonly": False,
        }

    if "full_access_path" not in serialized_value:
        # Backwards compatibility for JSON files not containing the
        # full_access_path
        logger.warning(
            "The format of your JSON file is out-of-date. This might lead "
            "to unexpected errors. Please consider updating it."
        )
        serialized_value["full_access_path"] = current_value_dict[
            "full_access_path"
        ]

    # only set value when it has changed
    if self.__attr_value_has_changed(serialized_value, current_value_dict):
        self.__update_attribute_by_path(path, serialized_value)
    else:
        logger.debug("Value of attribute '%s' has not changed...", path)

async def serve(self) -> Any:
    """Starts the server. This method should be implemented as an asynchronous
    method, which means that it should be able to run concurrently with other tasks.
    """

async def post_startup(self) -> None:
    """Override this in a subclass to register custom logic after startup."""

def run(self) -> None:
    """
    Initializes the asyncio event loop and starts the server.

    This method should be called to start the server after it's been instantiated.
    """
    try:
        self._loop.run_until_complete(self.serve())
    finally:
        self._loop.close()

def get_value(self, access_path: str) -> Any:
    """Retrieve the current value of a remote attribute.

    Args:
        access_path: The dot-separated path to the attribute in the remote service.

    Returns:
        The deserialized value of the remote attribute, or None if the client is not
        connected.

    Example:
        ```python
        value = client.get_value("my_device.temperature")
        print(value)
        ```
    """

    if self._loop is not None:
        return get_value(
            sio_client=self._sio,
            loop=self._loop,
            access_path=access_path,
        )
    return None

def trigger_method(self, access_path: str, *args: Any, **kwargs: Any) -> Any:
    """Trigger a remote method with optional arguments.

    Args:
        access_path: The dot-separated path to the method in the remote service.
        *args: Positional arguments to pass to the method.
        **kwargs: Keyword arguments to pass to the method.

    Returns:
        The return value of the method call, if any.

    Example:
        ```python
        result = client.trigger_method("my_device.calibrate", timeout=5)
        print(result)
        ```
    """

    if self._loop is not None:
        return trigger_method(
            sio_client=self._sio,
            loop=self._loop,
            access_path=access_path,
            args=list(args),
            kwargs=kwargs,
        )
    return None

def update_value(self, access_path: str, new_value: Any) -> Any:
    """Set a new value for a remote attribute.

    Args:
        access_path: The dot-separated path to the attribute in the remote service.
        new_value: The new value to assign to the attribute.

    Example:
        ```python
        client.update_value("my_device.power", True)
        ```
    """

    if self._loop is not None:
        update_value(
            sio_client=self._sio,
            loop=self._loop,
            access_path=access_path,
            value=new_value,
        )

def connect(self) -> None:
    """Tries connecting to the device and changes `self._connected` status
    accordingly. This method is called every `self._reconnection_wait_time` seconds
    when `self.connected` is False. Users should override this method to implement
    device-specific connection logic.
    """

def serialize(self) -> SerializedObject:
    """
    Serializes the instance into a dictionary, preserving the structure of the
    instance.

    For each attribute, method, and property, the method includes its name, type,
    value, readonly status, and documentation if any in the resulting dictionary.
    Attributes and methods starting with an underscore are ignored.

    For nested DataService instances, the method serializes recursively.
    For attributes of type list, each item in the list is serialized individually.
    If an item in the list is an instance of DataService, it is serialized
    recursively.

    Returns:
        dict: The serialized instance.
    """
    return Serializer.serialize_object(self)

def serialize(self) -> SerializedObject:
    """
    Serializes the instance into a dictionary, preserving the structure of the
    instance.

    For each attribute, method, and property, the method includes its name, type,
    value, readonly status, and documentation if any in the resulting dictionary.
    Attributes and methods starting with an underscore are ignored.

    For nested DataService instances, the method serializes recursively.
    For attributes of type list, each item in the list is serialized individually.
    If an item in the list is an instance of DataService, it is serialized
    recursively.

    Returns:
        dict: The serialized instance.
    """
    return Serializer.serialize_object(self)