USB 数据记录 Ubuntu MATE

2024-5-24 • tag-icon

text-processing

使用 Ubuntu Mate、树莓派。我对编程很陌生，对 bash 也很陌生。简单地说：USB 数据到 CSV，即使没有脚本我也可以尝试任何事情。

编辑：设备将处于循环状态，因此停止其接收数据是不允许的。实时数据过滤，仅接收真实数据。

对于我的另一个类似问题：Linux终端输出到文件，但是被过滤了？

我有制造 RF 接收器的公司提供的 python 脚本（脚本在最后）。这很可能是用户错误。

这是我在终端写的：

python3 'home/#USER/Desktop/python/script.py' /dev/ttyUSB0 | filter.pl    >> output.csv
[1+] Stopped
Filter.pl: command not found

使用 sed 终端：

python3 'home/#USER/Desktop/python/script.py' /dev/ttyUSB0 | sed -n "s/^b';[0-9]\{3\}];\(.\+\);'/\1/p" >> output_file

结果：闪烁输入 thingy 并在 CTRL + C 之后

Exception ignored in: <_io.TextIOrapper name='<stdout>' mode='w' ecpmdomg='UTF-8'>  
BrokenPipeError:[Errno32] Broken pipe.

注意：我只需要更长的部分，从那里我将对其进行更多解析，然后将其发送到 AT&T M2X 在线服务。

b';"Anything from 1-9999";'
b';311;'

b';312;'

b';312;00000000;036;552;1014f49;3020;2659;6294;1049;2659;S;'

b';313;'

脚本.py：

#! / Usr / bin / python
# Coding: UTF-8

################################################## #########################
# (C) Tokyo Cosmos Electric, Inc. (TOCOS) - all rights reserved.
# Terms and Conditions:
# - This source code, as long as the Tokyo Cosmos Electric Co., Ltd.         copyright separately there is no source code license description
# We will not hold.
# - This source code is no warranty-free support. Any damage which the  present source code and product
Tokyo Cosmos Electric Co., Ltd. does not guarantee also #. Report of trouble etc. will be welcome.
# - This source code is published on the premise that run with TWE series  Tokyo Cosmos Electric Co., Ltd. to sell
#     doing.
################################################## #########################

### Script to read the TWE-Lite standard application
# ? this script is read-only, to do the reading and writing both will require processing by multiple threads.

from serial import *
from sys import stdout, stdin, stderr, exit

Confirmation of # parameter
# First argument: serial port name
! if len (sys.argv) = 2:
    print "% s {serial port name}"% sys.argv [0]
    exit (1)

# Open a serial port
try:
ser = Serial (sys.argv [1], 115200)
print "open serial port:% s"% sys.argv [1]
except:
    print "can not open serial port:% s"% sys.argv [1]
exit (1)

# Display of other messages (output the payload)
def printPayload (l):
    if len (l) <3: return False # check of data size

print "command = 0x% 02x (other)"% l [1]
print "src = 0x% 02x"% l [0]

# Directly outputs the payload
print "payload =",
for c in l [2:]:
    print "% 02x"% c,
print "(hex)"
return True

# 0x81 interpretation of the message
def printPayload_0x81 (l):
    if len (l) = 23:! return False # check of data size

ladr = l [5] << 24 | l [6] << 16 | l [7] << 8 | l [8]
print "command = 0x% 02x (data arrival)"% l [1]
print "src = 0x% 02x"% l [0]
print "src long = 0x% 08x"% ladr
print "dst = 0x% 02x"% l [9]
print "pktid = 0x% 02x"% l [2]
print "prtcl ver = 0x% 02x"% l [3]
print "LQI =% d /% .2f [dbm]"% (l [4], (7 * l [4] -1970) / 20.)
ts = l [10] << 8 | l [11]
print "time stmp =% .3f [s]"% (ts / 64.0)
print "relay flg =% d"% l [12]
vlt = l [13] << 8 | l [14]
print "volt =% 04d [mV]"% vlt

Data of # DI1..4
dibm = l [16]
dibm_chg = l [17]
di = {} # of the current state
di_chg = {} # 1 Once in Lo (1) at least once
for i in range (1,5):
    di [i] = 0 if (dibm & 0x1) == 0 else 1
    di_chg [i] = 0 if (dibm_chg & 0x1) == 0 else 1
    dibm >> = 1
    dibm_chg >> = 1
    pass

print "DI1 =% d /% d DI2 =% d /% d DI3 =% d /% d DI4 =% d /% d"% (di [1], di_chg [1], di [2], di_chg [ 2], di [3], di_chg [3], di [4], di_chg [4])

Data of # AD1..4
ad = {}
er = l [22]
for i in range (1,5):
    av = l [i + 18 - 1]
    if av == 0xFF:
        # AD and if the port is unused treatment (approximately 2V or more) -1
        ad [i] = -1
    else:
        # Calculation, including the correction bits
        ad [i] = ((av * 4) + (er & 0x3)) * 4
    er >> = 2
print "AD1 =% 04d AD2 =% 04d AD3 =% 04d AD4 =% 04d [mV]"% (ad [1], ad [2], ad [3], ad [4])

return True

# Interpret the data line by line
while True:
    line = ser.readline (). rstrip () # to read in one line units, and remove the trailing line feed code (blocking read)

if len (line)> 0 and line [0] == ':':
    print "\ n% s"% line
else:
    continue

try:
    lst = map (ord, line [1:]. decode ('hex')) # convert the HEX string after decoding the string, to each ord and () the list
    csum = sum (lst) & 0xff # checksum if 0 by adding a total of 8bit calculation OK
    lst.pop () remove the # checksum from the list
    if csum == 0:
        if lst [1] == 0x81:
            printPayload_0x81 (lst) # reception of IO-related data
        else:
            printPayload (lst) # Other data reception
    else:
        print "checksum ng"
except:
    print "skip" # when an error

答案1

在你的第一个样本中

Filter.pl: command not found

您的 shell 未找到 filter.pl 脚本。

您将需要指定脚本的完整路径。如果它在当前目录中，请./filter.pl像这样使用：

/home/USER/Desktop/python/script.py /dev/ttyUSB0 | ./filter.pl >> output.csv

答案2

看来这比我想象的要容易。这是为了帮助有类似问题的人。

在终端中写入

python3 -u ./file.py | tee ./output.file

我现在正在研究如何使用 BASH 附加和过滤。

希望这对某人有帮助。

编辑：附加带有“a”的作品。在 tee 之后添加 -a。

python3 -u ./file.py | tee -a ./output.file

答案1

答案2

相关内容