node_monitor.py

Go to the documentation of this file.

#
# Copyright (C) 2014-2015  UAVCAN Development Team  <uavcan.org>
#
# This software is distributed under the terms of the MIT License.
#
# Author: Ben Dyer <ben_dyer@mac.com>
#         Pavel Kirienko <pavel.kirienko@zubax.com>
#
 
from __future__ import division, absolute_import, print_function, unicode_literals
import time
from logging import getLogger
import pyuavcan_v0
 
 
logger = getLogger(__name__)
 
 
class NodeMonitor(object):
    TIMEOUT = pyuavcan_v0.protocol.NodeStatus().OFFLINE_TIMEOUT_MS / 1000  # @UndefinedVariable
    TRANSFER_PRIORITY = pyuavcan_v0.TRANSFER_PRIORITY_LOWEST - 1
    MIN_RETRY_INTERVAL = 0.5
    MAX_RETRIES = 10
 
    class Entry:
        def __init__(self):
            self.node_id = None
            self.status = None
            self.info = None
            self.monotonic_timestamp = None
            self._remaining_retries = NodeMonitor.MAX_RETRIES
 
        @property
        def discovered(self):
            return self.info is not None or self._remaining_retries <= 0
 
        def _update_from_status(self, e):
            self.monotonic_timestamp = e.transfer.ts_monotonic
            self.node_id = e.transfer.source_node_id
            if self.status and e.message.uptime_sec < self.status.uptime_sec:
                self._remaining_retries = NodeMonitor.MAX_RETRIES
                self.info = None
            self.status = e.message
            if self.info:
                self.info.status = self.status
 
        def _update_from_info(self, e):
            self._remaining_retries = NodeMonitor.MAX_RETRIES
            self.monotonic_timestamp = e.transfer.ts_monotonic
            self.node_id = e.transfer.source_node_id
            self.status = e.response.status
            self.info = e.response
 
        def _register_retry(self):
            assert self._remaining_retries > 0
            self._remaining_retries -= 1
 
        def __str__(self):
            return '%d:%s' % (self.node_id, self.info if self.info else self.status)
 
        __repr__ = __str__
 
    class UpdateEvent:
        EVENT_ID_NEW = 'new'
        EVENT_ID_INFO_UPDATE = 'info_update'
        EVENT_ID_OFFLINE = 'offline'
 
        def __init__(self, entry, event_id):
            self.entry = entry
            self.event_id = event_id
 
        def __str__(self):
            return self.event_id + ':' + str(self.entry)
 
        __repr__ = __str__
 
    class UpdateHandlerRemover:
        def __init__(self, remover):
            self._remover = remover
 
        def remove(self):
            self._remover()
 
        def try_remove(self):
            try:
                self._remover()
            except ValueError:
                pass
 
    def __init__(self, node):
        self._update_callbacks = []
        self._handle = node.add_handler(pyuavcan_v0.protocol.NodeStatus, self._on_node_status)  # @UndefinedVariable
        self._registry = {}  # {node_id: Entry}
        self._timer = node.periodic(1, self._remove_stale)
 
    def add_update_handler(self, callback):
        """
        Args:
            callback:   The specified callback will be invoked when:
                        - A new node appears
                        - Node info for an existing node gets updated
                        - Node goes offline
        Returns:    Call remove() or try_remove() on the returned object to unregister the handler.
        """
        self._update_callbacks.append(callback)
        return self.UpdateHandlerRemover(lambda: self._update_callbacks.remove(callback))
 
    def _call_event_handlers(self, event):
        for cb in self._update_callbacks:
            cb(event)
 
    def exists(self, node_id):
        """
        Args:
            node_id:    Returns True if the given node ID exists, false otherwise
        """
        return node_id in self._registry
 
    def get(self, node_id):
        """
        Args:
            node_id:    Returns an Entry instance for the given node ID.
                        If the requested node ID does not exist, throws KeyError.
        """
        if (self._registry[node_id].monotonic_timestamp + self.TIMEOUT) < time.monotonic():
            self._call_event_handlers(self.UpdateEvent(self._registry[node_id],
                                                       self.UpdateEvent.EVENT_ID_OFFLINE))
            del self._registry[node_id]
        return self._registry[node_id]
 
    def get_all_node_id(self):
        """Returns a generator or an iterable containing all currently active node ID."""
        return self._registry.keys()
 
    def find_all(self, predicate):
        """Returns a generator that produces a sequence of Entry objects for which the predicate returned True.
        Args:
            predicate:  A callable that returns a value coercible to bool.
        """
        for _nid, entry in self._registry.items():
            if predicate(entry):
                yield entry
 
    def are_all_nodes_discovered(self):
        """Reports whether there are nodes whose node info is still unknown."""
        undiscovered = self.find_all(lambda e: not e.discovered)
        return len(list(undiscovered)) == 0
 
    def close(self):
        """Stops the instance. The registry will not be cleared."""
        self._handle.remove()
        self._timer.remove()
 
    def _remove_stale(self):
        for nid, e in list(self._registry.items())[:]:
            if (e.monotonic_timestamp + self.TIMEOUT) < time.monotonic():
                del self._registry[nid]
                self._call_event_handlers(self.UpdateEvent(e, self.UpdateEvent.EVENT_ID_OFFLINE))
 
    def _on_node_status(self, e):
        node_id = e.transfer.source_node_id
 
        try:
            entry = self.get(node_id)
            new_entry = False
        except KeyError:
            entry = self.Entry()
            entry._info_requested_at = 0
            self._registry[node_id] = entry
            new_entry = True
 
        # noinspection PyProtectedMember
        entry._update_from_status(e)
        if new_entry:
            self._call_event_handlers(self.UpdateEvent(entry, self.UpdateEvent.EVENT_ID_NEW))
 
        should_retry_now = entry.monotonic_timestamp - entry._info_requested_at > self.MIN_RETRY_INTERVAL
 
        if not entry.discovered and should_retry_now and not e.node.is_anonymous:
            entry._info_requested_at = entry.monotonic_timestamp
            # noinspection PyProtectedMember
            entry._register_retry()
            e.node.request(pyuavcan_v0.protocol.GetNodeInfo.Request(), node_id,  # @UndefinedVariable
                           priority=self.TRANSFER_PRIORITY, callback=self._on_info_response)
 
    def _on_info_response(self, e):
        if not e:
            return
 
        try:
            entry = self.get(e.transfer.source_node_id)
        except KeyError:
            entry = self.Entry()
            self._registry[e.transfer.source_node_id] = entry
 
        # noinspection PyProtectedMember
        entry._update_from_info(e)
 
        hw_unique_id = "".join(format(c, "02X") for c in e.response.hardware_version.unique_id)
        msg = (
            "[#{0:03d}:pyuavcan_v0.protocol.GetNodeInfo] " +
            "software_version.major={1:d} " +
            "software_version.minor={2:d} " +
            "software_version.vcs_commit={3:08x} " +
            "software_version.image_crc={4:016X} " +
            "hardware_version.major={5:d} " +
            "hardware_version.minor={6:d} " +
            "hardware_version.unique_id={7!s} " +
            "name={8!r}"
        ).format(
            e.transfer.source_node_id,
            e.response.software_version.major,
            e.response.software_version.minor,
            e.response.software_version.vcs_commit,
            e.response.software_version.image_crc,
            e.response.hardware_version.major,
            e.response.hardware_version.minor,
            hw_unique_id,
            e.response.name.decode()
        )
        logger.info(msg)
 
        self._call_event_handlers(self.UpdateEvent(entry, self.UpdateEvent.EVENT_ID_INFO_UPDATE))