SNMP Voice Stats Poller

Simple snmp poller - which stores snmp stats for call data from Cisco Voice gateways and Cisco CUBES.

• Database component - which contains inventory table and stats table
• Python Script which runs every 5 mins on the 5 min interval, e.g. 00,05,10 etc.

• Install Python for all users.
• Create a local windows account , e.g. voice-stats.
• Enable this user via local security the user right assigment of “log on as a batch job”

get-snmp-stats.py

# Author: Gerry O'Rourke
# Version 1.1 - Jan 2021
# Version 1.2 - Mar 2025 - updated Fetch Function to handle tuple rather than iterator (due to changes in get getCmd)
 
from pysnmp import hlapi
from datetime import datetime
import pyodbc 
import sys
 
database_write=1 #enable database writes (disabled later is argument 'ro' is passed)
 
# Check for argument 'ro' or 'readonly' - to disable database writes
if len(sys.argv) >1:
    if (sys.argv[1] == "ro") or (sys.argv[1]) == "readonly":
        print ("Read Only Mode - do not write to Database!")
        database_write = 0
    else:
        print ("Invalid Argument", sys.argv[1], "for read only mode enter argument 'ro' or 'readonly'")
 
#SNMPv2c Values
communitystring='myCommunityString!'
 
#SNMPv3 Values
snmpv3_username = "mySNMPv3_User"
snmpv3_password = "mySNMPv3_PW"
snmpv3_privKey = "mySNMPv3_PK"
snmp_version = 2 # Set version to '2' or to '3'. Other options not valid!
 
#Get the Current Time - rounded down to minute.
now = datetime.now()
current_datetime = now.strftime('%Y-%m-%d %H:%M')
print("Current DateTime =", current_datetime)
 
#Set the OID used - this script only supports one OID - but we could improve easily!
#oid = '1.3.6.1.4.1.9.9.63.1.3.8.2.0' #note OID that SNMP Solar Winds for getting all voice calls - but it includes MTP calls etc.
oid = '1.3.6.1.4.1.9.9.63.1.3.8.1.1.2.2' # OID to get SIP calls only!
 
 
 
def get(target, oids, credentials, port=161, engine=hlapi.SnmpEngine(), context=hlapi.ContextData()):
    handler = hlapi.getCmd(
        engine,
        credentials,
        hlapi.UdpTransportTarget((target, port), timeout=2.0, retries=0),
        context,
        *construct_object_types(oids)
    )
    return fetch(handler, 1)[0]
 
def construct_object_types(list_of_oids):
    object_types = []
    for oid in list_of_oids:
        object_types.append(hlapi.ObjectType(hlapi.ObjectIdentity(oid)))
    return object_types
 
def fetch(handler, count):
    result = []
    for i in range(count):
        try:
            # Since handler is a tuple, unpack it correctly
            error_indication, error_status, error_index, var_binds = handler
 
            if not error_indication and not error_status:
                items = {}
                for var_bind in var_binds:
                    items[str(var_bind[0])] = cast(var_bind[1])
                result.append(items)
            else:
                raise RuntimeError('Got SNMP error: {0}'.format(error_indication))
        except ValueError:
            raise RuntimeError(f"Unexpected return structure from getCmd(): {handler}")
    return result
 
 
def cast(value):
    try:
        return int(value)
    except (ValueError, TypeError):
        try:
            return float(value)
        except (ValueError, TypeError):
            try:
                return str(value)
            except (ValueError, TypeError):
                pass
    return value
 
 
conn = pyodbc.connect('Driver={SQL Server};'
                      'Server=localhost;'
                      'Database=stats_db;'
                      'Trusted_Connection=yes;')
 
#if using SQL auth instead of windows user (remove Trusted_Connection)
#'UID=mysqluser;' 
#'PWD=password123!;') 
 
cursor = conn.cursor()
rows = cursor.execute('SELECT id, ip_address FROM inventory WHERE (type = 1 OR type = 2) AND disabled = \'N\'')
 
for row in list(rows):
#for row in rows:
    try:
        if snmp_version == 2:
            voicestats_dict = get(row[1], [oid], hlapi.CommunityData(communitystring))
        elif snmp_version == 3:
            voicestats_dict = get(row[1], [oid], hlapi.UsmUserData(snmpv3_username, authKey=snmpv3_password, privKey=snmpv3_privKey, authProtocol=hlapi.usmHMACSHAAuthProtocol, privProtocol=hlapi.usmAesCfb128Protocol))
        else: 
            print ("Invalid SNMP version. Exiting...")
            exit
        for key in voicestats_dict:
            print (row[1],":", voicestats_dict[key],end = '') #the end = '', prevents a new line!
            sql = "INSERT INTO snmp_stats (DateTime, router_id, oid, value) VALUES (?, ?, ?, ?)"
            val = (current_datetime, row[0], key, voicestats_dict[key])
            #print (sql, val)
            if database_write == 1:
                cursor.execute(sql, val)
                conn.commit()
                print (" : Committed to Database.")
            else:
                print ("") # need a new line here.
    except Exception as e:
        print("IPaddress:",row[1],"Error:", e)        

• Install latest version of SQL Express
• Restore the attached database
• Add your list of gateways and cubes to the inventory table.
• Add the local windows user detailed above with read and write access to the stats_db

stats_db.zip