YuruC3/SNMP-Influx-collector
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added more code

Browse Source
This commit is contained in:
YuruC3
2026-02-26 17:09:56 +01:00
parent be895664d1
commit fe51d046f6
3 changed files with 132 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
cont/code/classes.py
View File

@@ -4,6 +4,7 @@ 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
from dataclasses import dataclass
# -----------
@@ -41,21 +42,23 @@ class snmpValueToHost():
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(),
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 # The time since the Epoch of the last sent data*
self.__lastSentTime = lasttime if lasttime else time.time() # The time since the Epoch of the last sent data*
def ___str__():
def ___str__(self):
return self.__lastValue
def __changeLastSentValue(slef):
def __changeLastSentValue(self, newValue: int) -> bool:
self.__nextValue == newValue
self.__lastValue = self.__nextValue
self.__nextValue == None
return True
def __updateLastSentTime(self):
@@ -71,5 +74,23 @@ class snmpValueToHost():
return True
def updateStats(self, nextV: Annotated[int, "Value that will be set as the new one"]):
...
__changeLastSentValue(nextV)
__updateLastSentTime()
...
# Dataclasses--------------------------------------------
@dataclass
class snmpPyData():
"""Dataclass for snmp data"""
hostname: str
powerDrawPSU1: int
powerDrawPSU2: int
voltagePSU1: int
voltagePSU2: int
inletTemp: float
exhaustTemp: float
uptime: int

11
cont/code/funct.py
View File

