Merge branch 'master' of http://jiecheng2024.tpddns.cn:3000/liyuezong/SmartPDUPro

04-软件设计及源码/03-核心板/V1.0/01.doc/04-用户文档/pdusnmp_example.py

@@ -0,0 +1,329 @@
+# _*_ coding:utf-8 _*_
+#依赖pysnmp 请自行安装(可以使用命令 pip install pysnmp)
+from pysnmp.hlapi import *
+import time
+import threading
+import time
+from datetime import datetime
+
+GroupDelay = 0.3
+Grouplock = threading.Lock()
+
+def validate_ip(ip_str: str) -> bool:
+    return True
+
+
+def validate_Sock(nSock: int) -> bool:
+    if nSock > 120 or nSock <= 0:
+        return False
+    else:
+        return True
+
+def validate_Group(nGroup: int) -> bool:
+    if nGroup > 120 or nGroup <= 0:
+        return False
+    else:
+        return True
+
+def validate_Grouplist(nGrouplist: list) -> bool:
+    nChNum = len(nGrouplist)
+    if nChNum > 120:
+        print('invalid GroupNum!')
+        return False
+    for item in nGrouplist:
+        nvalue = int(item)
+        if nvalue < 1 or nvalue > 120:
+            return False
+    return True
+
+def is_float(s):
+    try:
+        float(s)
+        return True
+    except ValueError:
+        return False
+
+def is_int(s):
+    try:
+        int(s)
+        return True
+    except ValueError:
+        return False
+
+def get_value(devip:str, soid:str) -> str:
+    # 实现snmpget
+    g = getCmd(SnmpEngine(),  # 创建SNMP引擎
+               CommunityData('public'),  # 团体属性
+               # userData,交换机没有配认证，这里就不需要
+               UdpTransportTarget((devip, 161), timeout=1, retries=0),  # 创建被管理设备信息 包含IP和端口号 超时1秒 不重试
+               ContextData(),  # 创建SNMP上下文信息
+               ObjectType(ObjectIdentity(soid))  # 创建MIB节点对象 #要跟参数 否则会报错
+               )
+    errorIndication, errorStatus, errorIndex, varBinds = next(g)  # 发送Get请求，获取sysName
+    if errorIndication:
+        print(errorIndication)
+        return '-1'
+    elif errorStatus:
+        print('%s at %s' % (errorStatus.prettyPrint(),
+                             errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))
+        return '-1'
+    else:
+        for i in varBinds:
+           value = i.prettyPrint().split("=")[1]
+        return value
+
+
+def set_value(devip:str, soid:str, varvalue:str) -> bool:
+    # 实现snmpset
+    # SNMP SET请求
+    errorIndication, errorStatus, errorIndex, varBinds = next(
+        setCmd(SnmpEngine(),
+               CommunityData('public'),
+               UdpTransportTarget((devip, 161)),
+               ContextData(),
+               ObjectType(ObjectIdentity(soid), Integer32(varvalue))
+               )
+    )
+    # 检查错误
+    if errorIndication:
+        print(errorIndication)
+        return False
+    elif errorStatus:
+        print('%s at %s' % (errorStatus.prettyPrint(),
+                             errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))
+        return False
+    else:
+        #for varBind in varBinds:
+        #  print(' = '.join([x.prettyPrint() for x in varBind]))
+        return True
+
+
+#获取设备名称
+def GetDeviceName(sDevIp: str) -> str:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return 'Invalid Name'
+    sOId = '1.3.6.1.4.1.2024.1.1.2.0'
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        print('invalid value!')
+        return 'Invalid Name'
+    rt_value = str(value)
+    return rt_value
+
+
+#打开或关闭指定插口
+def TurnOnOff(sDevIp:str, nSock:int, onoff:bool) -> bool:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return False
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return False
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.3.%d' % (nSock)
+    if onoff == True:
+        state = '1'
+    else:
+        state = '0'
+    return set_value(sDevIp, sOId, state)
+
+
+#打开或关闭制定分组/端口
+def TurnOnOffGroup(sDevIp:str, nGroup:int, nGrouplist:list, onoff:bool) -> bool:
+    Grouplock.acquire()
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return False
+    if validate_Group(nGroup) == False:
+        print('invalid Gro'
+              ''
+              'up!')
+        return False
+    if validate_Grouplist(nGrouplist) == False:
+        print('invalid GroupNum!')
+        return False
+
+    for item in nGrouplist:
+        start_time = time.time()
+        TurnOnOff(sDevIp,int(item),onoff)
+        end_time = time.time()
+        elapsed_time = end_time - start_time
+        #print(f"Chnnel took {elapsed_time:.6f} seconds to execute %d !!!" % (int(item)))
+    print('Set Group %d Status %d' % (nGroup, onoff))
+    time.sleep(GroupDelay)
+    Grouplock.release()
+    return True
+
+
+#获取分组状态 0-全部关闭 1-全部开启 2-部分开启、部分关闭 -1-获取失败
+def GetStatusGroup(sDevIp:str, nGroup:int, nGrouplist:list) -> int:
+    Grouplock.acquire()
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return -1
+    if validate_Group(nGroup) == False:
+        print('invalid Group!')
+        return -1
+    if validate_Grouplist(nGrouplist) == False:
+        print('invalid GroupNum!')
+        return -1
+
+    value0=False
+    value1=False
+
+    rt_value = -1
+    for item in nGrouplist:
+        if GetStatus(sDevIp,int(item)) == 0:
+            value0 = True
+            break
+
+    for item in nGrouplist:
+        if GetStatus(sDevIp,int(item)) == 1:
+            value1 = True
+            break
+
+    if value0 and not value1:
+        rt_value = 0
+    elif not value0 and value1:
+        rt_value = 1
+    elif value0 and value1:
+        rt_value = 2
+    else:
+        rt_value = -1
+
+    print('Get Group %d Status %d' % (nGroup,rt_value))
+    time.sleep(GroupDelay)
+    Grouplock.release()
+    return rt_value
+
+
+
+#获取插口状态 0-关闭 1-开 -1-查询失败
+def GetStatus(sDevIp:str, nSock:int) -> int:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return -1
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return -1
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.3.%d' % (nSock)
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        return -1
+
+    rt_value = -1
+    if is_int(value):
+        rt_value = int(value)
+    #print('Get Ch %d Status %d' % (nSock, rt_value))
+    return rt_value
+
+
+#获取指定插口电流 A
+def GetCurrent(sDevIp:str, nSock:int) -> float:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return 0
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return 0
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.5.%d' % (nSock)
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        return 0
+
+    rt_value = 0
+    if is_float(value):
+        rt_value = float(value)
+    return rt_value
+
+#获取指定插口电压 V
+def GetVoltage(sDevIp:str, nSock:int) -> float:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return 0
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return 0
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.4.%d' % (nSock)
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        return 0
+
+    rt_value = 0
+    if is_float(value):
+        rt_value = float(value)
+    return rt_value
+
+#获取指定插口电能 kWh
+def GetEnergy(sDevIp:str, nSock:int) -> float:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return 0
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return 0
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.7.%d' % (nSock)
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        return 0
+
+    rt_value = 0
+    if is_float(value):
+        rt_value = float(value)
+    return rt_value
+
+
+#获取指定插口名称
+def GetSockName(sDevIp:str, nSock:int) -> str:
+    if validate_ip(sDevIp) == False:
+        print('invalid ip address!')
+        return 0
+    if validate_Sock(nSock) == False:
+        print('invalid sock!')
+        return 0
+    sOId = '1.3.6.1.4.1.2024.1.3.9.1.2.%d' % (nSock)
+    value = get_value(sDevIp, sOId)
+    if value is None:
+        print('invalid value!')
+        return ''
+    rt_value = str(value)
+    return rt_value
+
+
+
+#以下为操作示例
+#1、每隔3s上电/下电关PDU的1-2通道，循环开关1000次，并打印在输出窗口中
+#2、每隔1min采集PDU的1-2通道的电压和电流值，并添加时间戳保存在pdu_log.txt中
+ip = '192.168.1.139'
+file_name = 'pdu_log.txt'
+# 定义一个函数，这个函数将被线程执行
+
+start_time = time.time()
+nstatus = False
+for i in range(1000):
+    TurnOnOff(ip, 1, nstatus)
+    TurnOnOff(ip, 2, nstatus)
+    time_now = time.strftime('%Y-%m-%d %H:%M:%S')
+    if nstatus:
+        print(f"{time_now} 通道1-2上电操作 !!!")
+    else:
+        print(f"{time_now} 通道1-2下电操作 !!!")
+
+    end_time = time.time()
+    elapsed_time = end_time - start_time
+    if elapsed_time >= 60:
+        nVoltage1 = GetVoltage(ip, 1)
+        nCurrent1 = GetCurrent(ip, 1)
+        nVoltage2 = GetVoltage(ip, 2)
+        nCurrent2 = GetCurrent(ip, 2)
+        log_info1 = f"{time_now} 通道1电压{nVoltage1:.3f}V  通道1电流{nCurrent1:.3f}A\n"
+        log_info2 = f"{time_now} 通道2电压{nVoltage2:.3f}V  通道2电流{nCurrent2:.3f}A\n"
+        with open(file_name, 'a') as file:
+            file.write(log_info1)
+            file.write(log_info2)
+        start_time = end_time
+        time.sleep(1)  # 等待1秒
+    nstatus = not nstatus
+    time.sleep(3)  # 等待3秒
+
+

04-软件设计及源码/03-核心板/V1.0/03.lib/valgrind/valraind调用.txt

@@ -1 +1,3 @@
-export VALGRIND_LIB=/root/valraind/libexec/valgrind/
+export VALGRIND_LIB=/root/run/valraind/libexec/valgrind/
+
+/root/run/valraind/bin/valgrind --time-stamp=yes --track-origins=yes --tool=memcheck --leak-check=full --show-leak-kinds=all ./smartPDU