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

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