156 lines
5.3 KiB
Python
156 lines
5.3 KiB
Python
from pydantic import BaseModel
|
|
from sqlalchemy import Column, Date, Float, Integer, String, text
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional, Annotated, Dict
|
|
from sqlalchemy.dialects.mysql import MEDIUMINT, TINYINT, TEXT, TIMESTAMP, FLOAT
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from dataclasses import dataclass
|
|
|
|
# -----------
|
|
|
|
AlchemyBase = declarative_base()
|
|
|
|
# DATABASE "CLASSESS"--------------------------------------------
|
|
# class idracSQLTable(AlchemyBase):
|
|
# def __init__(self, inputData: snmpPyIDRACData):
|
|
# __table_args__ = {'extend_existing': True}
|
|
|
|
# __tablename__ = table_name
|
|
|
|
# id = Column(MEDIUMINT(unsigned=True), primary_key=True, autoincrement=True)
|
|
# time_stamp = Column(
|
|
# TIMESTAMP,
|
|
# server_default=text("CURRENT_TIMESTAMP"),
|
|
# server_onupdate=text("CURRENT_TIMESTAMP"),
|
|
# nullable=False,
|
|
# )
|
|
|
|
# inputData.hostname = Column(TEXT)
|
|
# inputData.powerDrawPSU1 = Column(Float)
|
|
# inputData.powerDrawPSU2 = Column(Float, nullable=True)
|
|
# inputData.voltagePSU1 = Column(Float)
|
|
# inputData.voltagePSU2 = Column(Float, nullable=True)
|
|
# inputData.inletTemp = Column(Float)
|
|
# inputData.exhaustTemp = Column(Float)
|
|
# inputData.cpu1Temp = Column(Float)
|
|
# inputData.cpu2Temp = Column(Float, nullable=True)
|
|
# inputData.uptimeH = Column(Float)
|
|
# inputData.uptimeD = Column(Float, nullable=True)
|
|
|
|
|
|
# Custom--------------------------------------------
|
|
class snmpValueToHost():
|
|
def __init__(self,
|
|
lastV: Annotated[str, "Last value\nNone when initializing"] = None,
|
|
remote: Annotated[str, "Needs to be a valid IP address"] = None,
|
|
nextV: Annotated[str, "Value to insert next\nNone when initializing"] = None,
|
|
lasttime: Annotated[float, "Last time since sending data\nSince Epoch, so time.time()"] = None,
|
|
lastStatus: Annotated[bool, "True if value newest value was inserted\nFalse if data is pending to be sent"] = False
|
|
):
|
|
self.__lastValue = lastV # Save last sent value
|
|
self.__remote = remote # Remote IP address of the SNMP client
|
|
self.__nextValue = nextV # Next value to send. Maybe will be used
|
|
self.__lastSentStatus = lastStatus # If previously sent new data, set to True.
|
|
self.__lastSentTime = lasttime if lasttime else time.time() # The time since the Epoch of the last sent data*
|
|
|
|
|
|
def ___str__(self):
|
|
return self.__lastValue
|
|
|
|
def __changeLastSentValue(self, newValue: int) -> bool:
|
|
self.__nextValue == newValue
|
|
self.__lastValue = self.__nextValue
|
|
self.__nextValue == None
|
|
return True
|
|
|
|
def __updateLastSentTime(self):
|
|
self.__lastSentTime = time.time()
|
|
return True
|
|
|
|
def __updateSentStatus(self):
|
|
if self.__lastSentStatus:
|
|
self.__lastSentStatus = False
|
|
return False
|
|
else:
|
|
self.__lastSentStatus = True
|
|
return True
|
|
|
|
def updateStats(self, nextV: Annotated[int, "Value that will be set as the new one"]):
|
|
__changeLastSentValue(nextV)
|
|
__updateLastSentTime()
|
|
...
|
|
|
|
|
|
# Dataclasses--------------------------------------------
|
|
|
|
@dataclass
|
|
class snmpPyIDRACData():
|
|
"""Dataclass for snmp data"""
|
|
hostname: str
|
|
powerDrawPSU1: int
|
|
voltagePSU1: int
|
|
inletTemp: float
|
|
exhaustTemp: float
|
|
cpu1Temp: float
|
|
uptimeS: int
|
|
uptimeH: float
|
|
|
|
# optional fields
|
|
uptimeD: float | None = None
|
|
cpu2Temp: float | None = None
|
|
powerDrawPSU2: int | None = None
|
|
voltagePSU2: int | None = None
|
|
|
|
|
|
def __str__(self):
|
|
return (
|
|
f"{self.hostname}\n"
|
|
f"PSU1 {self.powerDrawPSU1} Watts\n"
|
|
f"PSU2 {self.powerDrawPSU2} Watts\n"
|
|
f"PSU1 {self.voltagePSU1} Volts\n"
|
|
f"PSU2 {self.voltagePSU2} Volts\n"
|
|
f"Inlet {self.inletTemp} C\n"
|
|
f"Exhaust {self.exhaustTemp} C\n"
|
|
f"CPU1 {self.cpu1Temp} C\n"
|
|
f"CPU2 {self.cpu2Temp} C\n"
|
|
f"Uptime {self.uptimeH} Hours\n"
|
|
f"Uptime {self.uptimeD} Days"
|
|
)
|
|
|
|
@dataclass
|
|
class snmpPyCiscoData():
|
|
"""Dataclass for snmp data"""
|
|
hostname: str
|
|
powerDrawPSU1: int
|
|
voltagePSU1: int
|
|
inletTemp: float
|
|
exhaustTemp: float
|
|
cpu1Temp: float
|
|
uptimeS: int
|
|
uptimeH: float
|
|
|
|
# optional fields
|
|
uptimeD: float | None = None
|
|
cpu2Temp: float | None = None
|
|
powerDrawPSU2: int | None = None
|
|
voltagePSU2: int | None = None
|
|
|
|
|
|
def __str__(self):
|
|
return (
|
|
f"{self.hostname}\n"
|
|
f"PSU1 {self.powerDrawPSU1} Watts\n"
|
|
f"PSU2 {self.powerDrawPSU2} Watts\n"
|
|
f"PSU1 {self.voltagePSU1} Volts\n"
|
|
f"PSU2 {self.voltagePSU2} Volts\n"
|
|
f"Inlet {self.inletTemp} C\n"
|
|
f"Exhaust {self.exhaustTemp} C\n"
|
|
f"CPU1 {self.cpu1Temp} C\n"
|
|
f"CPU2 {self.cpu2Temp} C\n"
|
|
f"Uptime {self.uptimeS} Seconds\n"
|
|
f"Uptime {self.uptimeH} Hours\n"
|
|
f"Uptime {self.uptimeD} Days"
|
|
)
|
|
# @dataclass
|
|
# class idracFanStatus:
|
|
# fans: Dict[str, int] | None = Dict[None, None] # name/index -> rpm |