openwebrx-clone/owrx/property/__init__.py

from abc import ABC, abstractmethod
import logging

logger = logging.getLogger(__name__)


class Subscription(object):
    def __init__(self, subscriptee, name, subscriber):
        self.subscriptee = subscriptee
        self.name = name
        self.subscriber = subscriber

    def getName(self):
        return self.name

    def call(self, *args, **kwargs):
        self.subscriber(*args, **kwargs)

    def cancel(self):
        self.subscriptee.unwire(self)


class PropertyManager(ABC):
    def __init__(self):
        self.subscribers = []

    @abstractmethod
    def __getitem__(self, item):
        pass

    @abstractmethod
    def __setitem__(self, key, value):
        pass

    @abstractmethod
    def __contains__(self, item):
        pass

    @abstractmethod
    def __dict__(self):
        pass

    @abstractmethod
    def keys(self):
        pass

    def filter(self, *props):
        return PropertyFilter(self, *props)

    def wire(self, callback):
        sub = Subscription(self, None, callback)
        self.subscribers.append(sub)
        return sub

    def wireProperty(self, name, callback):
        sub = Subscription(self, name, callback)
        self.subscribers.append(sub)
        if name in self:
            sub.call(self[name])
        return sub

    def unwire(self, sub):
        try:
            self.subscribers.remove(sub)
        except ValueError:
            # happens when already removed before
            pass
        return self

    def _fireCallbacks(self, name, value):
        for c in self.subscribers:
            try:
                if c.getName() is None:
                    c.call(name, value)
                elif c.getName() == name:
                    c.call(value)
            except Exception as e:
                logger.exception(e)


class PropertyLayer(PropertyManager):
    def __init__(self):
        super().__init__()
        self.properties = {}

    def __contains__(self, name):
        return name in self.properties

    def __getitem__(self, name):
        return self.properties[name]

    def __setitem__(self, name, value):
        if name in self.properties and self.properties[name] == value:
            return
        self.properties[name] = value
        self._fireCallbacks(name, value)

    def __dict__(self):
        return {k: v for k, v in self.properties.items()}

    def keys(self):
        return self.properties.keys()


class PropertyFilter(PropertyManager):
    def __init__(self, pm: PropertyManager, *props: str):
        super().__init__()
        self.pm = pm
        self.props = props
        self.pm.wire(self.receiveEvent)

    def receiveEvent(self, name, value):
        if name not in self.props:
            return
        self._fireCallbacks(name, value)

    def __getitem__(self, item):
        if item not in self.props:
            raise KeyError(item)
        return self.pm.__getitem__(item)

    def __setitem__(self, key, value):
        if key not in self.props:
            raise KeyError(key)
        return self.pm.__setitem__(key, value)

    def __contains__(self, item):
        if item not in self.props:
            return False
        return self.pm.__contains__(item)

    def __dict__(self):
        return {k: v for k, v in self.pm.__dict__().items() if k in self.props}

    def keys(self):
        return [k for k in self.pm.keys() if k in self.props]


class PropertyStack(PropertyManager):
    def __init__(self):
        super().__init__()
        self.layers = []

    def addLayer(self, priority: int, pm: PropertyManager):
        """
        highest priority = 0
        """
        self._fireChanges(self._addLayer(priority, pm))

    def _addLayer(self, priority: int, pm: PropertyManager):
        changes = {}
        for key in pm.keys():
            if key not in self or self[key] != pm[key]:
                changes[key] = pm[key]

        def eventClosure(name, value):
            self.receiveEvent(pm, name, value)

        sub = pm.wire(eventClosure)

        self.layers.append({"priority": priority, "props": pm, "sub": sub})

        return changes

    def removeLayer(self, pm: PropertyManager):
        for layer in self.layers:
            if layer["props"] == pm:
                self._fireChanges(self._removeLayer(layer))

    def _removeLayer(self, layer):
        layer["sub"].cancel()
        self.layers.remove(layer)
        changes = {}
        pm = layer["props"]
        for key in pm.keys():
            if key in self:
                if self[key] != pm[key]:
                    changes[key] = self[key]
            else:
                changes[key] = None
        return changes

    def replaceLayer(self, priority: int, pm: PropertyManager):
        layers = [x for x in self.layers if x["priority"] == priority]
        changes = {}
        if layers:
            changes = self._removeLayer(layers[0])

        for k, v in self._addLayer(priority, pm).items():
            changes[k] = v

        self._fireChanges(changes)

    def _fireChanges(self, changes):
        for k, v in changes.items():
            self._fireCallbacks(k, v)

    def receiveEvent(self, layer, name, value):
        if layer != self._getTopLayer(name):
            return
        self._fireCallbacks(name, value)

    def _getTopLayer(self, item):
        layers = [la["props"] for la in sorted(self.layers, key=lambda l: l["priority"])]
        for m in layers:
            if item in m:
                return m
        # return top layer by default
        if layers:
            return layers[0]

    def __getitem__(self, item):
        layer = self._getTopLayer(item)
        return layer.__getitem__(item)

    def __setitem__(self, key, value):
        layer = self._getTopLayer(key)
        return layer.__setitem__(key, value)

    def __contains__(self, item):
        layer = self._getTopLayer(item)
        if layer:
            return layer.__contains__(item)
        return False

    def __dict__(self):
        return {k: self.__getitem__(k) for k in self.keys()}

    def keys(self):
        return set([key for l in self.layers for key in l["props"].keys()])