Код: Выделить всё
import sys
import time
import math
import logging
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Any, List, Tuple
import queue
import numpy as np
from PyQt6 import QtCore, QtGui
from PyQt6.QtWidgets import QMainWindow, QApplication, QMessageBox
from PyQt6.QtSerialPort import QSerialPort, QSerialPortInfo
from PyQt6.QtCore import QIODevice, Qt, QTimer, QThread, pyqtSignal, pyqtSlot
import pyqtgraph as pg
# --- Настройка логирования ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# --- Константы и модели данных ---
class CommandType(Enum):
"""Перечисление команд для контроллера."""
RESET = "Rs"
SAVE_EEPROM = "Sv"
LOAD_EEPROM = "Ld;G"
GET_ALL = "G"
ENABLE_ON = "En=1"
ENABLE_OFF = "En=0"
@dataclass
class PidState:
"""Хранилище состояния PID-регулятора."""
tc: float = 1.0
kp: float = 0.0
ki: float = 0.0
kd: float = 0.0
ni: int = 0
na: int = 0
# --- Класс кольцевого буфера для оптимизации графика ---
class CircularBuffer:
"""
Кольцевой буфер на основе NumPy массива.
Позволяет эффективно добавлять данные и получать текущее состояние без копирования всего массива.
"""
def __init__(self, size: int):
self.size = size
self.buffer = np.zeros(size, dtype=np.float32)
self.write_index = 0
self.count = 0
def append(self, value: float):
"""Добавляет значение в буфер."""
self.buffer[self.write_index] = value
self.write_index += 1
if self.write_index >= self.size:
self.write_index = 0
if self.count < self.size:
self.count += 1
def clear(self):
"""Очищает буфер."""
self.buffer.fill(0)
self.write_index = 0
self.count = 0
def get_data(self) -> Tuple[np.ndarray, np.ndarray]:
"""
Возвращает (x_data, y_data) для отрисовки.
Данные возвращаются в хронологическом порядке (от старых к новым).
"""
if self.count == 0:
return np.array([], dtype=np.float32), np.array([], dtype=np.float32)
# Если буфер еще не заполнен полностью, данные идут от 0 до count-1
if self.count < self.size:
x = np.arange(self.count, dtype=np.float32) * 0.25
y = self.buffer[:self.count]
else:
# Буфер заполнен. Данные начинаются с write_index и идут до конца, затем с начала до write_index-1
x = np.arange(self.size, dtype=np.float32) * 0.25
# Переставляем массив так, чтобы старые данные были в начале
y = np.concatenate((self.buffer[self.write_index:], self.buffer[:self.write_index]))
return x, y
# --- Слой логики (Business Logic) ---
class ControllerManager:
"""Обработка бизнес-логики и состояния контроллера."""
# Карта соответствия ключей протокола атрибутам PidState и их типам
PARAM_TYPES = {
'Tc': ('tc', float),
'Kp': ('kp', float),
'Ki': ('ki', float),
'Kd': ('kd', float),
'Ni': ('ni', int),
'Na': ('na', int)
}
def __init__(self):
self.state = PidState()
self.set_point: int = 0
def update_from_dict(self, data: Dict[str, Any]) -> List[Tuple[str, float]]:
"""
Обновляет состояние контроллера из словаря данных.
Возвращает список обновленных параметров для UI.
"""
updates = []
for key, val in data.items():
# Обработка SetPoint отдельно
if key == 'St':
try:
self.set_point = int(round(float(val)))
except (ValueError, TypeError):
logger.warning(f"Invalid SetPoint value: {val}")
continue
# Обработка PID параметров
if key in self.PARAM_TYPES:
attr_name, target_type = self.PARAM_TYPES[key]
try:
converted_val = target_type(val)
setattr(self.state, attr_name, converted_val)
updates.append((key, converted_val))
except (ValueError, TypeError) as e:
logger.warning(f"Type conversion error for {key}={val}: {e}")
else:
pass
return updates
# --- Слой протокола (Protocol Layer) ---
class ProtocolParser:
"""Изолированная логика парсинга строк устройства."""
@staticmethod
def parse_line(line: str) -> Dict[str, float]:
data = {}
if "=" not in line:
return data
sep = None
for char in [';', ',', '\t']:
if char in line:
sep = char
break
terms = line.split(sep) if sep else [line]
for term in terms:
parts = term.split("=", 1)
if len(parts) == 2:
key = parts[0].strip()
val_str = parts[1].strip()
if not key or not val_str:
continue
try:
data[key] = float(val_str)
except ValueError:
logger.debug(f"Skipping non-numeric value for key '{key}': {val_str}")
return data
@staticmethod
def interpret_status(text: str) -> Optional[str]:
mapping = {
'Start': "Connected",
'Fail': "Communication Fail",
'Open': "Thermocouple Open",
'Error': f"Device Error: {text}"
}
for key, msg in mapping.items():
if key in text:
return msg
return None
# --- Слой связи (Communication Layer) ---
class SerialWorker(QThread):
"""
Поток для работы с последовательным портом.
Использует блокирующее чтение с таймаутом внутри потока для надежности,
эмитируя сигналы в главный поток при получении данных.
Команды отправляются через очередь для обеспечения потокобезопасности.
"""
data_received = pyqtSignal(dict)
error_occurred = pyqtSignal(str)
status_message = pyqtSignal(str)
def __init__(self, port_name: str, baud_rate: int = 250000):
super().__init__()
self.port_name = port_name
self.baud_rate = baud_rate
self.serial = QSerialPort()
self._is_running = True
self._buffer = bytearray()
# Очередь команд для потокобезопасной отправки
self.cmd_queue: queue.Queue[str] = queue.Queue()
def run(self):
"""Основной цикл потока."""
self.serial.setPortName(self.port_name)
self.serial.setBaudRate(self.baud_rate)
self.serial.setDataBits(QSerialPort.DataBits.Data8)
self.serial.setParity(QSerialPort.Parity.NoParity)
self.serial.setStopBits(QSerialPort.StopBits.OneStop)
self.serial.setFlowControl(QSerialPort.FlowControl.NoFlowControl)
if not self.serial.open(QIODevice.OpenModeFlag.ReadWrite):
err_msg = f"Failed to open port: {self.serial.errorString()}"
logger.error(err_msg)
self.error_occurred.emit(err_msg)
return
logger.info(f"Port {self.port_name} opened.")
# Инициализация DTR (Reset контроллера)
self.serial.setDataTerminalReady(False)
QThread.msleep(500)
self.serial.setDataTerminalReady(True)
self.serial.clear()
while self._is_running:
# 1. Отправляем накопленные команды из очереди
try:
while not self.cmd_queue.empty():
cmd = self.cmd_queue.get_nowait()
if self.serial.isOpen():
self.serial.write(cmd.encode('ascii'))
self.serial.flush()
except queue.Empty:
pass
# 2. Читаем данные из порта
if self.serial.waitForReadyRead(10): # Таймаут 10 мс
new_data = bytes(self.serial.readAll())
if new_data:
self._process_data(new_data)
# Небольшая задержка для контроля остановки, если waitForReadyRead возвращает False быстро
QThread.msleep(1)
self.serial.close()
logger.info("Port closed.")
def _process_data(self, new_data: bytes):
"""Обрабатывает входящие байты, накапливая их до конца строки."""
self._buffer.extend(new_data)
# Защита от переполнения буфера (например, 1024 байта)
if len(self._buffer) > 1024:
logger.warning("Buffer overflow detected, clearing buffer.")
self._buffer.clear()
return
# Ищем полные строки (оканчивающиеся \n или \r\n)
while b'\n' in self._buffer or b'\r' in self._buffer:
idx = -1
for i, byte in enumerate(self._buffer):
if byte == 0x0A or byte == 0x0D: # \n or \r
idx = i
break
if idx == -1:
break
line_bytes = self._buffer[:idx]
del self._buffer[:idx+1]
try:
text = line_bytes.decode('ascii').strip()
if not text:
continue
status = ProtocolParser.interpret_status(text)
if status:
# Эмитим сигнал в главный поток (thread-safe)
self.status_message.emit(status)
else:
data = ProtocolParser.parse_line(text)
if data:
self.data_received.emit(data)
except UnicodeDecodeError:
logger.warning("Non-ASCII data received, skipping.")
def send_command(self, cmd: str):
"""
Отправляет команду в порт через очередь.
Этот метод можно вызывать из любого потока (обычно из GUI).
"""
if not self._is_running:
return
if not cmd.endswith(("\n", ";")):
cmd += "\n"
# Кладем команду в очередь, поток run() обработает ее
self.cmd_queue.put(cmd)
def stop(self):
"""
Останавливает поток и закрывает порт.
Устанавливает флаг _is_running в False и ожидает завершения цикла run().
"""
self._is_running = False
# Ожидаем, пока поток не завершится сам (максимум 2 секунды)
# Метод wait() блокирует текущий поток до завершения целевого потока
if not self.wait(2000):
logger.warning("SerialWorker did not finish cleanly within timeout. The thread may still be active.")
# Важно: НЕ вызываем self.terminate() здесь, так как это небезопасно.
# В реальном приложении лучше логировать эту ошибку и проверять состояние позже,
# или убедиться, что цикл run() гарантированно прерывается по флагу _is_running.
# --- Слой представления (Presentation Layer) ---
try:
from MainForm import Ui_MainWindow
except ImportError:
class Ui_MainWindow:
def setupUi(self, MainWindow): pass
class MainWindow(QMainWindow, Ui_MainWindow):
sbdelay = 5000
strStop = "Stop"
strEnable = "Run!"
def __init__(self, controller: ControllerManager):
super().__init__()
self.setupUi(self)
self.controller = controller
# Настройка кольцевых буферов (4 минуты при 4 Гц = 960 точек)
self.max_points = int(10.0 * 60 / 0.25)
self.buf_time = CircularBuffer(self.max_points)
self.buf_msr = CircularBuffer(self.max_points)
self.buf_spt = CircularBuffer(self.max_points)
self.buf_out = CircularBuffer(self.max_points)
self.startme = time.time()
# Маппинг виджетов
self.widget_map = {
'Tc': self.cbTcycle,
'Kp': self.spKp,
'Ki': self.spKi,
'Kd': self.spKd,
'Ni': None,
'Na': None
}
self._init_graph_items()
self.controldis()
# Поток работы с портом
self.worker: Optional[SerialWorker] = None
# Сигналы и слоты
self.btnPort.clicked.connect(self.portclick)
self.cmbPorts.currentTextChanged.connect(self.descr_port)
self.pbReset.clicked.connect(lambda: self.send_cmd(CommandType.RESET.value))
self.pbSaveEEPROM.clicked.connect(lambda: self.send_cmd(CommandType.SAVE_EEPROM.value))
self.pbLoadEEPROM.clicked.connect(lambda: self.send_cmd(CommandType.LOAD_EEPROM.value))
self.pbSetKpid.clicked.connect(self.setsettings)
self.pbRun.clicked.connect(self.outputena)
self.pbSetPoint.clicked.connect(self.setpoint)
self.pbSet160.clicked.connect(lambda: self.send_cmd("St=160"))
self.pbSet180.clicked.connect(lambda: self.send_cmd("St=180"))
def _init_graph_items(self):
"""Инициализация элементов графика."""
self.Graph.clear()
self.graph_item_spt = self.Graph.plot(pen=pg.mkPen(color='green', style=Qt.PenStyle.DashLine), name="SetPoint")
self.graph_item_msr = self.Graph.plot(pen=pg.mkPen(color='red'), name="Measurement")
self.graph_item_out = self.Graph.plot(pen=pg.mkPen(color='yellow'), name="Output")
def cleargraph(self):
"""Очистка буферов и графика."""
self.startme = time.time()
self.buf_time.clear()
self.buf_msr.clear()
self.buf_spt.clear()
self.buf_out.clear()
if hasattr(self, 'graph_item_spt'):
self.graph_item_spt.setData([], [])
self.graph_item_msr.setData([], [])
self.graph_item_out.setData([], [])
def _update_graph_from_buffers(self):
"""Обновляет график из кольцевых буферов."""
# Получаем данные времени (X) и значений (Y)
t_data, _ = self.buf_time.get_data()
if len(t_data) == 0:
return
_, msr_data = self.buf_msr.get_data()
_, spt_data = self.buf_spt.get_data()
_, out_data = self.buf_out.get_data()
# Обновляем виджеты pyqtgraph
self.graph_item_spt.setData(t_data, spt_data)
self.graph_item_msr.setData(t_data, msr_data)
self.graph_item_out.setData(t_data, out_data)
def send_cmd(self, cmd: str):
"""Отправка команды контроллеру."""
if self.worker and self.worker.isRunning():
self.worker.send_command(cmd)
else:
logger.warning("Attempted to send command when port is not connected.")
def setpoint(self):
sp = self.sbSetPoint.value()
self.send_cmd(f"St={sp}")
def outputena(self):
cmd = CommandType.ENABLE_ON if self.pbRun.text() == self.strStop else CommandType.ENABLE_OFF
self.send_cmd(cmd)
@pyqtSlot()
def descr_port(self):
port_name = self.cmbPorts.currentText()
if port_name:
info = QSerialPortInfo(port_name)
if info.isValid():
self.statusbar.showMessage(info.description(), self.sbdelay)
@pyqtSlot()
def show(self):
super().show()
self.getports()
def getports(self):
self.cmbPorts.clear()
ports = QSerialPortInfo.availablePorts()
if ports:
self.cmbPorts.addItems([p.portName() for p in ports])
self.btnPort.setText("Connect")
self.cmbPorts.setDisabled(False)
else:
self.btnPort.setText("Refresh")
self.cmbPorts.setDisabled(True)
@pyqtSlot()
def portclick(self):
if self.worker and self.worker.isRunning():
logger.info("Disconnecting...")
self.worker.stop()
# После вызова stop(), мы уже ждали внутри stop().
# Теперь проверяем, завершился ли он.
if not self.worker.isFinished():
# Если он все еще не закончился (редкий случай), можно дать еще немного времени или просто логировать
logger.warning("Worker thread is still running after stop() call.")
self.worker.deleteLater()
self.worker = None
self.getports()
self.statusbar.showMessage("Disconnected", self.sbdelay)
self.controldis()
else:
selected_port = self.cmbPorts.currentText()
if selected_port:
try:
self.worker = SerialWorker(selected_port)
# Подключаем сигналы ДО запуска потока, чтобы не пропустить первые данные
self.worker.data_received.connect(self.onreceive)
self.worker.error_occurred.connect(lambda e: self.statusbar.showMessage(e, self.sbdelay))
self.worker.status_message.connect(lambda m: self.statusbar.showMessage(m, self.sbdelay))
self.worker.start()
if not self.worker.isRunning():
raise RuntimeError("Failed to start worker thread")
self.btnPort.setText("Disconnect")
self.cmbPorts.setDisabled(True)
self.cleargraph()
except Exception as e:
logger.error(f"Failed to start worker: {e}")
QMessageBox.critical(self, "Error", f"Could not connect to port: {str(e)}")
@pyqtSlot(dict)
def onreceive(self, data: Dict[str, float]):
"""Обработка данных через ControllerManager с обновлением UI."""
updates = self.controller.update_from_dict(data)
# Обновление PID параметров в UI
for key, val in updates:
widget = self.widget_map.get(key)
if widget:
if hasattr(widget, 'setCurrentText'):
widget.setCurrentText(f'{val:.2f}')
elif hasattr(widget, 'setValue'):
try:
if isinstance(val, int):
widget.setValue(int(val))
else:
widget.setValue(float(val))
except (ValueError, TypeError):
pass
# Обработка телеметрии
if 'Ms' in data and 'Ot' in data:
self._handle_telemetry(data['Ms'], data['Ot'])
# Отладочные данные
debug_map = {'Pr': 'edProportional', 'Nt': 'edIntegrator', 'Df': 'edDifferentiator'}
for key, widget_name in debug_map.items():
if key in data:
widget = getattr(self, widget_name, None)
if widget:
try:
widget.setText(f'{data[key]:.2f}')
except (ValueError, TypeError):
pass
# Статус выхода
if 'En' in data:
self.pbRun.setText(self.strStop if data['En'] > 0 else self.strEnable)
def _handle_telemetry(self, ms_val: float, ot_val: float):
"""Обработка текущих значений измерения и выхода."""
try:
self.edMeasurement.setText(f'{ms_val:.2f}')
self.edPidOutput.setText(f'{ot_val:.2f}')
current_time = time.time() - self.startme
# Добавляем данные в кольцевые буферы
self.buf_time.append(current_time)
self.buf_msr.append(ms_val)
self.buf_out.append(ot_val)
self.buf_spt.append(float(self.controller.set_point))
# Обновляем график ТОЛЬКО сейчас, когда пришли новые данные
self._update_graph_from_buffers()
except Exception as e:
logger.error(f"Error handling telemetry: {e}")
def setsettings(self):
cmd_parts = []
params = [
('Tc', 'tc', float, round, 2),
('Kp', 'kp', float, round, 3),
('Ki', 'ki', float, round, 3),
('Kd', 'kd', float, round, 3),
('Ni', 'ni', int, int, 0),
('Na', 'na', int, int, 0)
]
for key, attr, dtype, func, prec in params:
widget = self.widget_map.get(key)
if not widget:
continue
try:
raw_val = widget.value()
# Проверка на отрицательные значения для PID коэффициентов
if dtype == float and key in ['Kp', 'Ki', 'Kd'] and raw_val < 0:
logger.warning(f"Invalid negative value for {key}: {raw_val}")
continue
# Проверка на NaN/Inf
val = func(dtype(raw_val), prec) if dtype != int else int(raw_val)
if not math.isfinite(val):
logger.warning(f"Non-finite value for {key}: {val}")
continue
current_val = getattr(self.controller.state, attr)
if not math.isclose(val, current_val, abs_tol=1e-9):
cmd_parts.append(f'{key}={val}')
except Exception as e:
logger.error(f"Error processing param {key}: {e}")
if cmd_parts:
self.send_cmd(";".join(cmd_parts))
def controldis(self, disabled=True):
controls = [
self.spKp, self.spKi, self.spKd, self.spIntMinLim, self.spIntMaxLim,
self.pbReset, self.pbSetKpid, self.cbTcycle,
self.pbRun, self.sbSetPoint, self.pbSetPoint,
self.pbSaveEEPROM, self.pbLoadEEPROM,
self.pbSet160, self.pbSet180
]
for ctrl in controls:
if hasattr(ctrl, 'setDisabled'):
ctrl.setDisabled(disabled)
def closeEvent(self, event):
logger.info("Closing application...")
# Останавливаем поток порта и ждем его завершения
if self.worker and self.worker.isRunning():
self.worker.stop()
super().closeEvent(event)
def main():
app = QApplication(sys.argv)
try:
app.setStyle("fusion")
except Exception:
pass
controller = ControllerManager()
window = MainWindow(controller)
window.show()
sys.exit(app.exec())
if __name__ == '__main__':
main()
