51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
import collections
|
|
import dataclasses
|
|
import typing
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class Peer:
|
|
# is the peer a master
|
|
master: bool = dataclasses.field()
|
|
|
|
# public asymmetric key
|
|
public_key: bytes = dataclasses.field(repr=False)
|
|
# secret symmetric key
|
|
secret_key: Optional[bytes] = dataclasses.field(default=None, repr=False)
|
|
|
|
# additional public key hashes used by this peer
|
|
allowed_public_key_hashes: set[bytes] = dataclasses.field(default_factory=set, repr=False)
|
|
|
|
# list of the latest latencies values
|
|
latest_latencies: collections.deque[timedelta] = dataclasses.field(
|
|
default_factory=lambda: collections.deque(maxlen=50), # use a queue of 50 elements to only keep the latest values
|
|
repr=False
|
|
)
|
|
# when did the peer last communication with us occurred
|
|
last_interaction: datetime = dataclasses.field(default_factory=datetime.now)
|
|
|
|
|
|
def is_trusted(self, manager) -> bool:
|
|
"""
|
|
Check if the peer is trusted
|
|
:param manager: the manager
|
|
:return: is the peer trusted
|
|
"""
|
|
|
|
return manager.communication.is_peer_trusted(self.public_key)
|
|
|
|
@property
|
|
def latency(self) -> typing.Optional[timedelta]:
|
|
# check if there are data about the latency
|
|
if len(self.latest_latencies) == 0:
|
|
return None
|
|
|
|
# get a high percentile of the latency
|
|
return timedelta(seconds=typing.cast(float, np.percentile(
|
|
list(map(lambda latency: latency.total_seconds(), self.latest_latencies)),
|
|
90
|
|
)))
|