84 lines
2.1 KiB
Python
84 lines
2.1 KiB
Python
import sigrok.core as sr
|
|
import numpy as np
|
|
import ripyl
|
|
import ripyl.protocol.uart as uart
|
|
import socket
|
|
|
|
MCAST_GRP = '239.255.255.255'
|
|
MCAST_PORT = 5007
|
|
MULTICAST_TTL = 2
|
|
|
|
socket_path = '/tmp/logic_stream'
|
|
|
|
|
|
|
|
driver='fx2lafw'
|
|
|
|
context = sr.Context.create()
|
|
devices = context.drivers[driver].scan()
|
|
device = devices[0]
|
|
device.open()
|
|
|
|
|
|
session = context.create_session()
|
|
session.add_device(device)
|
|
|
|
output_format = 'bits'
|
|
output = context.output_formats[output_format].create_output(device)
|
|
#baud_rates_test = (2400, 4800, 9600, 19200, 28800, 38400, 56000, 57600, 115200, 128000, 153600, 230400, 256000)
|
|
baud_rates_test = (4800, 9600, 19200, 28800, 38400, 56000, 57600, 115200, 128000, 153600, 230400, 256000)
|
|
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
|
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, MULTICAST_TTL)
|
|
|
|
|
|
pack_size = 256
|
|
idx = 0
|
|
|
|
import pickle
|
|
import time
|
|
while True:
|
|
data = np.arange(idx*pack_size, (idx+1)*pack_size, 1)
|
|
idx +=1
|
|
sock.sendto(data, (MCAST_GRP,MCAST_PORT))
|
|
time.sleep(0.02)
|
|
# bit_data = np.ascontiguousarray(np.unpackbits(data).reshape((-1,8))[:,::-1])
|
|
|
|
# sock.sendto(bit_data, (MCAST_GRP,MCAST_PORT))
|
|
|
|
|
|
|
|
|
|
#~/sr/bin/sigrok-cli --config samplerate=500000 --driver=fx2lafw --samples 1M -P uart:baudrate=57600:tx=D0 -P uart:baudrate=9600:tx=D0 -A uart=tx-data:tx-warning
|
|
|
|
session.run()
|
|
session.stop()
|
|
#device.close()
|
|
# %%
|
|
import pickle
|
|
with open('data.p','rb') as ff:
|
|
bit_data = pickle.load(ff)
|
|
# %%
|
|
baud_rates_test = (2400, 4800, 9600, 19200, 28800, 38400, 56000, 57600, 115200, 128000, 153600, 230400, 256000)
|
|
import ripyl
|
|
import ripyl.protocol.uart as uart
|
|
txd = ripyl.streaming.samples_to_sample_stream(bit_data[:,0], 1/1000000)
|
|
for br in baud_rates_test:
|
|
# out = uart.uart_decode(txd,baud_rate=57600)
|
|
try:
|
|
txd = ripyl.streaming.samples_to_sample_stream(bit_data[:,0], 1/1000000)
|
|
out = uart.uart_decode(txd,baud_rate=br)
|
|
dat = list(out)
|
|
for y in dat:
|
|
print(chr(y.data),end='')
|
|
except:
|
|
pass
|
|
#print('Session running',session.is_running())
|
|
#session.stop()
|
|
#device.close()
|
|
|
|
# %%
|
|
|
|
|