<commit_before>from typing import Any, Callable, Dict, Tuple, Type

import time
from hedgehog.protocol import Header, Message
from hedgehog.protocol.errors import UnsupportedCommandError, FailedCommandError
from hedgehog.protocol.messages import ack, io, analog, digital, motor, servo
from hedgehog.protocol.proto.subscription_pb2 import Subscription
from hedgehog.utils.zmq.timer import TimerDefinition

from . import CommandHandler, command_handlers
from ..hardware import HardwareAdapter
from ..hedgehog_server import HedgehogServerActor


class SubscriptionInfo(object):
    def __init__(self, server: HedgehogServerActor, ident: Header, subscription: Subscription) -> None:
        self.server = server
        self.ident = ident
        self.subscription = subscription
        self.counter = 0

        self._last_time = None  # type: float
        self._last_value = None  # type: Any

        self.extra = None  # type: Any

    @property
    def last_time(self) -> float:
        return self._last_time

    @property
    def last_value(self) -> Any:
        return self._last_value

    @last_value.setter
    def last_value(self, value: Any) -> None:
        self._last_time = time.time()
        self._last_value = value

    def should_send(self, value: Any) -> bool:
        if value == self.last_value:
            return False
        if self.subscription.timeout is not None and self.last_time is not None and time.time() < self.last_time + self.subscription.timeout / 1000:
            return False
        return True

    def send_async(self, *msgs: Message) -> None:
        self.server.send_async(self.ident, *msgs)


class SubscriptionManager(object):
    def __init__(self) -> None:
        self.subscriptions = {}  # type: Dict[Tuple[Header, int], SubscriptionInfo]

    def subscribe(self, server: HedgehogServerActor, ident: Header, subscription: Subscription) -> None:
        # TODO incomplete
        key = (ident, subscription.timeout)

        if subscription.subscribe:
            if key not in self.subscriptions:
                info = SubscriptionInfo(server, ident, subscription)
                self.subscriptions[key] = info
                self.handle_subscribe(info)
            else:
                info = self.subscriptions[key]
            info.counter += 1
        else:
            try:
                info = self.subscriptions[key]
            except KeyError:
                raise FailedCommandError("can't cancel nonexistent subscription")
            else:
                info.counter -= 1
                if info.counter == 0:
                    self.handle_unsubscribe(info)
                    del self.subscriptions[key]

    def handle_subscribe(self, info: SubscriptionInfo) -> None:
        pass

    def handle_unsubscribe(self, info: SubscriptionInfo) -> None:
        pass


class _HWHandler(object):
    def __init__(self, adapter: HardwareAdapter) -> None:
        self.adapter = adapter
        self.subscription_handlers = {}  # type: Dict[Type[Message], SubscriptionManager]

    def subscribe(self, server: HedgehogServerActor, ident: Header, msg: Type[Message], subscription: Subscription) -> None:
        self.subscription_handlers[msg].subscribe(server, ident, subscription)


class _IOHandler(_HWHandler):
    def __command_subscription_manager(self) -> SubscriptionManager:
        outer_self = self

        class Extra(object):
            timer = None  # type: TimerDefinition

        class Mgr(SubscriptionManager):
            def __init__(self) -> None:
                super(Mgr, self).__init__()

            def handle_subscribe(self, info: SubscriptionInfo) -> None:
                info.extra = Extra()
                info.extra.timer = info.server.timer.register(info.subscription.timeout / 1000, lambda: self.handle_update(info))

            def handle_unsubscribe(self, info: SubscriptionInfo) -> None:
                info.server.timer.unregister(info.extra.timer)

            def handle_update(self, info: SubscriptionInfo) -> None:
                try:
                    flags, = outer_self.command
                except TypeError:
                    pass
                else:
                    if info.should_send(flags):
                        info.last_value = flags
                        info.send_async(io.CommandUpdate(outer_self.port, flags, info.subscription))

        return Mgr()

    def __analog_subscription_manager(self) -> SubscriptionManager:
        outer_self = self

        class Extra(object):
            timer = None  # type: TimerDefinition

        class Mgr(SubscriptionManager):
            def __init__(self) -> None:
                super(Mgr, self).__init__()

            def handle_subscribe(self, info: SubscriptionInfo) -> None:
                info.extra = Extra()
                info.extra.timer = info.server.timer.register(info.subscription.timeout / 1000, lambda: self.handle_update(info))

            def handle_unsubscribe(self, info: SubscriptionInfo) -> None:
                info.server.timer.unregister(info.extra.timer)

            def handle_update(self, info: SubscriptionInfo) -> None:
                value = outer_self.analog_value
                if info.should_send(value):
                    info.last_value = value
                    info.send_async(analog.Update(outer_self.port, value, info.subscription))

        return Mgr()

    def __init__(self, adapter: HardwareAdapter, port: int) -> None:
        super(_IOHandler, self).__init__(adapter)
        self.port = port
        self.command = None  # type: Tuple[int]

        self.subscription_handlers[io.CommandSubscribe] = self.__command_subscription_manager()
        self.subscription_handlers[analog.Subscribe] = self.__analog_subscription_manager()

    def action(self, flags: int) -> None:
        self.adapter.set_io_state(self.port, flags)
        self.command = flags,

    @property<commit_msg>shift logic into SubscriptionInfo subclasses<commit_after>from typing import Any, Callable, Dict, Tuple, Type

