Source code for ferrosoft.apps.ferrobase.frontend

"""Primitives for pushing real-time UI updates over WebSocket to the frontend.

Provides a hierarchy of serialisable element types (``Element``, ``UpdateSpec``,
etc.) and ``FrontendUIUpdater``, which broadcasts update specs to a Django
Channels channel group so connected clients can apply them without a full page
reload.
"""
import enum
from collections import namedtuple
from typing import Sequence, List

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from rest_framework import serializers


Position = namedtuple("Position", ["x", "y"])
Position.__doc__ = "A 2D coordinate used to position UI elements (x, y)."




class Attr:
    """
    Represents HTML element attribute.
    """

    def __init__(self, name: str, value: str | bool = None):
        self.name = name
        self.value = value





class Namespace(enum.StrEnum):
    """XML/HTML namespace identifiers used when serialising mixed-namespace elements."""

    SVG = "http://www.w3.org/2000/svg"
    HTML = "html"





class Element:
    """
    Element represents HTML elements or XML elements of certain namespace (i.e. SVG).
    """

    def __init__(
        self,
        name: str,
        *args,
        **kwargs,
    ):
        """
        Args:
            name (str): Tag name
            namespace (str): See Namespace enum for possible values. If no namespace is given,
                the serialized representation does not include the field. This is fine if all
                the elements are of uniform namespace. However, if for example SVG and HTML
                elements are mixed together, a namespace must be specified for those elements
                where their namespace differs from the factory namespace default.
            text_content (str): Text content
            args: Accepted are Element and Attr objects, allowing child elements and attributes to be declared.

        """
        self.name = name
        self.text_content = kwargs.pop("text_content", None)
        self.namespace = kwargs.pop("namespace", None)
        self.children = [item for item in args if isinstance(item, Element)]
        self.attrs = {attr.name: attr.value for attr in args if isinstance(attr, Attr)}





class ElementSerializer(serializers.Serializer):
    """DRF serialiser for ``Element`` that outputs camelCase JSON for the frontend."""

    attrs = serializers.DictField()
    children = serializers.SerializerMethodField()
    name = serializers.CharField()
    namespace = serializers.CharField()
    text_content = serializers.CharField(allow_blank=True, allow_null=True)



    @staticmethod
    def get_children(obj):
        if not obj.children:
            return []
        return [ElementSerializer(child).data for child in obj.children]




    def to_representation(self, instance):
        rep = super().to_representation(instance)

        # Want camel case.
        rep["textContent"] = rep["text_content"]
        del rep["text_content"]

        # Remove some fields if they are empty.
        if rep["namespace"] is None:
            del rep["namespace"]

        if rep["textContent"] is None:
            del rep["textContent"]

        if len(rep["children"]) == 0:
            del rep["children"]

        if len(rep["attrs"]) == 0:
            del rep["attrs"]

        return rep






class Transition:
    """A named UI transition with an optional detail payload."""

    def __init__(self, transition: str, detail: dict = None):
        self.transition = transition
        self.detail = detail





class TransitionSerializer(serializers.Serializer):
    """DRF serialiser for ``Transition``."""

    transition = serializers.CharField()
    detail = serializers.DictField()





class AppendSpec:
    """Specification for appending new elements to a CSS-selector target."""

    def __init__(self, selector: str, elements: Sequence[Element]):
        self.selector = selector
        self.elements = elements





class AppendSpecSerializer(serializers.Serializer):
    """DRF serialiser for ``AppendSpec``."""

    selector = serializers.CharField()
    elements = ElementSerializer(many=True)





class DeleteSpec:
    """Specification for removing all elements matching a CSS selector."""

    def __init__(self, selector: str):
        self.selector = selector





class DeleteSpecSerializer(serializers.Serializer):
    """DRF serialiser for ``DeleteSpec``."""

    selector = serializers.CharField()





class UpdateSpec:
    """Aggregate specification describing a complete frontend UI update.

    Combines element replacements, appends, and deletions into a single
    payload that ``FrontendUIUpdater.update`` broadcasts to connected clients.
    """

    def __init__(
        self,
        replace_children: List[Element] = None,
        append_children: List[AppendSpec] = None,
        delete_elements: List[DeleteSpec] = None,
    ):
        self.replace_children = replace_children or []
        self.append_children = append_children or []
        self.delete_elements = delete_elements or []





class UpdateSpecSerializer(serializers.Serializer):
    """DRF serialiser for ``UpdateSpec`` that outputs camelCase JSON."""

    replace_children = serializers.SerializerMethodField()
    append_children = serializers.SerializerMethodField()
    delete_elements = serializers.SerializerMethodField()



    @staticmethod
    def get_replace_children(obj):
        if not obj.replace_children:
            return []
        return [ElementSerializer(child).data for child in obj.replace_children]




    @staticmethod
    def get_append_children(obj):
        if not obj.append_children:
            return []
        return [AppendSpecSerializer(spec).data for spec in obj.append_children]




    @staticmethod
    def get_delete_elements(obj):
        if not obj.delete_elements:
            return []
        return [DeleteSpecSerializer(spec).data for spec in obj.delete_elements]




    def to_representation(self, instance):
        rep = super().to_representation(instance)
        # want camel case for both fields
        rep["replaceChildren"] = rep["replace_children"]
        rep["appendChildren"] = rep["append_children"]
        rep["deleteElements"] = rep["delete_elements"]
        del rep["replace_children"]
        del rep["append_children"]
        del rep["delete_elements"]
        return rep






class FrontendUIUpdater:
    """
    Update frontend UI with new or replacing elements.
    """

    def __init__(self, *args, **kwargs):
        self._group_name = None

    @property
    def group_name(self):
        """Django Channels group name this updater broadcasts to."""
        if self._group_name is None:
            raise Exception("_group_name property must be set")
        return self._group_name



    def update(self, spec: UpdateSpec, transition: Transition | None = None):
        """Broadcast an update specification to all clients in the channel group.

        Args:
            spec: The UI update to apply. Elements targeted by
                ``replace_children`` must carry an ``id`` attribute.
            transition: Optional named transition to accompany the update.
        """
        channel_layer = get_channel_layer()

        response = {
            "type": "frontend_update",
            "spec": UpdateSpecSerializer(spec).data,
        }
        if transition is not None:
            response["transition"] = TransitionSerializer(transition).data

        async_to_sync(channel_layer.group_send)(self.group_name, response)