Python用telnet设置,抓UDP抓采样点并显示

标签：UDP plt Python data telnet np import type

====main.bat====

echo off

rem "d:\Program\WiresharkPortable64\App\Wireshark\tshark.exe" --list-interfaces
rem "d:\Program\WiresharkPortable64\App\Wireshark\tshark.exe" -f udp -i 3 -a duration:1 -w data.pcapng
rem set FILE_NAME=dat

set FILE_NAME=data
set TSHARK="d:\Program\WiresharkPortable64\App\Wireshark\tshark.exe"
set ETH_PORT=3
set UDP_SECOND=1

cd /d %~dp0

rem python tel.py status

del %FILE_NAME%.pcapng
rem %TSHARK% --list-interfaces
%TSHARK% -f udp -i %ETH_PORT% -a duration:%UDP_SECOND% -w %FILE_NAME%.pcapng

rem Parse.exe %FILE_NAME%
del %FILE_NAME%
python Parse.py %FILE_NAME%

python DataPlot.py %FILE_NAME%

====ShowSmv.py====

import os
import tkinter
from tkinter import *
import tel
import Parse
import DataPlot

window = tkinter.Tk()
window.title('Show IO24x SMV')
window.geometry()
#window.geometry('800x400')
window.resizable(0, 0)

l1 = tkinter.Label(window, text='Tshark.exe:')
# 定义输入框1
e1 = tkinter.Entry(window)
e1.insert(0, r'd:\Program\WiresharkPortable64\App\Wireshark\tshark.exe')
l1.grid(row=0, column=0, sticky=E, ipadx=4)
e1.grid(row=0, column=1, sticky=W+E, columnspan=3, ipadx=4, padx=4)
# 定义输入框2
l2 = tkinter.Label(window, text='Eth num:')
e2 = tkinter.Entry(window)
e2.insert(0, '2')
l2.grid(row=1, column=0, sticky=E, ipadx=4)
e2.grid(row=1, column=1, sticky=W, padx=4)

# 定义输入框
l3 = tkinter.Label(window, text='UDP seconds:')
e3 = tkinter.Entry(window)
e3.insert(0, '1')
l3.grid(row=2, column=0, sticky=E, ipadx=4)
e3.grid(row=2, column=1, sticky=W, padx=4)

# 定义文本框
# 有disabled、normal 两个状态值，默认为normal
log = tkinter.Text(window, state='normal', width=64, height=16)
log.grid(row=4, column=0, rowspan=1, columnspan=4, sticky=W, padx=4)

def start_udp():
tshark = e1.get()
eth_port = e2.get()
udp_second = e3.get()

my_cmd = tshark + ' --list-interfaces'
my_str = os.popen(my_cmd)
log.insert('insert', my_cmd+'\n')
log.insert('insert', my_str.read())
my_cmd = tshark + ' -f udp -i ' + eth_port + ' -a duration:' + udp_second + ' -w data.pcapng'
my_str = os.popen(my_cmd)
log.insert('insert', my_cmd+'\n')
log.insert('insert', my_str.read())

def start_oct():
log.insert('insert', 'Show OCT\n')
tel.tel24x("EnableRCVTChannelsInUdp=0")
tel.tel24x("EnableOCTChannelsInUdp=1")
start_udp()
Parse.Parse('data.pcapng')
DataPlot.DataPlot('data')

def start_rcvt():
log.insert('insert', 'Show RCVT\n')
tel.tel24x("EnableRCVTChannelsInUdp=1")
tel.tel24x("EnableOCTChannelsInUdp=0")
start_udp()
Parse.Parse('data.pcapng')
DataPlot.DataPlot('data')

b1 = tkinter.Button(window, text="Show OCT", command=start_oct)
b2 = tkinter.Button(window, text="Show RCVT", command=start_rcvt)
b1.grid(row=3, column=0)
b2.grid(row=3, column=1)

window.mainloop()

====DataPlot.py====

# from distutils.command.install_egg_info import to_filename

import sys
import numpy as np
# import glob as glob
from pathlib import Path
# from pathlib import PureWindowsPath
# from pathlib import PurePath
# from scapy.all import *
import dpkt as dpkt
import matplotlib.pyplot as plt

uint32x48_type = np.dtype((np.uint32, 48))
int32x3_type = np.dtype((np.int32, 3))
floatx3_type = np.dtype((np.float32, 3))
floatx6_type = np.dtype((np.float32, 6))
# RCVT data structure
frame1_type = np.dtype(
[('SmpCnt', np.int32),
('RcvtRaw', int32x3_type),
('SvRcvt', floatx3_type)])
# OCT data structure
frame2_type = np.dtype(
[('SmpCnt', np.int32),
('AfeRaw', uint32x48_type),
('AfeTemp', floatx6_type),
('SvProtMeas', floatx6_type)])
# OCT+RCVT data structure
frame3_type = np.dtype(
[('SmpCnt', np.int32),
('AfeRaw', uint32x48_type),
('AfeTemp', floatx6_type),
('SvProtMeas', floatx6_type),
('RcvtRaw', int32x3_type),
('SvRcvt', floatx3_type)])

def DataPlot(filename='data'):
filepath = Path(filename)
try:
bufs = filepath.open("rb").read()
except:
print("Missing data file!")
sys.exit()

