-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathdatalink.py
165 lines (138 loc) · 4.58 KB
/
datalink.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
from dataclasses import dataclass
import dataclasses
from typing import Protocol
from zabbix_client import NUMERIC_FLOAT, NUMERIC_UNSIGNED, TEXT, TemplateItem
class HostProto(Protocol):
name: str
tags: dict
prefix: str
# Parsing stats
def stats(self) -> dict:
...
# For creating a template automatically in Zabbix
# TODO: Figure out better way to define templates
@staticmethod
def build_template():
...
# TODO: One day, I bet that I'll realize that all of these classes can
# be used for any UISP object. I think that Stats, plus Identification,
# plus a couple other things will equal a Site, and then a DataLink
# can be two Sites.
# It is already becoming the case that the names of these classes
# aren't really super accurate. a Site is just a way to refer to half of a
# DataLink, among other things.
# I'd like to actually try adding some other thing before committing to it, though.
# Becuase there's a pretty high chance we might not care. Maybe if we add routers.
@dataclass
class DataLinkStatistics:
rxRate: int
txRate: int
downlinkCapacity: int
uplinkCapacity: int
downlinkUtilization: float
uplinkUtilization: float
signalLocal: int
signalRemote: int
signalChain0: int
signalChain1: int
signalRemoteChain0: int
signalRemoteChain1: int
linkScore: float
score: int
scoreMax: int
airTimeScore: int
linkScoreHint: str
theoreticalDownlinkCapacity: int
theoreticalUplinkCapacity: int
@dataclass
class DataLinkDeviceID:
id: str
name: str
displayName: str
model: str
type: str
category: str
role: str
authorized: bool
class DataLink(HostProto):
name: str
tags: dict
signal: int
frequency: int
from_stats: DataLinkStatistics
to_stats: DataLinkStatistics
prefix = "uisp2zabbix.p2p"
# Use the JSON blob to set up the DataLink object
def __init__(self, link_json):
self.name = link_json["ssid"].strip()
self.signal = link_json["signal"]
self.frequency = link_json["frequency"]
self.tags = {
"from": link_json["from"]["site"]["identification"]["name"],
"to": link_json["to"]["site"]["identification"]["name"],
"from_dev": link_json["from"]["device"]["identification"]["model"],
"to_dev": link_json["to"]["device"]["identification"]["model"],
}
self.from_stats = DataLinkStatistics(
**link_json["from"]["interface"]["statistics"]
)
self.to_stats = DataLinkStatistics(**link_json["to"]["interface"]["statistics"])
def stats(self):
stats = {}
# Send top-level signal number
stats["signal"] = self.signal
# Send top-level frequency, multiplied by a million
# to get it into Hz so that Zabbix's units work without
# pre-processing.
stats["frequency"] = self.frequency * 1000000
# Send from stats
for k, v in self.from_stats.__dict__.items():
stats[f"from_{k}"] = v
# Send to stats
for k, v in self.to_stats.__dict__.items():
stats[f"to_{k}"] = v
return stats
@staticmethod
def build_template():
# Set up the items in the new template
items = []
# Add an item for overall signal (provided in the blob)
items.append(
TemplateItem(
"signal",
f"{DataLink.prefix}.signal",
NUMERIC_FLOAT,
"dB",
)
)
# Add an item for frequency (provided in the blob)
items.append(
TemplateItem(
"frequency",
f"{DataLink.prefix}.frequency",
NUMERIC_UNSIGNED,
"Hz",
)
)
for field in dataclasses.fields(DataLinkStatistics):
if field.type == float or field.type == int:
t = NUMERIC_FLOAT
if "signal" in field.name:
u = "dB"
elif "Rate" in field.name or "Capacity" in field.name:
u = "bps" # TODO: Is this bps or Kbps?
else:
u = ""
else:
t = TEXT
u = ""
for direction in ["from", "to"]:
items.append(
TemplateItem(
f"{direction}_{field.name}",
f"{DataLink.prefix}.{direction}_{field.name}",
t,
u,
)
)
return items