Source code for ferrosoft.apps.reporting.websockets.updater

#  Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from abc import ABC

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

from ferrosoft.apps.ferrobase.consumers import EntityIdRequired
from ferrosoft.apps.ferrobase.frontend import FrontendUIUpdater




class PassTroughUpdater(ABC, FrontendUIUpdater):

    _MESSAGE_TYPE = "pass_through"



    def send_update(self, content: str):
        async_to_sync(get_channel_layer().group_send)(
            self._group_name,
            {
                "type": self._MESSAGE_TYPE,
                "state": "updated",
                "content": content,
            },
        )




    def send_result(self, content: str):
        async_to_sync(get_channel_layer().group_send)(
            self._group_name,
            {
                "type": self._MESSAGE_TYPE,
                "state": "finished",
                "content": content,
            },
        )






class AutoGroupPassTrough(PassTroughUpdater):
    def __init__(self, entity_id: str | None = None):
        super().__init__()
        if entity_id is None:
            raise EntityIdRequired
        self._group_name = "%s_%s" % (self.__class__.__name__, entity_id)





class ReportGeneratorUpdater(AutoGroupPassTrough):
    pass