import time
from hedgehog.protocol import Header, Message
from hedgehog.protocol.errors import UnsupportedCommandError, FailedCommandError
from hedgehog.protocol.messages import ack, io, analog, digital, motor, servo
from hedgehog.protocol.proto.subscription_pb2 import Subscription
from hedgehog.utils.zmq.timer import TimerDefinition

from . import CommandHandler, command_handlers
from ..hardware import HardwareAdapter
from ..hedgehog_server import HedgehogServerActor


class SubscriptionInfo(object):
    def __init__(self, server: HedgehogServerActor, ident: Header, subscription: Subscription) -> None:
        self.server = server
        self.ident = ident
        self.subscription = subscription
        self.counter = 0

        self._last_time = None  # type: float
        self._last_value = None  # type: Any

    @property
    def last_time(self) -> float:
        return self._last_time

    @property
    def last_value(self) -> Any:
        return self._last_value

    @last_value.setter
    def last_value(self, value: Any) -> None:
        self._last_time = time.time()
        self._last_value = value

    def should_send(self, value: Any) -> bool:
        if value == self.last_value:
            return False
        if self.subscription.timeout is not None and self.last_time is not None and time.time() < self.last_time + self.subscription.timeout / 1000:
            return False
        return True

    def send_async(self, *msgs: Message) -> None:
        self.server.send_async(self.ident, *msgs)

    def handle_subscribe(self):
        self.counter += 1

    def handle_unsubscribe(self):
        self.counter -= 1


class SubscriptionManager(object):
    def __init__(self, SubscriptionType: Type[SubscriptionInfo]) -> None:
        self.SubscriptionType = SubscriptionType
        self.subscriptions = {}  # type: Dict[Tuple[Header, int], SubscriptionInfo]

    def subscribe(self, server: HedgehogServerActor, ident: Header, subscription: Subscription) -> None:
        # TODO incomplete
        key = (ident, subscription.timeout)

        if subscription.subscribe:
            if key not in self.subscriptions:
                info = self.SubscriptionType(server, ident, subscription)
                self.subscriptions[key] = info
            else:
                info = self.subscriptions[key]
            info.handle_subscribe()
        else:
            try:
                info = self.subscriptions[key]
            except KeyError:
                raise FailedCommandError("can't cancel nonexistent subscription")
            else:
                info.handle_unsubscribe()
                if info.counter == 0:
                    del self.subscriptions[key]


class _HWHandler(object):
    def __init__(self, adapter: HardwareAdapter) -> None:
        self.adapter = adapter
        self.subscription_handlers = {}  # type: Dict[Type[Message], SubscriptionManager]

    def subscribe(self, server: HedgehogServerActor, ident: Header, msg: Type[Message], subscription: Subscription) -> None:
        self.subscription_handlers[msg].subscribe(server, ident, subscription)


class _IOHandler(_HWHandler):
    def __command_subscription_manager(self) -> SubscriptionManager:
        outer_self = self

        class Info(SubscriptionInfo):
            def __init__(self, *args, **kwargs) -> None:
                super(Info, self).__init__(*args, **kwargs)
                self.timer = None  # type: TimerDefinition

            def handle_subscribe(self) -> None:
                if self.counter == 0:
                    self.timer = self.server.timer.register(self.subscription.timeout / 1000, self.handle_update)

                super(Info, self).handle_subscribe()

            def handle_unsubscribe(self) -> None:
                super(Info, self).handle_unsubscribe()

                if self.counter == 0:
                    self.server.timer.unregister(self.timer)

            def handle_update(self) -> None:
                try:
                    flags, = outer_self.command
                except TypeError:
                    pass
                else:
                    if self.should_send(flags):
                        self.last_value = flags
                        self.send_async(io.CommandUpdate(outer_self.port, flags, self.subscription))

        return SubscriptionManager(Info)

    def __analog_subscription_manager(self) -> SubscriptionManager:
        outer_self = self

        class Info(SubscriptionInfo):
            def __init__(self, *args, **kwargs) -> None:
                super(Info, self).__init__(*args, **kwargs)
                self.timer = None  # type: TimerDefinition

            def handle_subscribe(self) -> None:
                if self.counter == 0:
                    self.timer = self.server.timer.register(self.subscription.timeout / 1000, self.handle_update)

                super(Info, self).handle_subscribe()

            def handle_unsubscribe(self) -> None:
                super(Info, self).handle_unsubscribe()

                if self.counter == 0:
                    self.server.timer.unregister(self.timer)

            def handle_update(self) -> None:
                value = outer_self.analog_value
                if self.should_send(value):
                    self.last_value = value
                    self.send_async(analog.Update(outer_self.port, value, self.subscription))

        return SubscriptionManager(Info)

    def __init__(self, adapter: HardwareAdapter, port: int) -> None:
        super(_IOHandler, self).__init__(adapter)
        self.port = port
        self.command = None  # type: Tuple[int]

        self.subscription_handlers[io.CommandSubscribe] = self.__command_subscription_manager()
        self.subscription_handlers[analog.Subscribe] = self.__analog_subscription_manager()

    def action(self, flags: int) -> None:
        self.adapter.set_io_state(self.port, flags)
        self.command = flags,

    @property
{'fraction_non_alphanumeric': 0.09013732833957554, 'fraction_numerical': 0.003162713275072826, 'mean_word_length': 2.296570644718793, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1064323', 'n_tokens_mistral': 3297, 'n_tokens_neox': 3116, 'n_words': 759}