#!/usr/bin/env python
# -*- coding: utf-8 -*-
from dataclasses import dataclass, fields
from datetime import datetime
from dataclass_factory import Factory, Schema
from enum import Enum
from typing import List
from nms_search import cls_search
from const import VNO, ALL_NETWORK_ID_ISNOT_REF
from prettytable import PrettyTable
from send_snmp import send_mail
@dataclass
class Level:
ok: float
bad: float = None
@dataclass
class LevelSettings:
rx: Level
tx: Level
class LevelState(Enum):
ok = 'ok'
warning = 'warning'
bad = 'bad'
@dataclass()
class Station:
name: str
spyt: str
cn_hub: float
cn_rem: float
rcst: str
txlvl: float
acmch: str
faults: str
updated: str
state: LevelState = None
LEVELS = {
'ya': LevelSettings(rx=Level(13.), tx=Level(7., 6.5)),
'am5': LevelSettings(rx=Level(12.), tx=Level(11.5, 10.5)),
'am7': LevelSettings(rx=Level(11.5), tx=Level(7.5, 6.5)),
}
def check_level(actual: float, expected_level: Level) -> LevelState:
if actual > expected_level.ok:
return LevelState.ok
if expected_level.bad is None or actual < expected_level.bad:
return LevelState.bad
return LevelState.warning
def combine_level_states(*states: LevelState) -> LevelState:
if any(s is LevelState.bad for s in states):
return LevelState.bad
if all(s is LevelState.ok for s in states):
return LevelState.ok
return LevelState.warning
def check(stations: List[Station]):
for station in stations:
level_settings = LEVELS[station.spyt]
station.state = combine_level_states(
check_level(station.cn_rem, level_settings.rx),
check_level(station.cn_hub, level_settings.tx),
)
return stations
def print_state(vno):
s = cls_search()
st = s.list_remotestate_by_vno(vno)
date_format = '%Y-%m-%d %H:%M:%S'
factory = Factory(schemas={
datetime: Schema(
parser=lambda s: datetime.strptime(s, date_format),
serializer=lambda d: d.strftime(date_format)
)
})
stations = factory.load(st, List[Station])
report = PrettyTable()
bad_stations = (x for x in check(stations) if x.cn_hub > 0 and x.state == LevelState.bad)
dict_bad_stations = factory.dump(bad_stations, List[Station])
report.field_names = ('name', 'cn_rem', 'cn_hub', 'acmch', 'Время проверки')
report.align['name'] = 'l'
for i in dict_bad_stations:
report.add_row((i['name'], i['cn_rem'], i['cn_hub'], i['acmch'], i['updated'] ))
return report.get_string(title=f'Станции с плохими уровнями в популяции "{VNO[vno]}"',
sortby='cn_rem')
if __name__ == "__main__":
mail = send_mail()
file = {}
for i in ALL_NETWORK_ID_ISNOT_REF:
file[VNO[i]] = f'{print_state(i)}.txt'
mail.send('Report from bad station', 'Отчет по станциям с низкими уровнями', file=file)