I am using pysnmp
installed by the pysnmp-lextudio
. I chose this package because it is pure Python and thus cross-platform. Other libraries were either too hard for me to understand or not cross-platform or required external system dependencies to be installed.
Currently, I am talking to a CyberPower PDU, which is basically a power supply with controllable outlets, with the following get_data
command:
def get_data(ip_address: str, object_identity: str) -> int | str:
"""Get the OID's value. Only integer and string values are currently supported."""
iterator = getCmd(
SnmpEngine(),
CommunityData("public", mpModel=0),
UdpTransportTarget(transportAddr=(ip_address, 161), timeout=1, retries=0),
ContextData(),
ObjectType(ObjectIdentity(object_identity)),
)
error_indication, error_status, error_index, variable_bindings = next(iterator)
if error_indication:
raise RuntimeError(str(error_indication))
elif error_status:
raise RuntimeError(str(error_status))
else:
[variable_binding] = variable_bindings
[_oid, value] = variable_binding
return convert_snmp_type_to_python_type(value)
For a 16-port PDU, calling get_data
16 times takes a little over a second. Each call of get_data
takes around 70ms. This is problematic because it makes keeping a GUI responsive to the actual state of an outlet difficult. I want a sub-process effectively looping at a cadence of 1 Hz to 2 Hz getting the state of all the outlets. This is because an outlet could be turned off or on by something external to the GUI, so it needs to be able to accurately show the actual state.
So I tried adjusting my command to something like this:
def get_multiple_data(ip_address: str, object_identities: list[str]) -> int | str:
"""Get the OID's value. Only integer and string values are currently supported."""
# The OID for retrieving an outlet's state is hardcoded for debugging purposes
ids = [".1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.{}".format(outlet) for outlet in range(1, 17)]
oids = [ObjectType(ObjectIdentity(id)) for id in ids]
print("OIDs: " + str(oids))
iterator = getCmd(
SnmpEngine(),
CommunityData("public", mpModel=0),
UdpTransportTarget(transportAddr=(ip_address, 161), timeout=10, retries=0),
ContextData(),
*oids,
)
error_indication, error_status, error_index, variable_bindings = next(iterator)
...
This seems to work for when the oids
list is only one or two elements, but it times out for the full list of 16 OIDs for the 16 outlets. And it times out even if I wait like 5 seconds. So I'm not sure what's going on.
I realize there's also a bulkCmd
, but I'm not entirely sure how to use it, as SNMP is new to me and quite arcane.
Summary: I have the list of OIDs:
ids = [".1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.{}".format(outlet) for outlet in range(1, 17)]
I am looking for the fastest way to query these such that the response time is well-below a second for the state of all 16 outlets. Ideally the solution uses the pysnmp
package, but I am open to others as long as they are cross-platform and require no external system dependencies.
The solution I have found is to use the puresnmp
Python library instead of pysnmp
or any other Python SNMP solution. The major issues with pysnmp
are that it is incredibly slow, and its asyncio
API is not actually asynchronous. Using pysnmp
, getting the outlet state for 16 outlet ports on a CyberPower PDU, which is 16 separate OID "get"s, takes almost 1.2 seconds. Using puresnmp
, the same thing takes around 0.015 seconds. And with pysnmp
, I was seeing my asyncio
event loop being blocked for as much as 10-100ms when making single get calls using pysnmp
's asyncio
API, which shouldn't happen.
puresnmp
is a pure Python library that is fast and actually asynchronous. Unfortunately, pysnmp
is neither of those and the easysnmp
library relies on external system dependencies (i.e., it's not a pure Python library). The API interface for puresnmp
is also very simple.
Here is the solution using puresnmp
:
from puresnmp import V2C, Client, PyWrapper
async def main():
client = PyWrapper(Client(ip="<ip_address>", port=161, credentials=V2C("public")))
number_of_outlets = int(await self.__client.get(oid=".1.3.6.1.4.1.3808.1.1.3.3.1.3.0"))
for outlet in range(1, number_of_outlets + 1):
outlet_state = await client.get(oid=f".1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.{outlet}")
print(f"Outlet {outlet} state: {outlet_state}")
asyncio.run(main())
This takes anywhere from 10-15ms to complete for a 16-port CyberPower PDU.