SNMP-Influx-collector/cont/classes.py

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
    powerDrawBoard: 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"Board {self.powerDrawBoard} 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