Skip to content

naneos.gui.model

What the tray menu says about the devices, without Qt.

The tray asks the manager for the connected devices and remembers when each of them last delivered a data point, so the menu can show

P2 Pro  SN8764  USB  1 s
P2      SN8123  BLE  2 s

DeviceLike

Bases: Protocol

The part of PartectorDevice the menu needs.

Source code in src/naneos/gui/model.py 
27
28
29
30
31
32
33
34
35
36
37
class DeviceLike(Protocol):
    """The part of PartectorDevice the menu needs."""

    @property
    def serial_number(self) -> int | None: ...

    @property
    def device_type(self) -> DeviceType | None: ...

    @property
    def connection_type(self) -> ConnectionType: ...

LastSeen

When each serial number last delivered a data point (monotonic clock).

The arrival time is used, not the timestamp in the data point, so a device with a wrong clock still shows how long it has been silent.

Source code in src/naneos/gui/model.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class LastSeen:
    """When each serial number last delivered a data point (monotonic clock).

    The arrival time is used, not the timestamp in the data point, so a device
    with a wrong clock still shows how long it has been silent.
    """

    def __init__(self, clock: Callable[[], float] = time.monotonic) -> None:
        self._clock = clock
        self._seen: dict[int, float] = {}

    def note(self, serial_number: int | None) -> None:
        if serial_number is not None:
            self._seen[serial_number] = self._clock()

    def drain(self, points: "queue.Queue[P]") -> int:
        """Take every waiting point off the live queue. Returns how many."""
        taken = 0
        while taken < _MAX_DRAIN:
            try:
                point = points.get_nowait()
            except queue.Empty:
                break
            self.note(point.serial_number)
            taken += 1
        return taken

    def age(self, serial_number: int | None) -> float | None:
        """Seconds since the last point, None if there was none yet."""
        if serial_number is None or serial_number not in self._seen:
            return None
        return max(0.0, self._clock() - self._seen[serial_number])

    def forget_except(self, serial_numbers: Iterable[int | None]) -> None:
        """Drop devices that are gone, so a device that comes back starts at "–"."""
        keep = {sn for sn in serial_numbers if sn is not None}
        self._seen = {sn: at for sn, at in self._seen.items() if sn in keep}

age(serial_number)

Seconds since the last point, None if there was none yet.

Source code in src/naneos/gui/model.py 
76
77
78
79
80
def age(self, serial_number: int | None) -> float | None:
    """Seconds since the last point, None if there was none yet."""
    if serial_number is None or serial_number not in self._seen:
        return None
    return max(0.0, self._clock() - self._seen[serial_number])

drain(points)

Take every waiting point off the live queue. Returns how many.

Source code in src/naneos/gui/model.py 
64
65
66
67
68
69
70
71
72
73
74
def drain(self, points: "queue.Queue[P]") -> int:
    """Take every waiting point off the live queue. Returns how many."""
    taken = 0
    while taken < _MAX_DRAIN:
        try:
            point = points.get_nowait()
        except queue.Empty:
            break
        self.note(point.serial_number)
        taken += 1
    return taken

forget_except(serial_numbers)

Drop devices that are gone, so a device that comes back starts at "–".

Source code in src/naneos/gui/model.py 
82
83
84
85
def forget_except(self, serial_numbers: Iterable[int | None]) -> None:
    """Drop devices that are gone, so a device that comes back starts at "–"."""
    keep = {sn for sn in serial_numbers if sn is not None}
    self._seen = {sn: at for sn, at in self._seen.items() if sn in keep}

sort_devices(devices)

By serial number; a device that has not told its number yet goes last.

Source code in src/naneos/gui/model.py 
105
106
107
108
109
110
def sort_devices(devices: Iterable[D]) -> list[D]:
    """By serial number; a device that has not told its number yet goes last."""
    return sorted(
        devices,
        key=lambda d: (d.serial_number is None, d.serial_number if d.serial_number else 0),
    )