pkt_size = int.from_bytes(bufs[0:4], 'little')

if np.mod(len(bufs) - 4, pkt_size) != 0:
print("Wrong data file!")
sys.exit()

GET_RCVT_UDP = 0
GET_OCT_UDP = 0

if pkt_size == frame1_type.itemsize:
frame_type = frame1_type
GET_RCVT_UDP = 1
print((len(bufs) - 4) // pkt_size, "RCVT Samples found")
elif pkt_size == frame2_type.itemsize:
frame_type = frame2_type
GET_OCT_UDP = 1
print((len(bufs) - 4) // pkt_size, "OCT Samples found")
elif pkt_size == frame3_type.itemsize:
frame_type = frame3_type
GET_RCVT_UDP = 1
GET_OCT_UDP = 1
print((len(bufs) - 4) // pkt_size, "RCVT+OCT Samples found")
else:
print("Wrong data file!")
sys.exit()

# Cut the file header
# Reshape data to defined struct
data = np.frombuffer(bufs[4:], dtype=frame_type).copy()

# plot the data
plt.figure('SmpCnt')
plt.plot(data['SmpCnt'])

if GET_RCVT_UDP:
# int_value = 0x80000000|(~tmp + 1)
get_int = np.bitwise_and(data['RcvtRaw'], 0x807FFFFF)
get_int[get_int >= 0x80000000] = np.bitwise_or(np.bitwise_not(get_int[get_int >= 0x80000000]) + 1, 0x80000000)

plt.figure('RcvtRawToInt32')
plt.plot(get_int)
plt.legend(['PhA', 'PhB', 'PhC'])

plt.figure('SvRcvt')
plt.plot(data['SvRcvt'])
plt.legend(['PhA', 'PhB', 'PhC'])

if GET_OCT_UDP:
plt.figure('AfeRaw')
plt.plot(data['AfeRaw'])
# plt.legend(['PhA', 'PhB', 'PhC'])

plt.figure('AfeTemp')
plt.plot(data['AfeTemp'])
plt.legend(['Afe1-Up', 'Afe1-Low', 'Afe2-Up', 'Afe2-Low', 'Afe3-Up', 'Afe3-Low'])

plt.figure('SvProtMeas')
plt.plot(data['SvProtMeas'])
plt.legend(['ProtA', 'ProtB', 'ProtC', 'MeasA', 'MeasB', 'MeasC'])

plt.show()

====Parse.py====

#from distutils.command.install_egg_info import to_filename

import sys
#import numpy as np
#import glob as glob
from pathlib import Path
#from pathlib import PureWindowsPath
#from pathlib import PurePath
#from scapy.all import *
import dpkt as dpkt

def Parse(filename='data.pcapng'):
filepath = Path(filename)

try:
f = filepath.open("rb")
pcap = dpkt.pcapng.Reader(f)
print("reader created")
packets = [ dpkt.ethernet.Ethernet(buf) for _, buf in pcap if dpkt.ethernet.Ethernet(buf).type == dpkt.ethernet.ETH_TYPE_IP and dpkt.ethernet.Ethernet(buf).data.p == dpkt.ip.IP_PROTO_UDP]
print("packets read")
bufs = [x.data.data.data for x in packets if x.data.data.dport==6510]
except:
print('No valid IO24x UDP found!')
sys.exit()

if(len(bufs) < 3):
print('No valid IO24x UDP found!')
sys.exit()

pkt_len0=len(bufs[0])
pkt_len1=len(bufs[1])
cnt0=int.from_bytes(bufs[0][0:4],'little')
cnt1=int.from_bytes(bufs[1][0:4],'little')
cnt2=int.from_bytes(bufs[2][0:4],'little')
if cnt1>cnt0:
pkt_size=int(pkt_len0/(cnt1-cnt0))
else:
pkt_size=int(pkt_len1/(cnt2-cnt1))

print("packets filtered")
buf = b''.join(bufs)
print("packets joined")

#data = np.frombuffer(buf,dtype=frame2_type).copy()
#print("packets parsed")
#d = np.unwrap(data[:]['SmpCnt']/16000*np.pi*2)/np.pi/2*16000
#data[:]['SmpCnt'] = np.round(d)

buf_result =int.to_bytes(pkt_size,4,'little') + buf
to_filepath = filepath.with_suffix("")
with to_filepath.open('wb') as f:
f.write(buf_result)
print("data written")

 

====tel.py====

import telnetlib
import time
import sys

def tel24x(arg1='ver'):
command = arg1

host = "172.16.60.50"
#username = "admin"
#password = "admin"

tn = telnetlib.Telnet(host)

# 30.telnet登陆
# tn.read_until(b'Username: ', timeout=3)
# tn.write(username.encode('ascii') + b"\n")
# tn.read_until(b"Password: ", timeout=3)
# tn.write(password.encode('ascii') + b"\n")

time.sleep(0.3)
tn.write(command.encode('ascii') + b"\n")
time.sleep(0.2)

v_result = tn.read_very_eager().decode('ascii')

print(v_result)

标签：UDP,plt,Python,data,telnet,np,import,type
From： https://www.cnblogs.com/likelykids/p/16982549.html