RFC1759

prntmon.py

'''
License: GPL

https://www.ietf.org/rfc/rfc1759.txt
'''

from argparse import ArgumentParser
from collections import namedtuple
from enum import IntEnum
from functools import partial
import sys
import time

from flags import Flags
from pysnmp.hlapi import getCmd
from pysnmp.hlapi import SnmpEngine
from pysnmp.hlapi import CommunityData
from pysnmp.hlapi import ContextData
from pysnmp.hlapi import UdpTransportTarget
from pysnmp.hlapi import ObjectType
from pysnmp.hlapi import ObjectIdentity


class hrDeviceStatus(IntEnum):
    unknown = 1
    running = 2
    warning = 3
    testing = 4
    down = 5


class hrPrinterStatus(IntEnum):
    other = 1
    unknown = 2
    idle = 3
    printing = 4
    warmup = 5


class hrPrinterDetectedErrorState(Flags):
    lowPaper = 1 << 7
    noPaper = 1 << 6
    lowToner = 1 << 5
    noToner = 1 << 4
    doorOpen = 1 << 3
    jammed = 1 << 2
    offline = 1 << 1
    serviceRequested = 1 << 0


class PrinterMonitor(object):
    prtMarkerLifeCount = "1.3.6.1.2.1.43.10.2.1.4"
    hrDeviceDescr = "1.3.6.1.2.1.25.3.2.1.3"
    hrDeviceStatus = "1.3.6.1.2.1.25.3.2.1.5"
    hrPrinterStatus = "1.3.6.1.2.1.25.3.5.1.1"
    hrPrinterDetectedErrorState = "1.3.6.1.2.1.25.3.5.1.2"
    Result = namedtuple("QueryResult",
                        "errorIndication errorStatus errorIndex varBinds")

    def __init__(self, target, udpport=161, community="public"):
        self._query = partial(
            lambda *args: (PrinterMonitor.Result(*r) for r in getCmd(*args)),
            SnmpEngine(),
            CommunityData(community),
            UdpTransportTarget((target, udpport)),
            ContextData())

    def prt_marker_life_count(self):
        return self.query_single(
            PrinterMonitor.prtMarkerLifeCount + ".1.1")

    def hr_device_descr(self):
        return self.query_single(
            PrinterMonitor.hrDeviceDescr + ".1")

    def hr_device_status(self):
        return hrDeviceStatus(self.query_single(
            PrinterMonitor.hrDeviceStatus + ".1"))

    def hr_printer_status(self):
        return hrPrinterStatus(self.query_single(
            PrinterMonitor.hrPrinterStatus + ".1"))

    def hr_printer_detect_error_state(self):
        return hrPrinterDetectedErrorState(self.query_single(
            PrinterMonitor.hrPrinterDetectedErrorState + ".1")[0])

    def query_single(self, oid):
        r = next(self._query(ObjectType(ObjectIdentity(oid))))
        if r.errorIndication or r.errorStatus:
            raise RuntimeError(r)
        return r.varBinds[0][1]


def getargs(args):
    parser = ArgumentParser(description="Status Monitor",)
    parser.add_argument("-C", "--community", type=str, default="public",
                        help="community name")
    parser.add_argument("target", metavar="TARGET", type=str,
                        help="monitoring target")
    return parser.parse_args(args)


def main():
    args = getargs(sys.argv[1:])
    prntmon = PrinterMonitor(args.target, community=args.community)
    print(prntmon.hr_device_descr())
    while True:
        print(
            prntmon.prt_marker_life_count(),
            prntmon.hr_device_status()._name_,
            prntmon.hr_printer_status()._name_,
            prntmon.hr_printer_detect_error_state().to_simple_str(),
        )
        time.sleep(1)


if __name__ == "__main__":
    main()

requirements.txt

py-flags
pysnmp