Get ONU list from ONT by SNMP and generate EXCEL table

.gitignore

#################
## Eclipse
#################

*.pydevproject
.project
.metadata
bin/
tmp/
*.tmp
*.bak
*.swp
*~.nib
local.properties
.classpath
.settings/
.loadpath

# External tool builders
.externalToolBuilders/

# Locally stored "Eclipse launch configurations"
*.launch

# CDT-specific
.cproject

# PDT-specific
.buildpath


#################
## Visual Studio
#################

## Ignore Visual Studio temporary files, build results, and
## files generated by popular Visual Studio add-ons.

# User-specific files
*.suo
*.user
*.sln.docstates

# Build results

[Dd]ebug/
[Rr]elease/
x64/
build/
[Bb]in/
[Oo]bj/

# MSTest test Results
[Tt]est[Rr]esult*/
[Bb]uild[Ll]og.*

*_i.c
*_p.c
*.ilk
*.meta
*.obj
*.pch
*.pdb
*.pgc
*.pgd
*.rsp
*.sbr
*.tlb
*.tli
*.tlh
*.tmp
*.tmp_proj
*.log
*.vspscc
*.vssscc
.builds
*.pidb
*.log
*.scc

# Visual C++ cache files
ipch/
*.aps
*.ncb
*.opensdf
*.sdf
*.cachefile

# Visual Studio profiler
*.psess
*.vsp
*.vspx

# Guidance Automation Toolkit
*.gpState

# ReSharper is a .NET coding add-in
_ReSharper*/
*.[Rr]e[Ss]harper

# TeamCity is a build add-in
_TeamCity*

# DotCover is a Code Coverage Tool
*.dotCover

# NCrunch
*.ncrunch*
.*crunch*.local.xml

# Installshield output folder
[Ee]xpress/

# DocProject is a documentation generator add-in
DocProject/buildhelp/
DocProject/Help/*.HxT
DocProject/Help/*.HxC
DocProject/Help/*.hhc
DocProject/Help/*.hhk
DocProject/Help/*.hhp
DocProject/Help/Html2
DocProject/Help/html

# Click-Once directory
publish/

# Publish Web Output
*.Publish.xml
*.pubxml
*.publishproj

# NuGet Packages Directory
## TODO: If you have NuGet Package Restore enabled, uncomment the next line
#packages/

# Windows Azure Build Output
csx
*.build.csdef

# Windows Store app package directory
AppPackages/

# Others
sql/
*.Cache
ClientBin/
[Ss]tyle[Cc]op.*
~$*
*~
*.dbmdl
*.[Pp]ublish.xml
*.pfx
*.publishsettings

# RIA/Silverlight projects
Generated_Code/

# Backup & report files from converting an old project file to a newer
# Visual Studio version. Backup files are not needed, because we have git ;-)
_UpgradeReport_Files/
Backup*/
UpgradeLog*.XML
UpgradeLog*.htm

