forked from gatsby/df_datalink
563 lines
21 KiB
Python
563 lines
21 KiB
Python
# from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QLabel, QSlider, QLineEdit, QGridLayout, QPushButton, QCheckBox, QHBoxLayout, QInputDialog)
|
|
# from PyQt5.QtCore import pyqtSignal
|
|
# from PyQt5.QtGui import QDoubleValidator
|
|
import os
|
|
from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QLabel, QSlider, QLineEdit,
|
|
QGridLayout, QPushButton, QCheckBox, QHBoxLayout, QInputDialog, QComboBox, QFileDialog)
|
|
from PyQt5.QtCore import pyqtSignal, Qt, QTimer
|
|
from PyQt5.QtGui import QDoubleValidator, QPixmap
|
|
|
|
try:
|
|
from i18n import _
|
|
import utils_ui
|
|
import utils
|
|
from widgets import EditRemarDialog
|
|
except Exception:
|
|
from COMTool.i18n import _
|
|
from COMTool import utils_ui
|
|
from COMTool import utils
|
|
from COMTool.widgets import EditRemarDialog
|
|
|
|
# from .graph_widget_metasenselite import Graph_MetaSenseLite
|
|
from .graph_widgets_base import Graph_Widget_Base
|
|
import pyqtgraph as pg
|
|
from struct import unpack, pack
|
|
import time
|
|
|
|
|
|
class Graph_Image(Graph_Widget_Base):
|
|
id = "image"
|
|
updateSignal = pyqtSignal(bytes)
|
|
startTimeoutSignal = pyqtSignal()
|
|
stopTimeoutSignal = pyqtSignal()
|
|
|
|
def __init__(self, parent=None, hintSignal=lambda type, title, msg: None,
|
|
rmCallback=lambda widget: None, send=lambda x: None, config=None):
|
|
# 初始化配置，默认缩放倍数为10
|
|
default = {"scale_factor": 10}
|
|
if config:
|
|
default.update(config)
|
|
super().__init__(parent, hintSignal, rmCallback, send, default, default)
|
|
|
|
self.image_data = bytearray()
|
|
# 从配置获取缩放倍数
|
|
self.scale_factor = self.config.get("scale_factor", 10)
|
|
|
|
layout = QVBoxLayout()
|
|
self.setLayout(layout)
|
|
|
|
self.imageLabel = QLabel("[ Waiting for image... ]")
|
|
self.imageLabel.setAlignment(Qt.AlignCenter)
|
|
layout.addWidget(self.imageLabel)
|
|
|
|
self.clearBtn = QPushButton("Clear Image")
|
|
self.clearBtn.clicked.connect(self.clear)
|
|
layout.addWidget(self.clearBtn)
|
|
|
|
self.updateSignal.connect(self.update_image)
|
|
|
|
self.timeoutTimer = QTimer(self)
|
|
self.timeoutTimer.setInterval(2000) # 2 seconds timeout
|
|
self.timeoutTimer.timeout.connect(self.clear_buffer)
|
|
|
|
self.startTimeoutSignal.connect(lambda: self.timeoutTimer.start())
|
|
self.stopTimeoutSignal.emit()
|
|
|
|
def clear(self):
|
|
self.image_data.clear()
|
|
self.imageLabel.clear()
|
|
self.imageLabel.setText("[ Cleared ]")
|
|
self.stopTimeoutSignal.emit()
|
|
|
|
def clear_buffer(self):
|
|
self.image_data.clear()
|
|
self.imageLabel.setText("[ Timeout: buffer cleared ]")
|
|
self.stopTimeoutSignal.emit()
|
|
|
|
def onData(self, data: bytes):
|
|
self.stopTimeoutSignal.emit() # Stop any running timer when data is received
|
|
self.image_data.extend(data)
|
|
|
|
# Check for common image signatures
|
|
if self._is_complete_jpeg(self.image_data) or \
|
|
self._is_complete_png(self.image_data) or \
|
|
self._is_complete_bmp(self.image_data):
|
|
self.updateSignal.emit(bytes(self.image_data))
|
|
self.image_data.clear()
|
|
else:
|
|
self.startTimeoutSignal.emit() # Restart timer only if not complete
|
|
|
|
def _is_complete_jpeg(self, data):
|
|
return data[:2] == b'\xFF\xD8' and data[-2:] == b'\xFF\xD9'
|
|
|
|
def _is_complete_png(self, data):
|
|
return data[:8] == b'\x89PNG\r\n\x1a\n' and b'IEND' in data[-20:]
|
|
|
|
def _is_complete_bmp(self, data):
|
|
return data[:2] == b'BM' and len(data) >= int.from_bytes(data[2:6], 'little')
|
|
|
|
def update_image(self, img_bytes: bytes):
|
|
pixmap = QPixmap()
|
|
if pixmap.loadFromData(img_bytes):
|
|
# 使用配置的缩放倍数进行缩放
|
|
scaled_width = pixmap.width() * self.scale_factor
|
|
scaled_height = pixmap.height() * self.scale_factor
|
|
scaled = pixmap.scaled(
|
|
scaled_width,
|
|
scaled_height,
|
|
Qt.KeepAspectRatio,
|
|
Qt.SmoothTransformation
|
|
)
|
|
self.imageLabel.setPixmap(scaled)
|
|
else:
|
|
self.imageLabel.setText("[ Failed to decode image ]")
|
|
|
|
class Graph_Plot(Graph_Widget_Base):
|
|
updateSignal = pyqtSignal(dict)
|
|
id = "plot"
|
|
|
|
def __init__(self, parent=None, hintSignal=lambda type, title, msg: None, rmCallback=lambda widget: None, send=lambda x: None, config=None):
|
|
default = {
|
|
"xRange": 10,
|
|
"xRangeEnable": True,
|
|
"header": "\\xAA\\xCC\\xEE\\xBB",
|
|
"binary_protocol": False,
|
|
}
|
|
super().__init__(parent, hintSignal=hintSignal, rmCallback=rmCallback,
|
|
send=send, config=config, defaultConfig=default)
|
|
self.headerBytes = utils.str_to_bytes(
|
|
self.config["header"], escape=True, encoding="utf-8")
|
|
self.layout = QGridLayout()
|
|
self.setLayout(self.layout)
|
|
self.plotWin = pg.GraphicsLayoutWidget()
|
|
self.plotWin.setMinimumHeight(200)
|
|
self.default_x = 0;
|
|
pg.setConfigOptions(antialias=True)
|
|
rmBtn = QPushButton(_("Remove"))
|
|
clearBtn = QPushButton(_("Clear"))
|
|
rangeLabel = QLabel(_("Range:"))
|
|
rangeConf = QLineEdit(str(self.config["xRange"]))
|
|
rangeEnable = QCheckBox(_("Enable"))
|
|
rangeEnable.setChecked(self.config["xRangeEnable"])
|
|
self.binary_protocol = QCheckBox(_("Binary Protocol"))
|
|
headerLabel = QLabel(_("Header:"))
|
|
headerConf = QLineEdit(self.config["header"])
|
|
self.headerBtn = QPushButton(_("Set"))
|
|
hint = _("Protocol: header + 1Byte name length + name + 8Bytes x(double) + 8Bytes y(double) + 1Byte sum\n"
|
|
"Protocol example code see help")
|
|
headerConf.setToolTip(hint)
|
|
self.binary_widgets = [headerLabel, headerConf, self.headerBtn]
|
|
self.binary_protocol.setToolTip(_('Check this if you want to use binary protocol, or use ASCII protocol(e.g. "$data1,10.0,200.0\\n"), more see help button on the top of window'))
|
|
self.show_binary_protocol_widgets(self.config["binary_protocol"])
|
|
headerLabel.setToolTip(hint)
|
|
self.headerBtn.setToolTip(hint)
|
|
validator = QDoubleValidator()
|
|
rangeConf.setValidator(validator)
|
|
self.layout.addWidget(self.plotWin, 0, 0, 1, 3)
|
|
self.layout.addWidget(rmBtn, 1, 0, 1, 1)
|
|
self.layout.addWidget(clearBtn, 1, 2, 1, 1)
|
|
self.layout.addWidget(rangeLabel, 2, 0, 1, 1)
|
|
self.layout.addWidget(rangeConf, 2, 1, 1, 1)
|
|
self.layout.addWidget(rangeEnable, 2, 2, 1, 1)
|
|
self.layout.addWidget(self.binary_protocol, 3, 0, 1, 1)
|
|
self.layout.addWidget(headerLabel, 4, 0, 1, 1)
|
|
self.layout.addWidget(headerConf, 4, 1, 1, 1)
|
|
self.layout.addWidget(self.headerBtn, 4, 2, 1, 1)
|
|
self.resize(600, 400)
|
|
self.p = self.plotWin.addPlot(colspan=2)
|
|
# self.p.setLabel('bottom', 'x', '')
|
|
self.p.addLegend()
|
|
self.p.setXRange(0, self.config["xRange"])
|
|
self.updateSignal.connect(self.update)
|
|
self.curves = {}
|
|
self.rawData = b''
|
|
self.data = {}
|
|
self.builtinColors = [
|
|
"#BD4B4B",
|
|
"#3BB273",
|
|
"#FFFFFA",
|
|
"#307473",
|
|
"#3C6997",
|
|
"#746D75",
|
|
"#228CDB",
|
|
"#824C71",
|
|
"#7768AE",
|
|
"#DC6BAD",
|
|
"#607d8b",
|
|
"#F18701",
|
|
"#912F40",
|
|
"#414288",
|
|
"#ED4D6E",
|
|
"#FFD29D",
|
|
"#B56576",
|
|
"#503B31",
|
|
"#93E1D8",
|
|
"#596157",
|
|
]
|
|
self.notUsedColors = self.builtinColors.copy()
|
|
self.colors = {
|
|
|
|
}
|
|
rangeConf.textChanged.connect(self.setRange)
|
|
rangeEnable.clicked.connect(
|
|
lambda: self.setEnableRange(rangeEnable.isChecked()))
|
|
self.headerBtn.clicked.connect(
|
|
lambda: self.setHeader(headerConf.text()))
|
|
rmBtn.clicked.connect(self.remove)
|
|
headerConf.textChanged.connect(self.headerChanged)
|
|
clearBtn.clicked.connect(self.clear)
|
|
self.binary_protocol.clicked.connect(self.en_binary_protocol)
|
|
|
|
def remove(self):
|
|
self.rmCallback(self)
|
|
|
|
def en_binary_protocol(self):
|
|
self.rawData = b''
|
|
self.config["binary_protocol"] = self.binary_protocol.isChecked()
|
|
self.show_binary_protocol_widgets(self.config["binary_protocol"])
|
|
|
|
def show_binary_protocol_widgets(self, show=True):
|
|
for w in self.binary_widgets:
|
|
w.setVisible(show)
|
|
|
|
def clear(self):
|
|
self.data = {}
|
|
self.curves = {}
|
|
self.notUsedColors = self.builtinColors.copy()
|
|
self.colors = {}
|
|
self.p.clear()
|
|
|
|
def setRange(self, text):
|
|
if text:
|
|
self.config["xRange"] = float(text)
|
|
|
|
def setEnableRange(self, en):
|
|
self.config["xRangeEnable"] = en
|
|
|
|
def headerChanged(self, text):
|
|
if self.config["header"] != text:
|
|
self.headerBtn.setText(_("Set") + " *")
|
|
else:
|
|
self.headerBtn.setText(_("Set"))
|
|
|
|
def setHeader(self, text):
|
|
if text:
|
|
try:
|
|
textBytes = utils.str_to_bytes(
|
|
text, escape=True, encoding="utf-8")
|
|
self.config["header"] = text
|
|
self.headerBytes = textBytes
|
|
self.headerBtn.setText(_("Set"))
|
|
except Exception:
|
|
self.hintSignal.emit("error", _("Error"), _("Format error"))
|
|
|
|
def decodeData(self, data: bytes):
|
|
'''
|
|
@data bytes, protocol:
|
|
| header(4B) | line name len(1B) | line name | x(8B) | y(8B) | sum(1B) |
|
|
| AA CC EE BB | 1~255 | roll | double | double | uint8 |
|
|
@return haveFrame, dict {
|
|
"name": {
|
|
"x": [],
|
|
"y": []
|
|
}
|
|
}
|
|
'''
|
|
# append data
|
|
self.rawData += data
|
|
# find header
|
|
header = self.headerBytes
|
|
idx = self.rawData.find(header)
|
|
if idx < 0:
|
|
return False, self.data
|
|
self.rawData = self.rawData[idx:]
|
|
# check data length
|
|
nameLen = unpack("B", self.rawData[len(header): len(header) + 1])[0]
|
|
frameLen = len(header) + nameLen + 18 # 5 + nameLen + 16 + 1
|
|
if len(self.rawData) < frameLen:
|
|
return False, self.data
|
|
# get data
|
|
frame = self.rawData[:frameLen]
|
|
self.rawData = self.rawData[frameLen:]
|
|
_sum = frame[-1]
|
|
# check sum
|
|
if _sum != sum(frame[:frameLen - 1]) % 256:
|
|
return True, self.data
|
|
name = frame[len(header) + 1: len(header) +
|
|
1 + nameLen].decode("utf-8")
|
|
x = unpack("d", frame[-17:-9])[0]
|
|
y = unpack("d", frame[-9:-1])[0]
|
|
if not name in self.data:
|
|
self.data[name] = {
|
|
"x": [x],
|
|
"y": [y]
|
|
}
|
|
else:
|
|
self.data[name]["x"].append(x)
|
|
self.data[name]["y"].append(y)
|
|
return True, self.data
|
|
|
|
def decodeDataAscii(self, data: bytes):
|
|
'''
|
|
@data bytes, protocol:
|
|
$[line name],[x],[y]<,checksum>\n
|
|
or $[line name],[y]
|
|
"$" means start of frame, "," means separator,
|
|
checksum is optional, checksum is sum of all bytes in frame except ",checksum".
|
|
e.g.
|
|
"$roll,2.0\n"
|
|
"$roll,1.0,2.0\n"
|
|
"$pitch,1.0,2.0\r\n"
|
|
"$pitch,1.0,2.0,179\n" , the 179 = sum(ord(c) for c in "$pitch,1.0,2.0") % 256
|
|
@return haveFrame, dict {
|
|
"name": {
|
|
"x": [],
|
|
"y": []
|
|
}
|
|
}
|
|
'''
|
|
# append data
|
|
self.rawData += data
|
|
# find header
|
|
header = b"$"
|
|
end = b"\n"
|
|
idx = self.rawData.find(header)
|
|
if idx < 0:
|
|
return False, self.data
|
|
self.rawData = self.rawData[idx:]
|
|
# find \n
|
|
idx = self.rawData.find(end)
|
|
if idx < 0:
|
|
return False, self.data
|
|
# get data
|
|
frame = self.rawData[1:idx] # pitch,1.0,2.0 pitch,1.0,2.0,179
|
|
self.rawData = self.rawData[idx + 1:]
|
|
items = frame.split(b",")
|
|
if len(items) != 2 and len(items) != 3 and len(items) != 4:
|
|
print("-- format error, frame item len(%s, should 3 or 4) error: %s" % (len(items), frame))
|
|
return False, self.data
|
|
try:
|
|
if len(items) == 2:
|
|
x = float(self.default_x)
|
|
y = float(items[1])
|
|
self.default_x += 1
|
|
else:
|
|
x = float(items[1])
|
|
y = float(items[2])
|
|
except Exception:
|
|
print("-- x or y format error, frame: %s" % frame)
|
|
return False, self.data
|
|
# checksum
|
|
if len(items) == 4:
|
|
try:
|
|
_sum = int(items[3])
|
|
except Exception:
|
|
print("-- checksum format error({} not int), frame: {}".format(items[3], frame))
|
|
return False, self.data
|
|
# get last , index
|
|
idx = frame.rfind(b",")
|
|
if sum(header + frame[:idx]) % 256 != _sum:
|
|
print("-- checksum error")
|
|
return False, self.data
|
|
name = items[0].decode("utf-8")
|
|
if not name in self.data:
|
|
self.data[name] = {
|
|
"x": [x],
|
|
"y": [y]
|
|
}
|
|
else:
|
|
self.data[name]["x"].append(x)
|
|
self.data[name]["y"].append(y)
|
|
return True, self.data
|
|
|
|
def pickColor(self, name: str):
|
|
if name in self.colors:
|
|
return self.colors[name]
|
|
else:
|
|
if not self.notUsedColors:
|
|
self.notUsedColors = self.builtinColors.copy()
|
|
color = self.notUsedColors.pop(0)
|
|
self.colors[name] = color
|
|
return color
|
|
|
|
def update(self, data: dict):
|
|
for k, v in data.items():
|
|
if not k in self.curves:
|
|
print("add curve:", k)
|
|
color = self.pickColor(k)
|
|
self.colors[k] = color
|
|
self.curves[k] = self.p.plot(pen=pg.mkPen(color=color, width=2),
|
|
name=k,)
|
|
# symbolBrush=color, symbolPen='w', symbol='o', symbolSize=1)
|
|
if self.config["xRangeEnable"] and self.config["xRange"] > 0:
|
|
self.p.setXRange(
|
|
v["x"][-1] - self.config["xRange"], v["x"][-1])
|
|
self.curves[k].setData(x=v["x"], y=v["y"])
|
|
|
|
def onData(self, data: bytes):
|
|
while 1:
|
|
haveFrame, allData = self.decodeData(data) if self.config["binary_protocol"] else self.decodeDataAscii(data)
|
|
if not haveFrame:
|
|
break
|
|
data = b''
|
|
self.updateSignal.emit(allData)
|
|
|
|
|
|
class Graph_Button(Graph_Widget_Base):
|
|
class Button(QWidget):
|
|
def __init__(self, rmCallback, addCallback, clickCallback, changeCallback,
|
|
value=("", "hello", "hello", "fa.circle", [])):
|
|
super().__init__()
|
|
if not value[0]:
|
|
self.id = str(int(time.time() * 1000))
|
|
else:
|
|
self.id = value[0]
|
|
self.rmCallback = rmCallback
|
|
self.changeCallback = changeCallback
|
|
self.setProperty("class", "graphBtn")
|
|
layout = QHBoxLayout()
|
|
layoutSetting = QVBoxLayout()
|
|
self.setLayout(layout)
|
|
self.button = QPushButton(value[2])
|
|
self.button.setProperty("class", "bigBtn")
|
|
self.value = value[2]
|
|
self.icon = value[3]
|
|
self.shortcut = value[4]
|
|
layout.addLayout(layoutSetting)
|
|
layout.addWidget(self.button)
|
|
self.addBtn = QPushButton()
|
|
self.rmBtn = QPushButton()
|
|
self.editBtn = QPushButton()
|
|
self.addBtn.setProperty("class", "smallBtn3")
|
|
self.rmBtn.setProperty("class", "smallBtn3")
|
|
self.editBtn.setProperty("class", "smallBtn3")
|
|
utils_ui.setButtonIcon(self.addBtn, "fa.plus")
|
|
utils_ui.setButtonIcon(self.rmBtn, "fa.minus")
|
|
utils_ui.setButtonIcon(self.editBtn, "fa.edit")
|
|
if self.icon:
|
|
utils_ui.setButtonIcon(self.button, self.icon)
|
|
if self.shortcut:
|
|
tip = self.getShortcutTip()
|
|
self.button.setToolTip(tip)
|
|
layoutSetting.addStretch()
|
|
layoutSetting.addWidget(self.rmBtn)
|
|
layoutSetting.addWidget(self.addBtn)
|
|
layoutSetting.addWidget(self.editBtn)
|
|
self.rmBtn.clicked.connect(self.onRm)
|
|
self.addBtn.clicked.connect(lambda: addCallback(self))
|
|
self.editBtn.clicked.connect(self.editButton)
|
|
self.button.clicked.connect(lambda: clickCallback(
|
|
self, utils.str_to_bytes(self.value, escape=True, encoding="utf-8")))
|
|
|
|
def getInfo(self):
|
|
return [self.id, self.button.text(), self.value, self.icon, self.shortcut]
|
|
|
|
def onRm(self):
|
|
utils_ui.clearButtonIcon(self.addBtn)
|
|
utils_ui.clearButtonIcon(self.rmBtn)
|
|
utils_ui.clearButtonIcon(self.editBtn)
|
|
utils_ui.clearButtonIcon(self.button)
|
|
self.rmCallback(self)
|
|
|
|
def editButton(self):
|
|
ok, remark, value, icon, shortcut = EditRemarDialog(
|
|
self.button.text(), self.icon, shortcut=self.shortcut, value=self.value).exec()
|
|
if ok:
|
|
self.value = value
|
|
self.icon = icon
|
|
self.shortcut = shortcut
|
|
self.button.setText(remark)
|
|
if self.shortcut:
|
|
tip = self.getShortcutTip()
|
|
self.button.setToolTip(tip)
|
|
utils_ui.setButtonIcon(self.button, icon)
|
|
self.changeCallback(self)
|
|
|
|
def getShortcutTip(self):
|
|
return _("Shortcut") + ": " + "+".join((name for v, name in self.shortcut))
|
|
|
|
id = "button"
|
|
|
|
def __init__(self, parent=None, hintSignal=lambda type, title, msg: None, rmCallback=lambda widget: None, send=lambda x: None, config=None):
|
|
default = {
|
|
"items": []
|
|
}
|
|
super().__init__(parent, hintSignal, rmCallback, send, config, default)
|
|
self.layout = QHBoxLayout()
|
|
self.setLayout(self.layout)
|
|
if not self.config["items"]:
|
|
button = Graph_Button.Button(
|
|
self.onRm, self.onAdd, self.onClick, self.onChange)
|
|
self.layout.addWidget(button)
|
|
self.buttons = [button]
|
|
self.config["items"] = [button.getInfo()]
|
|
else:
|
|
self.buttons = []
|
|
for item in self.config["items"]:
|
|
button = Graph_Button.Button(
|
|
self.onRm, self.onAdd, self.onClick, self.onChange, value=item)
|
|
self.layout.addWidget(button)
|
|
self.buttons.append(button)
|
|
self.pressedKeys = []
|
|
|
|
def onRm(self, widget):
|
|
'''
|
|
@widget Graph_Button.Button
|
|
'''
|
|
item = widget.getInfo()
|
|
for item_ in self.config["items"]:
|
|
if item_[0] == item[0]: # id
|
|
self.config["items"].remove(item_)
|
|
break
|
|
self.buttons.remove(widget)
|
|
widget.setParent(None)
|
|
if not self.buttons:
|
|
self.rmCallback(self)
|
|
|
|
def onAdd(self, widget):
|
|
'''
|
|
@widget Graph_Button.Button
|
|
'''
|
|
button = Graph_Button.Button(
|
|
self.onRm, self.onAdd, self.onClick, self.onChange)
|
|
self.layout.addWidget(button)
|
|
self.buttons.append(button)
|
|
self.config["items"].append(button.getInfo())
|
|
|
|
def onClick(self, widget, data: bytes):
|
|
self.send(data)
|
|
|
|
def onChange(self, widget):
|
|
self.config["items"][self.buttons.index(widget)] = widget.getInfo()
|
|
|
|
def onKeyPressEvent(self, event):
|
|
key = event.key()
|
|
self.pressedKeys.append(key)
|
|
for uid, name, value, icon, shortcut in self.config["items"]:
|
|
if not shortcut:
|
|
continue
|
|
if len(shortcut) == len(self.pressedKeys):
|
|
same = True
|
|
for i in range(len(shortcut)):
|
|
if shortcut[i][0] != self.pressedKeys[i]:
|
|
same = False
|
|
break
|
|
if same:
|
|
value = utils.str_to_bytes(
|
|
value, escape=True, encoding="utf-8")
|
|
self.send(value)
|
|
|
|
def onKeyReleaseEvent(self, event):
|
|
key = event.key()
|
|
if key in self.pressedKeys:
|
|
self.pressedKeys.remove(key)
|
|
|
|
|
|
graphWidgets = {
|
|
Graph_Plot.id: Graph_Plot,
|
|
Graph_Button.id: Graph_Button,
|
|
Graph_Image.id: Graph_Image
|
|
# Graph_DragTof.id: Graph_DragTof,
|
|
# Graph_MetaSenseLite.id: Graph_MetaSenseLite,
|
|
}
|