184 lines
8.5 KiB
Python
184 lines
8.5 KiB
Python
"""Reusable byte/word terminal page with an injected protocol or word handler."""
|
|
from PySide6.QtCore import Qt, Signal
|
|
from PySide6.QtWidgets import (
|
|
QWidget, QVBoxLayout, QLabel, QHBoxLayout, QGroupBox, QGridLayout,
|
|
QLineEdit, QPushButton, QScrollArea, QSizePolicy,
|
|
)
|
|
|
|
from .rawprotocol import RawProtocolWidget
|
|
from .serialbus import SerialBus, SerialTab
|
|
|
|
|
|
class ByteWorkPanel(QScrollArea):
|
|
"""Embed the complete terminal page in any layout or QTabWidget.
|
|
|
|
protocol supplies request_async/cancel_async, peek/poke and RAW methods.
|
|
device_address can be an integer or a callable returning the current
|
|
controller address. An optional start_word(write, address, value) handler
|
|
lets an existing tab keep its own memory queue; it must report results with
|
|
set_read_value() and set_status(). Otherwise this panel performs asynchronous
|
|
reads and verifies writes by reading the same word back.
|
|
"""
|
|
|
|
status_changed = Signal(str)
|
|
|
|
def __init__(self, protocol, parent=None, *, device_address=10,
|
|
start_word=None, macros_file=None, default_address="$1000",
|
|
hint=None, initial_status="Нет данных от контроллера"):
|
|
super().__init__(parent)
|
|
self.protocol = protocol
|
|
self.device_address = device_address
|
|
self._start_word = start_word
|
|
self._word_running = False
|
|
self._word_generation = 0
|
|
self.setWidgetResizable(True)
|
|
content = QWidget()
|
|
self.setWidget(content)
|
|
layout = QVBoxLayout(content)
|
|
self.hint = QLabel(hint if hint is not None else
|
|
"Адрес слова — 32 бита; значение — 16 бит (0000…FFFF). "
|
|
"Адрес: десятичный, $HEX или 0xHEX. Запись проверяется чтением.")
|
|
self.hint.setWordWrap(True)
|
|
layout.addWidget(self.hint)
|
|
panels = QHBoxLayout()
|
|
for write in (True, False):
|
|
group = QGroupBox("Запись" if write else "Чтение")
|
|
group.setSizePolicy(QSizePolicy.Maximum, QSizePolicy.Fixed)
|
|
form = QGridLayout(group)
|
|
address = QLineEdit(default_address)
|
|
value = QLineEdit("0000" if write else "")
|
|
binary = QLineEdit("0000 0000 0000 0000" if write else "")
|
|
address.setFixedWidth(150)
|
|
value.setFixedWidth(100)
|
|
binary.setFixedWidth(240)
|
|
value.setReadOnly(not write)
|
|
binary.setReadOnly(not write)
|
|
for row, (title, widget) in enumerate((("Адрес", address), ("Байт (hex)", value),
|
|
("Байт (bin)", binary))):
|
|
form.addWidget(QLabel(title), row, 0)
|
|
form.addWidget(widget, row, 1)
|
|
button = QPushButton("Записать" if write else "Прочитать")
|
|
form.addWidget(button, 3, 1, Qt.AlignLeft)
|
|
form.setColumnStretch(2, 1)
|
|
if write:
|
|
self.byte_write_address, self.byte_write_value, self.byte_write_binary = address, value, binary
|
|
self.byte_write_button = button
|
|
value.textEdited.connect(lambda text: self.sync_byte_value(text, 16))
|
|
binary.textEdited.connect(lambda text: self.sync_byte_value(text, 2))
|
|
button.clicked.connect(lambda: self.start_byte(True))
|
|
else:
|
|
self.byte_read_address, self.byte_read_value, self.byte_read_binary = address, value, binary
|
|
self.byte_read_button = button
|
|
button.clicked.connect(lambda: self.start_byte(False))
|
|
panels.addWidget(group)
|
|
panels.addStretch()
|
|
layout.addLayout(panels)
|
|
self.status = QLabel(initial_status)
|
|
self.status.setWordWrap(True)
|
|
layout.addWidget(self.status)
|
|
self.raw_widget = RawProtocolWidget(protocol, device_address=device_address)
|
|
self.bus_widget = SerialTab(SerialBus(protocol), protocol, self.raw_widget,
|
|
device_address=device_address, macros_file=macros_file)
|
|
layout.addWidget(self.bus_widget)
|
|
layout.addWidget(self.raw_widget)
|
|
layout.addStretch()
|
|
|
|
def set_status(self, text):
|
|
"""Display external connection/operation status without emitting it again."""
|
|
self.status.setText(text)
|
|
|
|
def _report(self, text):
|
|
self.set_status(text)
|
|
self.status_changed.emit(text)
|
|
|
|
@staticmethod
|
|
def _binary(value):
|
|
bits = f"{value:016b}"
|
|
return " ".join(bits[i:i + 4] for i in range(0, 16, 4))
|
|
|
|
def sync_byte_value(self, text, base):
|
|
try:
|
|
value = int(text.replace(" ", ""), base)
|
|
if not 0 <= value <= 0xFFFF:
|
|
return
|
|
except ValueError:
|
|
return
|
|
target = self.byte_write_binary if base == 16 else self.byte_write_value
|
|
target.setText(self._binary(value) if base == 16 else f"{value:04X}")
|
|
|
|
def set_read_value(self, address, value):
|
|
self.byte_read_address.setText(f"${address:X}")
|
|
self.byte_read_value.setText(f"{value:04X}")
|
|
self.byte_read_binary.setText(self._binary(value))
|
|
|
|
def start_byte(self, write):
|
|
if self._word_running:
|
|
self._report("Дождитесь завершения текущего обмена")
|
|
return
|
|
try:
|
|
field = self.byte_write_address if write else self.byte_read_address
|
|
text = field.text().strip()
|
|
address = int(text[1:], 16) if text.startswith("$") else int(
|
|
text, 16 if text.lower().startswith("0x") else 10)
|
|
value = int(self.byte_write_value.text().replace(" ", ""), 16) if write else None
|
|
if not 0 <= address <= 0xFFFFFFFF:
|
|
raise ValueError()
|
|
if write and (not 0 <= value <= 0xFFFF or
|
|
int(self.byte_write_binary.text().replace(" ", ""), 2) != value):
|
|
raise ValueError()
|
|
except ValueError:
|
|
self._report("Проверьте адрес и значение HEX/BIN")
|
|
return
|
|
if self._start_word is not None:
|
|
self._start_word(write, address, value)
|
|
return
|
|
controller = self.device_address() if callable(self.device_address) else self.device_address
|
|
self._word_generation += 1
|
|
generation = self._word_generation
|
|
self._set_word_running(True)
|
|
self._report("Запись и проверка чтением…" if write else "Чтение…")
|
|
self.protocol.request_async(address, value, controller,
|
|
lambda success, result: self._word_reply(generation, address, value, controller, success, result))
|
|
|
|
def _word_reply(self, generation, address, expected, controller, success, result):
|
|
if generation != self._word_generation or not self._word_running:
|
|
return
|
|
if not success:
|
|
self._set_word_running(False)
|
|
self._report(str(result))
|
|
return
|
|
if expected is not None and "value" not in result:
|
|
self.protocol.request_async(address, None, controller,
|
|
lambda ok, reply: self._word_reply(generation, address, expected, controller, ok, reply))
|
|
return
|
|
value = result["value"]
|
|
self.set_read_value(address, value)
|
|
self._set_word_running(False)
|
|
if expected is not None and value != expected:
|
|
self._report(f"Запись 0x{address:08X} не подтверждена: прочитано {value:04X}, ожидалось {expected:04X}")
|
|
else:
|
|
self._report("Запись проверена чтением" if expected is not None else "Чтение завершено")
|
|
|
|
def _set_word_running(self, running):
|
|
self._word_running = running
|
|
self.byte_write_button.setEnabled(not running)
|
|
self.byte_read_button.setEnabled(not running)
|
|
|
|
def stop(self):
|
|
"""Cancel this page's scheduled work before closing its connection."""
|
|
self.bus_widget.stop_macro()
|
|
if self._word_running:
|
|
self._word_generation += 1
|
|
self._set_word_running(False)
|
|
self.protocol.cancel_async()
|
|
self._report("Обмен прерван; переданные команды могли быть выполнены")
|
|
|
|
def closeEvent(self, event):
|
|
self.stop()
|
|
super().closeEvent(event)
|
|
|
|
|
|
def build_byte_page(protocol, **options):
|
|
"""Create a standalone page; add the returned widget to a tab or layout."""
|
|
return ByteWorkPanel(protocol, **options)
|