here the plugincode.
#!/usr/bin/env python3
from cmk.agent_based.v2 import (
CheckPlugin,
DiscoveryResult,
Result,
Service,
SimpleSNMPSection,
SNMPTree,
State,
StringTable,
exists,
OIDEnd,
)
# ------------------------------------------------------------
# Helpers
# ------------------------------------------------------------
def _as_int(s, default: int = 0) -> int:
try:
return int(str(s).strip())
except Exception:
return default
def _get_row0(st: StringTable):
# Scalar SNMPTree often returns one row with many columns.
# But keep it robust.
if not st:
return []
if len(st) == 1:
return list(st[0])
# fallback: treat as first column per row
return [r[0] if r else "" for r in st]
def _col(st: StringTable, idx: int, default: str = "") -> str:
try:
row0 = _get_row0(st)
return str(row0[idx]).strip()
except Exception:
return default
# ============================================================
# 1) DEVICE (Router basics) 48690.1.* (fetched in one go)
# 1.1 ID, 1.2 Name, 1.3 Model, 1.4 HWRev, 1.5 (?), 1.6 FW
# ============================================================
def parse_teltonika_device(st: StringTable) -> dict:
# unterstützt sowohl 1-Zeilen- als auch Multi-Zeilen-SNMP
flat = []
for row in st:
flat.extend(row)
def g(i):
try:
return str(flat[i])
except Exception:
return ""
return {
"id": g(0),
"name": g(1),
"model": g(2),
"hw": g(3),
"fw": g(5),
}
snmp_section_teltonika_device = SimpleSNMPSection(
name="teltonika_device",
parse_function=parse_teltonika_device,
detect=exists(".1.3.6.1.4.1.48690.1.1.0"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.1",
# IMPORTANT: without ".0" to reduce separate scalar calls
oids=["1", "2", "3", "4", "5", "6"],
),
)
def discover_teltonika_device(section_teltonika_device, section_teltonika_modeminfo) -> DiscoveryResult:
yield Service()
def check_teltonika_device(section_teltonika_device, section_teltonika_modeminfo):
d = section_teltonika_device or {}
name = d.get("name", "?")
model = d.get("model", "?")
rid = d.get("id", "?")
hw = d.get("hw", "?")
fw = d.get("fw", "?")
yield Result(
state=State.OK,
summary=f"Name={name} Model={model} ID={rid} HWRev={hw} FW={fw}",
)
check_plugin_teltonika_device = CheckPlugin(
name="teltonika_device",
service_name="Teltonika Device",
sections=["teltonika_device", "teltonika_modeminfo"],
discovery_function=discover_teltonika_device,
check_function=check_teltonika_device,
)
# ============================================================
# 2) GPS 48690.3.* (lat, lon, speed, time, fix)
# 3.1 lat, 3.2 lon, 3.3 speed, 3.4 time, 3.5 fix
# ============================================================
def parse_teltonika_gps(st: StringTable) -> dict:
flat = []
for row in st:
flat.extend(row)
def g(i):
try:
return str(flat[i])
except Exception:
return ""
return {
"lat": g(0),
"lon": g(1),
"speed": g(2),
"time": g(3),
"fix": g(4),
}
snmp_section_teltonika_gps = SimpleSNMPSection(
name="teltonika_gps",
parse_function=parse_teltonika_gps,
detect=exists(".1.3.6.1.4.1.48690.3.1.0"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.3",
oids=["1", "2", "3", "4", "5"],
),
)
def discover_teltonika_gps(section) -> DiscoveryResult:
yield Service()
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
def check_teltonika_gps(section):
lat = section.get("lat") or "?"
lon = section.get("lon") or "?"
fix = section.get("fix") or "?"
spd = section.get("speed") or "?"
tme = section.get("time") or ""
tz = ZoneInfo("Europe/Berlin")
time_txt = "?"
if tme and tme != "?":
try:
dt = datetime.strptime(tme, "%Y-%m-%d %H:%M:%S")
now_local = datetime.now(tz).replace(tzinfo=None)
# Wenn der Timestamp "in der Zukunft" liegt, ist die Teltonika-Zeitbasis bei dir off
# -> korrigieren wir automatisch um 1h oder 2h (was näher an jetzt liegt)
if dt - now_local > timedelta(minutes=10):
cand0 = dt
cand1 = dt - timedelta(hours=1)
cand2 = dt - timedelta(hours=2)
d0 = abs((now_local - cand0).total_seconds())
d1 = abs((now_local - cand1).total_seconds())
d2 = abs((now_local - cand2).total_seconds())
dt = min([(d0, cand0), (d1, cand1), (d2, cand2)], key=lambda x: x[0])[1]
time_txt = dt.strftime("%Y-%m-%d %H:%M:%S")
except Exception:
time_txt = tme
yield Result(
state=State.OK,
summary=f"GPS {lat}, {lon} (fix={fix}, speed={spd}, time={time_txt})",
)
check_plugin_teltonika_gps = CheckPlugin(
name="teltonika_gps",
service_name="Teltonika GPS",
discovery_function=discover_teltonika_gps,
check_function=check_teltonika_gps,
)
# ============================================================
# 3) MODEM/IMEI/ICCID/IMSI etc. (from 48690.2.2.1.*)
# This is parsed separately so SIM checks can include IMEI.
#
# From your walk:
# 3 IMEI
# 4 modem model (e.g. RG520N-EB)
# 5 vendor (Quectel)
# 7 modem serial (MPY...)
# 8 IMSI
# 27 ICCID
# 9 sim inserted
# ============================================================
def parse_teltonika_modeminfo(st: StringTable) -> dict:
res = {}
for i, row in enumerate(st, start=1):
cols = [str(x) for x in row]
cols += [""] * (7 - len(cols))
res[str(i)] = {
"imei": cols[0],
"modem_model": cols[1],
"vendor": cols[2],
"modem_serial": cols[3],
"imsi": cols[4],
"iccid": cols[5],
"sim_state": cols[6],
}
return res
snmp_section_teltonika_modeminfo = SimpleSNMPSection(
name="teltonika_modeminfo",
parse_function=parse_teltonika_modeminfo,
detect=exists(".1.3.6.1.4.1.48690.2.2.1.3.1"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.2.2.1",
oids=["3", "4", "5", "7", "8", "27", "9"],
),
)
# ============================================================
# 4) MOBILE (SIM status + traffic) 48690.2.2.1.*
# 13 operator, 15 conn, 16 tech, 11 reg, 12 rssi, 24 ip, 22 rx, 23 tx
# ============================================================
def parse_teltonika_mobile(st: StringTable) -> dict:
res = {}
for i, row in enumerate(st, start=1):
cols = [str(x) for x in row]
cols += [""] * (8 - len(cols))
res[str(i)] = {
"oper": cols[0],
"tech": cols[1],
"conn": cols[2],
"reg": cols[3],
"rssi": cols[4],
"ip": cols[5],
"rx": cols[6],
"tx": cols[7],
}
return res
snmp_section_teltonika_mobile = SimpleSNMPSection(
name="teltonika_mobile",
parse_function=parse_teltonika_mobile,
detect=exists(".1.3.6.1.4.1.48690.2.2.1.15.1"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.2.2.1",
# order: oper, tech, conn, reg, rssi, ip, rx, tx
oids=["13", "16", "15", "11", "12", "24", "22", "23"],
),
)
def discover_teltonika_mobile(section_teltonika_mobile, section_teltonika_modeminfo) -> DiscoveryResult:
for item in sorted((section_teltonika_mobile or {}).keys(), key=lambda x: _as_int(x, 0)):
yield Service(item=item)
def check_teltonika_mobile(item: str, section_teltonika_mobile, section_teltonika_modeminfo):
r = (section_teltonika_mobile or {}).get(item, {})
m = (section_teltonika_modeminfo or {}).get(item, {})
imei = (m.get("imei") or "").strip()
imei_txt = f"IMEI={imei} | " if imei else ""
oper = (r.get("oper") or "?").strip() or "?"
tech = (r.get("tech") or "?").strip() or "?"
conn = (r.get("conn") or "?").strip() or "?"
reg = (r.get("reg") or "?").strip() or "?"
rssi = (r.get("rssi") or "?").strip() or "?"
ip = (r.get("ip") or "?").strip() or "?"
rx = (r.get("rx") or "0").strip() or "0"
tx = (r.get("tx") or "0").strip() or "0"
state = State.OK
if "connected" not in conn.lower():
state = State.CRIT
elif "registered" not in reg.lower():
state = State.WARN
# konsistente Ausgabe: Operator Connected Tech
yield Result(
state=state,
summary=f"{imei_txt}{oper} {conn} {tech}, reg={reg}, rssi={rssi} dBm, ip={ip} RX={rx} TX={tx}",
)
check_plugin_teltonika_mobile = CheckPlugin(
name="teltonika_mobile",
service_name="Teltonika SIM %s",
sections=["teltonika_mobile", "teltonika_modeminfo"],
discovery_function=discover_teltonika_mobile,
check_function=check_teltonika_mobile,
)
# ============================================================
# 5) IF DESCR (with real ifIndex via OIDEnd)
# ============================================================
def parse_teltonika_ifdescr(st: StringTable) -> dict:
out = {}
for row in st:
if len(row) < 2:
continue
idx = _as_int(row[0], -1)
if idx < 0:
continue
out[idx] = str(row[1]).strip()
return out
snmp_section_teltonika_ifdescr = SimpleSNMPSection(
name="teltonika_ifdescr",
parse_function=parse_teltonika_ifdescr,
detect=exists(".1.3.6.1.2.1.2.2.1.2.1"),
fetch=SNMPTree(
base=".1.3.6.1.2.1.2.2.1",
oids=[OIDEnd(), "2"],
),
)
# ============================================================
# 6) IPADDR
# ============================================================
def parse_teltonika_ipaddr(st: StringTable) -> dict:
out = {}
for row in st:
if len(row) < 2:
continue
ip = str(row[0]).strip()
idx = _as_int(row[1], -1)
if idx < 0:
continue
out.setdefault(idx, []).append(ip)
return out
snmp_section_teltonika_ipaddr = SimpleSNMPSection(
name="teltonika_ipaddr",
parse_function=parse_teltonika_ipaddr,
detect=exists(".1.3.6.1.2.1.4.20.1.2.127.0.0.1"),
fetch=SNMPTree(base=".1.3.6.1.2.1.4.20.1", oids=["1", "2"]),
)
# ============================================================
# 7) ROUTE
# ============================================================
def parse_teltonika_route(st: StringTable) -> list:
out = []
for row in st:
if len(row) < 2:
continue
dest = str(row[0]).strip()
ifidx = _as_int(row[1], -1)
if ifidx >= 0:
out.append((dest, ifidx))
return out
snmp_section_teltonika_route = SimpleSNMPSection(
name="teltonika_route",
parse_function=parse_teltonika_route,
detect=exists(".1.3.6.1.2.1.4.21.1.1.0.0.0.0"),
fetch=SNMPTree(base=".1.3.6.1.2.1.4.21.1", oids=["1", "2"]),
)
# ============================================================
# 8) WAN LINKS (Teltonika table 48690.12.2.1: name + status)
# 12.2.1.2.x = name (mob1s1a1 / mob2s2a1 / wan2)
# 12.2.1.5.x = status (online/standby/...)
# ============================================================
def parse_teltonika_wanlinks(st: StringTable) -> dict:
out = {}
for i, row in enumerate(st, start=1):
cols = [str(x) for x in row]
cols += [""] * (2 - len(cols))
out[str(i)] = {
"name": cols[0].strip(),
"status": cols[1].strip().lower(),
}
return out
snmp_section_teltonika_wanlinks = SimpleSNMPSection(
name="teltonika_wanlinks",
parse_function=parse_teltonika_wanlinks,
detect=exists(".1.3.6.1.4.1.48690.12.2.1.2.1"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.12.2.1",
oids=["2", "5"],
),
)
# ============================================================
# 8b) MULTIWAN links (48690.12.2.1.*)
# 12.2.1.2.<idx> = link name (mob1s1a1, mob2s2a1, wan2)
# 12.2.1.5.<idx> = state (online/standby/down/...)
# ============================================================
def parse_teltonika_wanlinks(st: StringTable) -> dict:
res = {}
for row in st:
if len(row) < 3:
continue
idx = _as_int(row[0], -1)
if idx < 0:
continue
name = str(row[1]).strip()
state = str(row[2]).strip().lower()
res[idx] = {"name": name, "state": state}
return res
snmp_section_teltonika_wanlinks = SimpleSNMPSection(
name="teltonika_wanlinks",
parse_function=parse_teltonika_wanlinks,
detect=exists(".1.3.6.1.4.1.48690.12.2.1.2.1"),
fetch=SNMPTree(
base=".1.3.6.1.4.1.48690.12.2.1",
oids=[OIDEnd(), "2", "5"], # idx, name, state
),
)
# ============================================================
# 9) ACTIVE WAN (prefer MultiWAN table over route guessing)
# ============================================================
def discover_teltonika_active_wan(section_teltonika_route, section_teltonika_ifdescr, section_teltonika_ipaddr, section_teltonika_wanlinks) -> DiscoveryResult:
yield Service()
def check_teltonika_active_wan(section_teltonika_route, section_teltonika_ifdescr, section_teltonika_ipaddr, section_teltonika_wanlinks):
# 1) Default route -> interface + IPs (technical reality)
default_if = None
for dest, idx in section_teltonika_route or []:
if dest == "0.0.0.0":
default_if = idx
break
if default_if is None:
yield Result(state=State.UNKNOWN, summary="No default route (0.0.0.0) found via SNMP")
return
ifname = (section_teltonika_ifdescr or {}).get(default_if) or f"ifIndex {default_if}"
ips = (section_teltonika_ipaddr or {}).get(default_if, [])
ip_txt = ", ".join(sorted(ips)) if ips else "no IP"
# 2) Determine ACTIVE PATH from MultiWAN states (human-friendly)
# Prefer: wan2 (wifi) if online, else mob1s1a1, else mob2s2a1, else fallback
links = section_teltonika_wanlinks or {}
online_names = [v.get("name") for v in links.values() if v.get("state") == "online" and v.get("name")]
online_names = [n.strip() for n in online_names if n]
path_txt = None
chosen_link = None
def _pick(preferred: str):
for n in online_names:
if n == preferred:
return n
return None
# map your MultiWAN
chosen_link = _pick("wan2") or _pick("mob1s1a1") or _pick("mob2s2a1")
if chosen_link == "wan2":
path_txt = f"WiFi uplink ({ifname})"
elif chosen_link == "mob1s1a1":
path_txt = "LTE SIM1"
elif chosen_link == "mob2s2a1":
path_txt = "LTE SIM2"
else:
# if multiple online (loadbalance) or unknown naming
if online_names:
path_txt = " / ".join(online_names)
else:
path_txt = f"{ifname}"
# Important: keep it unambiguous. One active path shown.
yield Result(
state=State.OK,
summary=f"Active WAN: {path_txt} IPs: {ip_txt}",
details=f"if={ifname} (ifIndex {default_if}); online_links={', '.join(online_names) if online_names else '-'}",
)
check_plugin_teltonika_active_wan = CheckPlugin(
name="teltonika_active_wan",
service_name="Teltonika Active WAN",
sections=["teltonika_route", "teltonika_ifdescr", "teltonika_ipaddr", "teltonika_wanlinks"],
discovery_function=discover_teltonika_active_wan,
check_function=check_teltonika_active_wan,
)
# ============================================================
# 10) WLAN interfaces (wlan*), show IPs if present
# ============================================================
def discover_teltonika_wlan(section_teltonika_ifdescr, section_teltonika_ipaddr) -> DiscoveryResult:
for ifidx, name in sorted((section_teltonika_ifdescr or {}).items()):
if str(name).startswith("wlan"):
yield Service(item=f"{ifidx}:{name}")
def check_teltonika_wlan(item: str, section_teltonika_ifdescr, section_teltonika_ipaddr):
try:
ifidx_s, ifname = item.split(":", 1)
ifidx = _as_int(ifidx_s, -1)
except Exception:
yield Result(state=State.UNKNOWN, summary="Bad WLAN item format")
return
ips = (section_teltonika_ipaddr or {}).get(ifidx, [])
ip_txt = ", ".join(sorted(ips)) if ips else "no IP"
yield Result(state=State.OK, summary=f"{ifname} IPs: {ip_txt}")
check_plugin_teltonika_wlan = CheckPlugin(
name="teltonika_wlan",
service_name="Teltonika WLAN %s",
sections=["teltonika_ifdescr", "teltonika_ipaddr"],
discovery_function=discover_teltonika_wlan,
check_function=check_teltonika_wlan,
)