@@ -1,11 +1,10 @@
import influxdb_client, sqlalchemy, random, os
# from sqlmodel import Field, Session, SQLModel, create_engine, select
from sqlalchemy import create_engine, exc # , Column, Integer, String, Numeric
from sqlalchemy.ext.declarative import declarative_base
# from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from typing import Annotated, Final
from classes import *
from typing import Annotated, Final
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS, WriteOptions
@@ -68,12 +67,12 @@ def tempsnsrIntoSQLDB(
try:
data = sensorsqldata(tableInp)
sqlData = sensorsqldata(tableInp)
# print("2")
if inpayload:
# print("3")
session.add(data(
session.add(sqlData(
data = insertData,
))
# print("4")
@@ -88,7 +87,7 @@ def tempsnsrIntoSQLDB(
def tempsnsrIntoFluxQLDB(
inpayload: Annotated[SensorPayload, "Payload"]
inpayload: Annotated[snmpPyData, "Payload"]
) -> int:
"""

129
cont/code/main.py
View File

@@ -1,32 +1,61 @@
# import pysnmp, sqlalchemy, ipaddress, os, influxdb_client, re, time
# # SQL
# from sqlalchemy import create_engine, exc # , Column, Integer, String, Numeric
# from sqlalchemy.ext.declarative import declarative_base
# from sqlalchemy.orm import sessionmaker
# # InfluxDB
# from influxdb_client import InfluxDBClient, Point, WritePrecision
# from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS, WriteOptions
import pysnmp, ipaddress, os, re, time
import ipaddress, os, re, time, funct, classes
#SNMP
from easysnmp import Session
# from easysnmp import Session
# QoL
from typing import Annotated
from typing import Annotated, Final
# Program ENV---------------------------------------
HOST_LIST: Final[list] = os.getenv("HOST_LIST")
try:
RESOLV_ADDR: Final[str] = ipaddress(os.getenv("RESOLV_ADDR", "1.1.1.1"))
except ValueError:
raise Exception("Malformed DNS IP address.\nExiting...")
SNMPUSER: Final[str] = os.getenv("SNMPUSER", None)
SNMPPRIVKEY: Final[str] = os.getenv("SNMPPRIVKEY", None)
SNMPAUTHKEY: Final[str] = os.getenv("SNMPAUTHKEY", None)
SNMPAUTHPROTO: Final[str] = os.getenv("SNMPAUTHPROTO", "SHA")
SNMPPRIVPROTO: Final[str] = os.getenv("SNMPPRIVPROTO", "SHA")
SNMPORT: Final[int] = os.getenv("SNMPORT", 161)
# Flightchecks-------------------------------------------
# Helper functions
def check_host_string(input_text: Annotated[str, "Text to check"]):
pattern = re.compile(r"^[A-Za-z].*$", re.IGNORECASE)
return pattern.match(input_text)
if SNMPAUTHPROTO and SNMPPRIVPROTO == "SHA":
USMDATA = UsmUserData(
userName=SNMPUSER,
authKey=SNMPPRIVKEY,
privKey=SNMPAUTHKEY,
authProtocol=usmHMACSHAAuthProtocol,
privProtocol=usmHMACSHAAuthProtocol,
)
elif SNMPAUTHPROTO and SNMPPRIVPROTO == "AES128":
USMDATA = UsmUserData(
userName=SNMPUSER,
authKey=SNMPPRIVKEY,
privKey=SNMPAUTHKEY,
authProtocol=usmAesCfb128Protocol,
privProtocol=usmAesCfb128Protocol,
)
elif SNMPAUTHPROTO == "SHA" and SNMPPRIVPROTO == "AES128":
USMDATA = UsmUserData(
userName=SNMPUSER,
authKey=SNMPPRIVKEY,
privKey=SNMPAUTHKEY,
authProtocol=usmHMACSHAAuthProtocol,
privProtocol=usmAesCfb128Protocol,
)
elif SNMPAUTHPROTO == "AES128" and SNMPPRIVPROTO == "SHA":
USMDATA = UsmUserData(
userName=SNMPUSER,
authKey=SNMPPRIVKEY,
privKey=SNMPAUTHKEY,
authProtocol=usmAesCfb128Protocol,
privProtocol=usmHMACSHAAuthProtocol,
)
else:
raise Exception(f"No PrivAuth option like {SNMPPRIVPROTO}")
# Helper functions
def wattageCalc(amperageInp: float, voltageInp: float) -> float:
return amperageInp * voltageInp
# Check if some neccesary ENVs are passed
if not USEINFLUX and not USESQL:
@@ -37,25 +66,67 @@ if not HOST_LIST:
raise Exception("No hosts passed\nExiting...")
# for host in HOST_LIST check if valid IP and create a list with strings
HOST_LIST_STR = []
for host in HOST_LIST:
try:
if not check_host_string(host):
ip = str(ipaddress.IPv4Address(host))
HOST_LIST_STR.append(ip)
ip = str(ipaddress.IPv4Address(host))
except ValueError:
raise Exception("Malformed IP address.\nExiting...")
raise Exception(f" IP {ip} is invalid.\nExiting...")
# Checking if DNS IP is a valid one
# Done under "Program ENV" on line ~18
# Code-------------------------------------------
import asyncio
from pysnmp.hlapi.v3arch.asyncio import *
async def poolRemote(remoteIP: str):
snmpEngine = SnmpEngine()
iterator = get_cmd(
snmpEngine,
UsmUserData(
userName=SNMPUSER,
authKey=SNMPPRIVKEY,
privKey=SNMPAUTHKEY,
authProtocol=usmHMACSHAAuthProtocol,
privProtocol=usmHMACSHAAuthProtocol,
),
await UdpTransportTarget.create((remoteIP, SNMPORT)),
ContextData(),
ObjectType(ObjectIdentity("SNMPv2-MIB", "sysDescr", 0)),
)
errorIndication, errorStatus, errorIndex, varBinds = await iterator
if errorIndication:
print(errorIndication)
elif errorStatus:
print(
"{} at {}".format(
errorStatus.prettyPrint(),
errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
)
)
else:
for varBind in varBinds:
print(" = ".join([x.prettyPrint() for x in varBind]))
snmpEngine.close_dispatcher()
# tasks = [poolRemote(ip) for ip in HOST_LIST]
# results = await asyncio.gather(*tasks)
# Create connections for MySQL and InfluxDB
# functions for inserting data into two DBs