我正在使用由
pysnmp
安装的
pysnmp-lextudio
。我选择这个包是因为它是纯Python的,因此是跨平台的。其他库要么对我来说太难理解,要么不跨平台,或者需要安装外部系统依赖项。
目前,我正在与一个CyberPower PDU交谈,它基本上是一个具有可控插座的电源,具有以下
get_data
命令:
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)
对于 16 端口 PDU,调用
get_data
16 次需要一秒多一点的时间。每次调用 get_data
大约需要 70 毫秒。这是有问题的,因为它使得 GUI 难以响应插座的实际状态。我想要一个以 1 Hz 到 2 Hz 的节奏有效循环的子进程,以获取所有出口的状态。这是因为插座可能会被 GUI 外部的东西关闭或打开,因此它需要能够准确地显示实际状态。
所以我尝试将命令调整为这样的:
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)
...
这似乎适用于
oids
列表只有一两个元素的情况,但对于 16 个网点的 16 个 OID 的完整列表,它会超时。即使我等了 5 秒,它也会超时。所以我不确定发生了什么。
我意识到还有一个
bulkCmd
,但我不完全确定如何使用它,因为 SNMP 对我来说是新的并且相当神秘。
摘要:我有 OID 列表:
ids = [".1.3.6.1.4.1.3808.1.1.3.3.5.1.1.4.{}".format(outlet) for outlet in range(1, 17)]
我正在寻找查询这些信息的最快方法,以便所有 16 个网点的状态的响应时间远低于一秒。理想情况下,该解决方案使用
pysnmp
包,但我对其他人开放,只要它们是跨平台的并且不需要外部系统依赖项。
puresnmp
Python 库 而不是 pysnmp
或任何其他 Python SNMP 解决方案。 pysnmp
的主要问题是它非常慢,而且它的 asyncio
API 实际上并不是异步的。使用 pysnmp
,获取 CyberPower PDU 上 16 个插座端口的插座状态(即 16 个独立的 OID“获取”)大约需要 1.2 秒。使用 puresnmp
,同样的事情大约需要 0.015 秒。对于 pysnmp
,当使用 asyncio
的 pysnmp
API 进行单个 get 调用时,我看到我的 asyncio
事件循环被阻塞了多达 10-100 毫秒,这是不应该发生的。
puresnmp
是一个纯 Python 库,速度很快并且实际上是异步的。不幸的是, pysnmp
都不是,并且 easysnmp
库 依赖于外部系统依赖项(即,它不是纯 Python 库)。 puresnmp
的API接口也非常简单。
这是使用
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())
对于 16 端口 CyberPower PDU,这需要 10-15 毫秒才能完成。