# SQL Server files
App_Data/*.mdf
App_Data/*.ldf

#############
## Windows detritus
#############

# Windows image file caches
Thumbs.db
ehthumbs.db

# Folder config file
Desktop.ini

# Recycle Bin used on file shares
$RECYCLE.BIN/

# Mac crap
.DS_Store


#############
## Python
#############

*.py[cod]

# Packages
*.egg
*.egg-info
dist/
build/
eggs/
parts/
var/
sdist/
develop-eggs/
.installed.cfg

# Installer logs
pip-log.txt

# Unit test / coverage reports
.coverage
.tox

#Translations
*.mo

#Mr Developer
.mr.developer.cfg


#############
## Others
#############

*.cfg
*.local

.idea

*.xls*

README.md

GET ONU DATA

Loads information about ONU from fro ONT by snmp

Prerequest

Required python3 and additional modules:

• pysnmp

• openpyxl

• pillow

• tqdm

  sudo -H pip3 install pysnmp openpyxl pillow tqdm
  

oids.csv

Contain list of oids to scan

oid	name	type	multiplier	format	high	low
1.3.6.1.4.1.17409.2.3.4.1.1.1	onuDeviceIndex	int
1.3.6.1.4.1.17409.2.3.4.1.1.2	onuName	str
1.3.6.1.4.1.17409.2.3.4.1.1.7	onuMacAddress	mac		:
1.3.6.1.4.1.17409.2.3.4.2.1.8	onuWorkingTemperature.	float	0.01.	"#.## ""°C""".	75.
1.3.6.1.4.1.17409.2.3.4.2.1.4	onuReceivedOpticalPower	float	0.01	"#.## ""dBm"""	-10	-30

example

oid,name,type,multiplier,format,high,low
1.3.6.1.4.1.17409.2.3.4.1.1.1,onuDeviceIndex,,,,,
1.3.6.1.4.1.17409.2.3.4.1.1.2,onuName,str,,,,
1.3.6.1.4.1.17409.2.3.4.1.1.7,onuMacAddress,mac,,:,,
1.3.6.1.4.1.17409.2.3.4.2.1.8,onuWorkingTemperature,float,0.01,"#.## ""°C""",50,
1.3.6.1.4.1.17409.2.3.4.2.1.4,onuReceivedOpticalPower,float,0.01,"#.## ""dBm""",-10,-30```

fd12xx.csv

get_onu_data.py

#!/usr/bin/python3

# sudo -H pip3 install openpyxl pillow pysnmp tqdm
# Get onu list

__version__ = '0.3'
__copyright__ = "Vladimir Kushnir aka Kvantum i(c)2020"

import csv
import os
from datetime import datetime, timedelta
from optparse import OptionParser, OptionGroup

import pysnmp.smi
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Alignment
from openpyxl.utils.cell import get_column_letter
from openpyxl.worksheet.table import Table, TableStyleInfo
from pysnmp.hlapi import *
from pysnmp.proto.rfc1905 import endOfMibView

from tqdm import tqdm

# init pysnmp
snmpEngine = SnmpEngine()
contextData = ContextData()
mibBuilder = pysnmp.smi.builder.MibBuilder()
mibViewController = pysnmp.smi.view.MibViewController(mibBuilder)
pysnmp.smi.compiler.addMibCompiler(mibBuilder)


def get_options(arguments=None):
    """Parse Command-Line parameters"""
    parser = OptionParser(usage="%prog [options] <Device>",
                          version="%prog " + __version__)
    parser.add_option('--vendor', dest='vendor',
                      choices=['C-DATA', 'VSOL'],
                      help='PON Vendor')
    parser.add_option('--oids', dest='oids',
                      help='CSV file with list oids and its descriptions to scan')
    parser.add_option('--excel', dest='excel', default=datetime.now().strftime("%d-%m-%Y.xlsx"),
                      help='Excel file to save results')
    parser.add_option('--sheet', dest='sheet', default=datetime.now().strftime("%d-%m-%Y"),
                      help='Excel sheet to save results')

    group = OptionGroup(parser, "SNMP")
    group.add_option('--snmp_version', dest='snmp_version', choices=['1', '2c', '3'], default='2c',
                     help="specifies SNMP version to use")
    group.add_option('--snmp_community', dest='snmp_community', default='public',
                     help='set the community string')
    parser.add_option_group(group)

    (opt, args) = parser.parse_args(arguments)

    if opt.snmp_version == '3':
        parser.error("SNMP version 3 not supported yet !")
    if len(args) < 1:
        parser.error("You must specify device IP !")

    return args[0], opt


def load_csv(file_name):
    data = []
    with open(file_name, newline='') as cf:
        reader = csv.DictReader(cf)
        for row in reader:
            data.append(row)
    return data


def get_indexes(scan_oid, opt):
    indexes = []
    for (errorIndication, errorStatus, errorIndex, varBinds) in \
            bulkCmd(snmpEngine,
                    opt.auth_data,
                    opt.transport_target,
                    contextData,
                    1, 50,
                    ObjectType(ObjectIdentity(scan_oid)), lexicographicMode=False):
        if errorIndication:
            print(errorIndication)
        else:
            if errorStatus:
                print('%s at %s\n' % (
                    errorStatus.prettyPrint(),
                    errorIndex and varBinds[-1][int(errorIndex) - 1] or '?'))
            else:
                for varBind in varBinds:
                    scan_oid, value = varBind
                    if value == endOfMibView:
                        break
                    else:
                        indexes.append(int(value))
    return indexes


def get_data(indexes, data, worksheet, col, opt):
    for (errorIndication, errorStatus, errorIndex, varBinds) in \
            bulkCmd(snmpEngine,
                    opt.auth_data,
                    opt.transport_target,
                    contextData,
                    1, 50,
                    ObjectType(ObjectIdentity(data['oid'])), lexicographicMode=False):
        if errorIndication:
            print(errorIndication)
        else:
            if errorStatus:
                print('%s at %s\n' % (
                    errorStatus.prettyPrint(),
                    errorIndex and varBinds[-1][int(errorIndex) - 1] or '?'))
            else:
                for varBind in varBinds:
                    oid, value = varBind

                    if value == endOfMibView:
                        break
                    # fix china hardware
                    for i in range(-1, -5, -1):
                        sidx = int(oid[i])
                        if sidx > 0:
                            break
                    # unpack value
                    try:
                        row = indexes.index(sidx) + 2
                    except ValueError:
                        indexes.append(sidx)
                        row = indexes.index(sidx) + 2
                    if data['type'] == 'str':
                        value = str(value)
                        worksheet.cell(column=col, row=row, value=value)
                    elif data['type'] == 'int':
                        if data['multiplier']:
                            value = int(value) * int(data['multiplier'])
                        else:
                            value = int(value)
                        worksheet.cell(column=col, row=row, value=value)
                        if data['format']:
                            worksheet.cell(column=col, row=row).number_format = data['format']
                    elif data['type'] == 'float':
                        if data['multiplier']:
                            value = float(value) * float(data['multiplier'])
                        else:
                            value = float(value)
                        worksheet.cell(column=col, row=row, value=value)
                        if data['format']:
                            worksheet.cell(column=col, row=row).number_format = data['format']
                    elif data['type'] == 'time':
                        if data['multiplier']:
                            value = timedelta(seconds=(int(value) * data['multiplier']))
                        else:
                            value = timedelta(seconds=int(value))
                        worksheet.cell(column=col, row=row, value=value)
                    elif data['type'] == 'hex':
                        value = ''.join(['%0.2x' % x for x in value.asNumbers()])
                        worksheet.cell(column=col, row=row, value=value)
                    elif data['type'] == 'mac':
                        value = data['format'].join(['%0.2x' % x for x in value.asNumbers()]).upper()
                        worksheet.cell(column=col, row=row, value=value)
                    elif data['type'] == 'dict':
                        value = eval(data['format'])[value]
                        worksheet.cell(column=col, row=row, value=value)
                    else:
                        worksheet.cell(column=col, row=row, value=value)
                    if worksheet.cell(column=1, row=row).value is None:
                        worksheet.cell(column=1, row=row, value=sidx)


if __name__ == '__main__':
    host, options = get_options()
    if options.snmp_version == '1':
        options.auth_data = CommunityData(options.snmp_community, mpModel=0)
    elif options.snmp_version.lower() == '2c':
        options.auth_data = CommunityData(options.snmp_community, mpModel=1)
    try:
        ip, port = host.split(':')
    except ValueError:
        ip = host
        port = 161
    options.transport_target = UdpTransportTarget((ip, port), timeout=2, retries=5)

    if os.path.isfile(options.excel):
        wb = load_workbook(options.excel)
    else:
        wb = Workbook()

    oids = load_csv(options.oids)
    oid_idx = oids.pop(0)
    ws = wb.create_sheet(options.sheet)
    column = 2
    ws.cell(row=1, column=1, value='#').alignment = Alignment(horizontal='center', vertical='center')

    print("\nLoading ONU snmp indexes ...")
    idxs = get_indexes(oid_idx['oid'], options)
    print("Found {} indexes!\n".format(len(idxs)))

    print("Loading snmp values ...")
    for oid in tqdm(oids):
        ws.cell(row=1, column=column, value=oid['name']).alignment = \
            Alignment(horizontal='center', vertical='center')
        get_data(idxs, oid, ws, column, options)
        column += 1
    print("All job done!")

    # :TODO Add conditional formatting support

    table_ref = "A1:{}{}".format(get_column_letter(column - 1), len(idxs) + 1)
    table_style = TableStyleInfo(name='TableStyleMedium2', showRowStripes=True, showFirstColumn=True)
    table = Table(displayName='Table', ref=table_ref, tableStyleInfo=table_style)
    ws.add_table(table)
    wb.save(options.excel)