75 lines
2.8 KiB
Python
75 lines
2.8 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
|
||
|
|
from sqlalchemy.dialects.mysql import MEDIUMINT, TINYINT, TEXT, TIMESTAMP, FLOAT
|
||
|
|
from sqlalchemy.ext.declarative import declarative_base
|
||
|
|
|
||
|
|
# -----------
|
||
|
|
|
||
|
|
AlchemyBase = declarative_base()
|
||
|
|
|
||
|
|
# DATABASE "CLASSESS"--------------------------------------------
|
||
|
|
def sensorsqldata(table_name: Annotated[str, "Name of the table"]):
|
||
|
|
|
||
|
|
class SensorSQLDataClass(AlchemyBase):
|
||
|
|
__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,
|
||
|
|
)
|
||
|
|
TEMP = Column(TINYINT)
|
||
|
|
TEMP = Column(TEXT)
|
||
|
|
TEMP = Column(Float)
|
||
|
|
TEMP = Column(Float)
|
||
|
|
TEMP = Column(Float, nullable=True)
|
||
|
|
TEMP = Column(Float, nullable=True)
|
||
|
|
TEMP = Column(Float, nullable=True)
|
||
|
|
|
||
|
|
return SensorSQLDataClass
|
||
|
|
|
||
|
|
|
||
|
|
# 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"] = time.time(),
|
||
|
|
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 # The time since the Epoch of the last sent data*
|
||
|
|
|
||
|
|
|
||
|
|
def ___str__():
|
||
|
|
return self.__lastValue
|
||
|
|
|
||
|
|
def __changeLastSentValue(slef):
|
||
|
|
self.__lastValue = self.__nextValue
|
||
|
|
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"]):
|
||
|
|
|
||
|
|
...
|