Newer
Older
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# License: GNU General Public License v2
# Author: thl-cmk[at]outlook[dot]com
# URL : https://thl-cmk.hopto.org
# Date : 2023-10-12
# File : create_topology_classes.py
from os import environ
from time import strftime
from enum import Enum, unique
from create_topology_utils import (
cdp_path,
cdp_columns,
cdp_label,
lldp_path,
lldp_columns,
lldp_label,
user_data_file,
)
class TopologyParams(NamedTuple):
path: str
columns: str
label: str
@unique
class Topologies(Enum):
CDP = TopologyParams(path=cdp_path, columns=cdp_columns, label=cdp_label)
LLDP = TopologyParams(path=lldp_path, columns=lldp_columns, label=lldp_label)
class InventoryColumns(NamedTuple):
neighbour: str
local_port: str
neighbour_port: str
class Interface(NamedTuple):
host: str
index: str
name: str
description: str
alias: str
class StaticConnection(NamedTuple):
host: str
local_port: str
neighbour_port: str
neighbour: str
label: str
class Settings:
def __init__(
self,
cli_args: Dict[str, Any],
):
self.__topology_save_path = 'var/topology_data'
self.__topology_file_name = 'network_data.json'
self.__path_to_if_table = 'networking,interfaces'
# self.__inventory_path = 'var/check_mk/inventory'
# self.__autochecks_path = 'var/check_mk/autochecks'
'path_in_inventory': cdp_path,
'inventory_columns': cdp_columns,
'data_source': cdp_label,
'user_data_file': f'{self.__omd_root}/local/bin/topology_data/{user_data_file}',
'keep_domain': False,
'lowercase': False,
'uppercase': False,
'debug': False,
'lldp': False,
'version': False,
# args in the form {'s, __seed_devices': 'CORE01', 'p, __path_in_inventory': None, ... }}
# we will remove 's, __'
self.__args = ({k.split(',')[-1].strip(' ').strip('_'): v for k, v in cli_args.items() if v})
# first set values
if self.__args.get('lldp'):
self.__settings['data_source'] = lldp_label
self.__settings['path_in_inventory'] = lldp_path
self.__settings['inventory_columns'] = lldp_columns
if not self.merge:
if not self.lldp:
self.merge = ['CDP']
else:
self.merge = ['LLDP']
def set_topology_param(self, topology: TopologyParams):
self.__settings['path_in_inventory'] = topology.path
self.__settings['inventory_columns'] = topology.columns
self.__settings['data_source'] = topology.label
@property
def version(self) -> bool:
return self.__settings['version']
@property
def merge(self) -> List[str] | None:
return self.__settings['merge']
@merge.setter
def merge(self, topologies: List[str]):
self.__settings['merge'] = topologies
def keep(self) -> int | None:
if self.__settings['keep']:
return max(self.__settings['keep'], 1)
@property
def min_age(self) -> int:
if self.__settings['min_age']:
return 1
@property
def check_user_data_only(self) -> bool:
return self.__settings['check_user_data_only']
@property
def lldp(self) -> bool:
return self.__settings['lldp']
@property
def debug(self) -> bool:
return self.__settings['debug']
@property
def dont_compare(self) -> bool:
return self.__settings['dont_compare']
def default(self) -> bool:
return self.__settings['default']
@property
def keep_domain(self) -> bool:
return self.__settings['keep_domain']
@property
def uppercase(self) -> bool:
return self.__settings['uppercase']
@property
def lowercase(self) -> bool:
return self.__settings['lowercase']
@property
def time_format(self) -> str:
return self.__settings['time_format']
@property
def omd_root(self) -> str:
# replaced by live status query
# @property
# def inventory_path(self) -> str:
# return self.__inventory_path
# @property
# def autochecks_path(self) -> str:
# return self.__autochecks_path
@property
def user_data_file(self) -> str:
return self.__settings['user_data_file']
@property
def path_in_inventory(self) -> List[str]:
path = ('Nodes,' + ',Nodes,'.join(self.__settings['path_in_inventory'].split(',')) + ',Table,Rows').split(',')
return path
@property
def path_to_if_table(self) -> List[str]:
path = ('Nodes,' + ',Nodes,'.join(self.__path_to_if_table.split(',')) + ',Table,Rows').split(',')
return path
@property
def inventory_columns(self) -> InventoryColumns:
columns = self.__settings['inventory_columns'].split(',')
return InventoryColumns(
neighbour=columns[0],
local_port=columns[1],
neighbour_port=columns[2],
)
@property
def seed_devices(self) -> List[str]:
if self.__settings['seed_devices']:
else:
return []
@property
def data_source(self) -> str:
return self.__settings['data_source']
@property
def output_directory(self) -> str:
if not self.__settings['output_directory']:
return f'{strftime(self.__settings["time_format"])}'
else:
return self.__settings['output_directory']