Отчет по станциям

  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
#!/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)