diff --git a/.gitignore b/.gitignore index 843dba4..0d1ca56 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,10 @@ qrc_*.cpp moc_*.cpp *.qmlcache *.qmldir.cache + +# Build / packaging artifacts +build/ +dist/ +*.spec +*.icns +*.ico \ No newline at end of file diff --git a/ISC/HomePage.qml b/ISC/HomePage.qml index de31cfe..541a5a1 100644 --- a/ISC/HomePage.qml +++ b/ISC/HomePage.qml @@ -22,7 +22,7 @@ Rectangle { // ===== View State ===== property int selectedTabIndex: 0 - property int selectedSideActionIndex: 0 + property int selectedSideActionIndex: -1 // nothing active on startup // ===== Main Two-Column Layout ===== RowLayout { @@ -57,7 +57,13 @@ Rectangle { Layout.fillWidth: true Layout.preferredHeight: Math.max(Theme.sideButtonMinHeight, sideRail.computedButtonHeight) hoverEnabled: true - onClicked: root.selectedSideActionIndex = index + onClicked: { + root.selectedSideActionIndex = index + root.selectedTabIndex = 0 // always jump to Status tab + if (index === 0) deviceController.detectWafer() + else if (index === 1) deviceController.readMemoryAsync() + else if (index === 2) deviceController.openCsvFile() + } background: Rectangle { color: { @@ -138,9 +144,21 @@ Rectangle { source: parent.tabName === "Settings" ? "Tabs/SettingsTab.qml" : "" } + Loader { + anchors.fill: parent + active: parent.tabName === "Status" + source: parent.tabName === "Status" ? "Tabs/StatusTab.qml" : "" + } + + Loader { + anchors.fill: parent + active: parent.tabName === "Data" + source: parent.tabName === "Data" ? "Tabs/DataTab.qml" : "" + } + Label { anchors.centerIn: parent - visible: parent.tabName !== "Settings" + visible: parent.tabName !== "Settings" && parent.tabName !== "Status" && parent.tabName !== "Data" text: parent.tabName + " content" color: Theme.bodyColor font.pixelSize: 20 diff --git a/ISC/Tabs/DataTab.qml b/ISC/Tabs/DataTab.qml new file mode 100644 index 0000000..7a2b4b2 --- /dev/null +++ b/ISC/Tabs/DataTab.qml @@ -0,0 +1,153 @@ +import QtQuick +import QtQuick.Controls +import QtQuick.Layouts +import ISC + +// ===== Data Tab ===== +// Displays parsed temperature data in a scrollable table. +// Column 0 = Row index, remaining columns = Sensor1, Sensor2, ... +Item { + id: root + anchors.fill: parent + + readonly property int rowCount: deviceController.dataRowCount + + // Re-layout columns whenever the model is reset + Connections { + target: deviceController + function onParsedDataReady(result) { + if (result && result.success) { + dataTable.forceLayout() + } + } + } + + // ===== Empty State ===== + Column { + anchors.centerIn: parent + spacing: 16 + visible: root.rowCount === 0 + + Label { + text: "No Data" + font.pixelSize: 24 + font.bold: true + color: Theme.headingColor + anchors.horizontalCenter: parent.horizontalCenter + } + + Label { + text: "Read and parse wafer data to see temperature readings here." + font.pixelSize: 14 + color: Theme.bodyColor + anchors.horizontalCenter: parent.horizontalCenter + wrapMode: Text.WordWrap + } + } + + // ===== Data Table ===== + ColumnLayout { + anchors.fill: parent + anchors.margins: Theme.panelPadding + spacing: Theme.rightPaneGap + visible: root.rowCount > 0 + + // --- Header row --- + RowLayout { + Layout.fillWidth: true + spacing: Theme.rightPaneGap + + Label { + text: "Temperature Data" + font.pixelSize: 18 + font.bold: true + color: Theme.headingColor + } + + Item { Layout.fillWidth: true } + + Label { + text: deviceController.dataRowCount + " rows × " + + deviceController.dataColCount + " sensors" + font.pixelSize: 13 + color: Theme.bodyColor + } + } + + // --- Table container --- + Rectangle { + Layout.fillWidth: true + Layout.fillHeight: true + color: Theme.panelBackground + border.color: Theme.cardBorder + border.width: Theme.borderThin + radius: Theme.radiusSm + clip: true + + ColumnLayout { + anchors.fill: parent + anchors.margins: 1 + spacing: 0 + + // Column header strip + HorizontalHeaderView { + id: headerView + Layout.fillWidth: true + syncView: dataTable + clip: true + + delegate: Rectangle { + implicitHeight: 28 + color: Theme.cardBackground + border.color: Theme.cardBorder + border.width: 1 + + Text { + anchors.centerIn: parent + // 'display' is the Qt.DisplayRole value from headerData() + text: display !== undefined ? display : "" + color: Theme.headingColor + font.pixelSize: 11 + font.bold: true + elide: Text.ElideRight + } + } + } + + // Data rows + TableView { + id: dataTable + Layout.fillWidth: true + Layout.fillHeight: true + model: deviceController.dataModel + clip: true + reuseItems: true + + columnWidthProvider: function(col) { + if (col === 0) return 52 // row index column + const sensorCols = Math.max(1, dataTable.columns - 1) + const available = dataTable.width - 52 + return Math.max(52, Math.min(90, available / sensorCols)) + } + + rowHeightProvider: function() { return 24 } + + delegate: Rectangle { + color: row % 2 === 0 ? Theme.panelBackground : Theme.cardBackground + + Text { + anchors.centerIn: parent + text: display !== undefined ? display : "" + color: Theme.bodyColor + font.pixelSize: 11 + elide: Text.ElideRight + } + } + + ScrollBar.horizontal: ScrollBar { policy: ScrollBar.AsNeeded } + ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded } + } + } + } + } +} diff --git a/ISC/Tabs/StatusTab.qml b/ISC/Tabs/StatusTab.qml new file mode 100644 index 0000000..e39777b --- /dev/null +++ b/ISC/Tabs/StatusTab.qml @@ -0,0 +1,276 @@ +import QtQuick +import QtQuick.Layouts +import QtQuick.Controls +import ISC + +// ===== Status Tab ===== +// Blank on startup. Fills in after Detect Wafer fires from the left rail. +ColumnLayout { + id: root + anchors.fill: parent + anchors.margins: Theme.panelPadding + spacing: Theme.rightPaneGap + + property alias waferFamilyCode: waferInfoFamily.text + property alias waferSerialNumber: waferSerial.text + property alias waferSensorCount: waferSensors.text + property alias waferRuntime: waferRuntime.text + property alias waferCycles: waferCycles.text + property bool waferDetected: false + + // Data summary after parse + property int dataRows: 0 + property int dataCols: 0 + property string csvPath: "" + property bool dataParsed: false + + // ===== Empty state — nothing shown until detect fires ===== + Item { + Layout.fillWidth: true + Layout.fillHeight: true + visible: !root.waferDetected && !deviceController.operationInProgress + } + + // ===== Results area (shown once detect/operation runs) ===== + ColumnLayout { + Layout.fillWidth: true + Layout.fillHeight: true + visible: root.waferDetected || deviceController.operationInProgress + spacing: Theme.rightPaneGap + + // --- Connection Status --- + Rectangle { + Layout.fillWidth: true + Layout.preferredHeight: 86 + color: { + if (deviceController.connectionStatus === "Connected") return Theme.statusSuccessColor + "22" + if (deviceController.connectionStatus === "Disconnected") return Theme.statusErrorColor + "22" + return Theme.cardBackground + } + border.color: { + if (deviceController.connectionStatus === "Connected") return Theme.statusSuccessColor + if (deviceController.connectionStatus === "Disconnected") return Theme.statusErrorColor + return Theme.cardBorder + } + radius: Theme.radiusSm + + ColumnLayout { + anchors.fill: parent + anchors.margins: Theme.panelPadding + spacing: 4 + + Text { + text: deviceController.connectionStatus + color: { + if (deviceController.connectionStatus === "Connected") return Theme.statusSuccessColor + if (deviceController.connectionStatus === "Disconnected") return Theme.statusErrorColor + return Theme.bodyColor + } + font.bold: true + font.pixelSize: 16 + Layout.fillWidth: true + } + + Text { + text: "Port: " + (deviceController.selectedPort || "None selected") + color: Theme.bodyColor + font.pixelSize: 14 + Layout.fillWidth: true + } + + Text { + text: "Save dir: " + deviceController.saveDataDir + color: Theme.bodyColor + font.pixelSize: 12 + opacity: 0.7 + elide: Text.ElideMiddle + Layout.fillWidth: true + } + } + } + + // --- Wafer Info --- + GroupBox { + title: "Wafer Information" + visible: root.waferDetected + Layout.fillWidth: true + + ColumnLayout { + anchors.fill: parent + anchors.margins: Theme.panelPadding + spacing: 8 + + RowLayout { + Layout.fillWidth: true + spacing: Theme.rightPaneGap + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Family Code"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { id: waferInfoFamily; text: "—"; color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Serial Number"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { id: waferSerial; text: "—"; color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Sensor Count"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { id: waferSensors; text: "—"; color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + } + + RowLayout { + Layout.fillWidth: true + spacing: Theme.rightPaneGap + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Runtime"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { id: waferRuntime; text: "—"; color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Cycles"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { id: waferCycles; text: "—"; color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + } + } + } + + // --- Data Summary (shown after parse) --- + GroupBox { + title: "Data Summary" + visible: root.dataParsed + Layout.fillWidth: true + + RowLayout { + anchors.fill: parent + anchors.margins: Theme.panelPadding + spacing: Theme.rightPaneGap + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Rows"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { text: String(root.dataRows); color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "Sensors"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { text: String(root.dataCols); color: Theme.bodyColor; font.pixelSize: 16; font.bold: true } + } + + ColumnLayout { + Layout.fillWidth: true + spacing: 4 + Text { text: "CSV"; color: Theme.subheadingColor; font.pixelSize: 12 } + Text { text: root.csvPath; color: Theme.bodyColor; font.pixelSize: 12; elide: Text.ElideMiddle } + } + } + } + + // --- Activity Log --- + GroupBox { + title: "Activity Log" + Layout.fillWidth: true + Layout.fillHeight: true + + ScrollView { + anchors.fill: parent + clip: true + + TextArea { + id: activityLog + width: parent.width + text: "" + readOnly: true + font.family: "monospace" + font.pixelSize: 12 + wrapMode: TextArea.WordWrap + leftPadding: 4 + rightPadding: 4 + color: Theme.bodyColor + } + } + + Connections { + target: deviceController + function onActivityLogUpdated(newLog) { + activityLog.text = newLog + } + } + } + } + + // --- Signal Handlers --- + Connections { + target: deviceController + function onDetectResult(result) { + if (result && result.familyCode) { + root.waferDetected = true + waferInfoFamily.text = result.familyCode + waferSerial.text = result.serialNumber + waferSensors.text = String(result.sensorCount) + waferRuntime.text = result.runtime + "s" + waferCycles.text = String(result.cycleCount) + } + // On failure, keep the last displayed wafer info and do not change waferDetected. + } + + function onStatusRestored() { + // Show restored wafer info if available + var info = deviceController.lastWaferInfo + if (info && info.length > 0) { + root.waferDetected = true + waferInfoFamily.text = info[0] || "—" + waferSerial.text = info[1] || "—" + waferSensors.text = String(info[2] || 0) + waferRuntime.text = (info[3] || 0) + "s" + waferCycles.text = String(info[4] || 0) + } + // Show data summary if data was previously parsed + if (deviceController.dataRowCount > 0) { + root.dataParsed = true + root.dataRows = deviceController.dataRowCount + root.dataCols = deviceController.dataColCount + root.csvPath = "" // CSV path not persisted as full path, just show it was parsed + } + } + } + + Connections { + target: deviceController + function onReadResult(result) { + root.dataParsed = false + root.dataRows = 0 + root.dataCols = 0 + root.csvPath = "" + } + } + + Connections { + target: deviceController + function onParsedDataReady(result) { + if (result && result.success) { + root.dataParsed = true + root.dataRows = result.rows + root.dataCols = result.cols + root.csvPath = result.csv_path || "" + } else { + root.dataParsed = false + } + } + } +} diff --git a/ISC/Theme.qml b/ISC/Theme.qml index 6dd5b0a..d9b9312 100644 --- a/ISC/Theme.qml +++ b/ISC/Theme.qml @@ -50,7 +50,9 @@ QtObject { // ── 5. Text ────────────────────────────────────────────────────────────── readonly property color headingColor: toneText readonly property color bodyColor: toneMute + readonly property color subheadingColor: toneMute readonly property color panelTitleText: toneMute + readonly property color disabledText: toneMute // ── 6. Controls ────────────────────────────────────────────────────────── // Fields diff --git a/backend/data_model.py b/backend/data_model.py new file mode 100644 index 0000000..6e3cf07 --- /dev/null +++ b/backend/data_model.py @@ -0,0 +1,84 @@ +"""QAbstractTableModel for displaying parsed wafer temperature data in QML.""" + +from __future__ import annotations + +import logging +from typing import Any + +from PySide6.QtCore import QAbstractTableModel, Qt + +log = logging.getLogger(__name__) + +# Column roles for the temperature data table +ROW_ROLE = Qt.ItemDataRole.UserRole + 1 +COL_ROLE = Qt.ItemDataRole.UserRole + 2 + + +class TemperatureTableModel(QAbstractTableModel): + """Table model for parsed wafer temperature data. + + Exposes a 2D list of temperature strings to QML TableView. + Column 0 = row index, remaining columns = Sensor1, Sensor2, ... + """ + + def __init__(self, parent: Any = None) -> None: + super().__init__(parent) + self._data: list[list[str]] = [] # 2D array of temperature strings + self._col_count: int = 0 # Number of sensor columns (excludes row index) + + def reset(self) -> None: + """Clear all data.""" + self.beginResetModel() + self._data = [] + self._col_count = 0 + self.endResetModel() + + def load_data(self, data: list[list[str]], col_count: int) -> None: + """Load parsed temperature data into the model. + + Args: + data: 2D list of temperature strings (e.g. [["25.30", "24.80", ...], ...]). + col_count: Number of sensor columns (may differ from data[0] length). + """ + self.beginResetModel() + self._data = data + self._col_count = col_count + self.endResetModel() + log.info("Loaded %d rows × %d cols into model", len(data), col_count) + + # ---- QAbstractTableModel interface ---- + + def rowCount(self, parent: Any = None) -> int: + return len(self._data) + + def columnCount(self, parent: Any = None) -> int: + # Column 0 = row index, columns 1..N = sensor values + return self._col_count + 1 + + def data(self, index: Any, role: int = ...) -> Any: + if not index.isValid(): + return None + + row = index.row() + col = index.column() + + if role == Qt.ItemDataRole.DisplayRole: + if col == 0: + return str(row + 1) # 1-based row index + sensor_col = col - 1 + if row < len(self._data) and sensor_col < len(self._data[row]): + return self._data[row][sensor_col] + return "0" + + return None + + def headerData( + self, section: int, orientation: int, role: int = ... + ) -> Any: + if role != Qt.ItemDataRole.DisplayRole: + return None + if orientation == Qt.Orientation.Horizontal: + if section == 0: + return "Row" + return f"Sensor{section}" + return str(section + 1) diff --git a/backend/device_controller.py b/backend/device_controller.py new file mode 100644 index 0000000..cd014b8 --- /dev/null +++ b/backend/device_controller.py @@ -0,0 +1,521 @@ +"""QML-exposed controller for wafer device communication. + +Bridges DeviceService (serialcomm) to QML via @Slot methods and signals. +All hardware operations run synchronously on the main thread (matching +C# Form1.cs per-operation open/close pattern). For long operations +(erase ~15s, read up to 120s) the caller should show a loading state. +""" + +from __future__ import annotations + +import logging +import threading +from datetime import datetime +from typing import Any, Optional + +from PySide6.QtCore import QObject, Property, Qt, Signal, Slot + +from backend.data_model import TemperatureTableModel +from backend.graph_view import GraphView +from backend.local_settings import LocalSettings +from serialcomm.data_parser import ( + convert_to_temperatures, + parse_binary_data, + remove_trailing_zeros, + save_to_csv, +) +from serialcomm.device_service import DeviceService +from serialcomm.serial_port import WaferInfo + +log = logging.getLogger(__name__) + + +class DeviceController(QObject): + """Controls serial communication with the temperature-sensing wafer. + + Exposed to QML as ``deviceController`` context property. + """ + + # ---- public signals ---- + portsUpdated = Signal(list) + detectResult = Signal(object) # WaferInfo dict or None + readResult = Signal(object) # {"success": bool, "bytes": int} or {"error": str} + eraseResult = Signal(object) # {"success": bool} + debugResult = Signal(object) # {"success": bool, "sensor_bytes": int, "debug_bytes": int} + parsedDataReady = Signal(object) # {"success": bool, "rows": int, "cols": int, "data": ..., "csv_path": str} + statusRestored = Signal() # Emitted when status is restored from previous session + logMessage = Signal(str) + activityLogUpdated = Signal(str) + + # ---- private signals (marshal worker-thread results back to main thread) ---- + _detectFinished = Signal(object, object) # (port_or_None, WaferInfo_or_None) + _readFinished = Signal(str, str, object) # (port, family_code, bytes_or_None) + + def __init__(self, settings: LocalSettings, data_dir: str, parent: QObject | None = None) -> None: + super().__init__(parent) + self._settings = settings + self._data_dir = data_dir + self._service = DeviceService(settings) + + # Initialize status from persisted settings (with defaults for new installations) + self._connection_status = getattr(settings, 'connection_status', "Disconnected") + self._operation_in_progress = False + self._activity_log: list[str] = getattr(settings, 'activity_log', []) + self._raw_bytes: Optional[bytes] = None + # Default save dir lives next to the settings file (~/Documents/isc_data/csv) + from pathlib import Path + self._save_data_dir: str = getattr(settings, 'save_data_dir', str(Path.home() / "Documents" / "isc_data" / "csv")) + self._last_wafer_info: dict[str, Any] = getattr(settings, 'last_wafer_info', {}) + self._last_csv_path: str = getattr(settings, 'last_csv_path', "") + self._selected_port: str = getattr(settings, 'selected_port', "") + self._data_row_count: int = getattr(settings, 'data_row_count', 0) + self._data_col_count: int = getattr(settings, 'data_col_count', 0) + self._data_model = TemperatureTableModel(self) + self._graph_view = GraphView(self) + + # If we have persisted activity log, emit it to QML + if self._activity_log: + self.activityLogUpdated.emit("\n".join(self._activity_log)) + + # Log status restoration and emit signal for UI updates + if self._connection_status != "Disconnected" or self._selected_port or self._last_wafer_info: + self._append_log("Restored previous session status") + if self._selected_port: + self._append_log(f"Last selected port: {self._selected_port}") + if self._last_wafer_info: + info = self._last_wafer_info + self._append_log(f"Last wafer: {info.get('serialNumber', 'Unknown')} ({info.get('familyCode', 'Unknown')})") + self.statusRestored.emit() + + # Marshal worker-thread results onto the Qt main thread. + self._detectFinished.connect(self._handle_detect_finished, Qt.QueuedConnection) + self._readFinished.connect(self._handle_read_finished, Qt.QueuedConnection) + + # ---- Properties ---- + @Property(list, notify=portsUpdated) + def availablePorts(self) -> list[str]: + """Currently available serial port names.""" + return self._service.enumerate_ports() + + @Property(str, notify=portsUpdated) + def connectionStatus(self) -> str: + """Current connection status string for display.""" + return self._connection_status + + @Property(bool, notify=portsUpdated) + def operationInProgress(self) -> bool: + """Whether a hardware operation is currently running.""" + return self._operation_in_progress + + @Property(str, notify=activityLogUpdated) + def saveDataDir(self) -> str: + """Directory where parsed CSV data files are saved.""" + return self._save_data_dir + + @Slot(str) + def setSaveDataDir(self, path: str) -> None: + """Set the directory for saving parsed CSV data files.""" + self._save_data_dir = path + self._append_log(f"Save data dir set to: {path}") + self._save_status() + + @Property(str, notify=portsUpdated) + def selectedPort(self) -> str: + """Currently selected serial port shared across the UI.""" + return self._selected_port + + @Slot(str) + def setSelectedPort(self, port: str) -> None: + """Set the active serial port from the port selector.""" + self._selected_port = port + self._save_status() + + @Property(int, notify=parsedDataReady) + def dataRowCount(self) -> int: + """Number of rows in the parsed temperature dataset.""" + return self._data_row_count + + @Property(int, notify=parsedDataReady) + def dataColCount(self) -> int: + """Number of sensor columns in the parsed temperature dataset.""" + return self._data_col_count + + @Property(list, notify=detectResult) + def lastWaferInfo(self) -> list: + """Last detected wafer info as a flat list for QML bindings. + + Returns [familyCode, serialNumber, sensorCount, mfgDateHex, runtime, cycleCount]. + """ + info = self._last_wafer_info + return [ + info.get("familyCode", ""), + info.get("serialNumber", ""), + info.get("sensorCount", 0), + info.get("mfgDateHex", ""), + info.get("runtime", 0), + info.get("cycleCount", 0), + ] + + @Property(object, notify=detectResult) + def dataModel(self) -> TemperatureTableModel: + """QAbstractTableModel for the parsed temperature data table.""" + return self._data_model + + @Property(object, notify=debugResult) + def graphView(self) -> GraphView: + """GraphView for rendering sensor temperature line charts.""" + return self._graph_view + + @Slot(result=object) + def getChartData(self) -> dict[str, Any]: + """Extract chart-ready data from the current model. + + Returns a dict with sensor names and per-sensor value lists + suitable for GraphView.updateChart(). + """ + if not self._data_model or self._data_model.rowCount() == 0: + return {"success": False} + + num_sensors = self._data_model.columnCount() - 1 # Exclude row index + sensor_names = [f"Sensor{i+1}" for i in range(num_sensors)] + + # Extract per-sensor values from the model + series_data = [] + for col in range(1, self._data_model.columnCount()): + sensor_values = [] + for row in range(self._data_model.rowCount()): + idx = self._data_model.index(row, col) + val = self._data_model.data(idx) + sensor_values.append(val if val else "0") + series_data.append(sensor_values) + + return { + "success": True, + "sensor_names": sensor_names, + "series": series_data, + } + + def _append_log(self, message: str) -> None: + """Append a message to the activity log and emit updated log.""" + ts = datetime.now().strftime("%H:%M:%S") + self._activity_log.append(f"[{ts}] {message}") + # Keep only last 200 lines + if len(self._activity_log) > 200: + self._activity_log = self._activity_log[-200:] + self.activityLogUpdated.emit("\n".join(self._activity_log)) + + def _set_operation_progress(self, in_progress: bool) -> None: + """Set operation progress state and notify QML. + + portsUpdated is the shared notify signal for availablePorts, + connectionStatus, and operationInProgress — emitting it forces + QML to re-evaluate all three bindings. + """ + self._operation_in_progress = in_progress + self.portsUpdated.emit(self._service.enumerate_ports()) + + # ---- Public Slots ---- + + @Slot() + def refreshPorts(self) -> None: + """Scan for available serial ports and emit updated list.""" + ports = self._service.enumerate_ports() + self._connection_status = "Disconnected" + self.portsUpdated.emit(ports) + self._append_log(f"Scanned {len(ports)} serial port(s)") + self._save_status() + + @Slot() + def detectWafer(self) -> None: + """Scan all serial ports for a wafer (runs in background thread). + + Mirrors C# chkConnection(): tries every available port with s0. + Stores the found port in _selected_port for subsequent read/erase. + Emits detectResult with WaferInfo dict on success, or None. + """ + if self._operation_in_progress: + self._append_log("Already busy — ignoring detect request") + return + + self._set_operation_progress(True) + self._connection_status = "Detecting..." + self.portsUpdated.emit(self._service.enumerate_ports()) + self._append_log("Scanning all ports for wafer ...") + + # Spawn a daemon worker — UI stays responsive during the scan. + threading.Thread(target=self._detect_worker, daemon=True).start() + + def _detect_worker(self) -> None: + """Background-thread body for detectWafer.""" + try: + port, info = self._service.detect_all_ports() + except Exception as exc: + log.exception("Detect worker crashed: %s", exc) + port, info = None, None + self._detectFinished.emit(port, info) + + @Slot(object, object) + def _handle_detect_finished( + self, port: Optional[str], info: Optional[WaferInfo] + ) -> None: + """Main-thread completion handler for detectWafer.""" + if info is not None: + self._selected_port = port or "" + self._connection_status = "Connected" + self._append_log( + f"Detected: {info.serial_number} (family={info.family_code}, " + f"port={port}, runtime={info.runtime}s, cycles={info.cycle_count})" + ) + wafer_dict = self._wafer_info_to_dict(info) + self._last_wafer_info = wafer_dict + self.detectResult.emit(wafer_dict) + self._save_status() + else: + self._selected_port = "" + self._connection_status = "Disconnected" + self._append_log("No wafer detected on any port") + self._last_wafer_info = {} + self.detectResult.emit(None) + self._set_operation_progress(False) + + @Slot() + def readMemoryAsync(self) -> None: + """Read wafer memory in a background thread. + + Uses the family code and port from the last detect. + Emits readResult on completion, then auto-chains to parseAndSaveData. + """ + if self._operation_in_progress: + self._append_log("Already busy — ignoring read request") + return + + if not self._selected_port: + self._append_log("No wafer detected — run Detect Wafer first") + self.readResult.emit({"error": "No port — detect wafer first"}) + return + + port = self._selected_port + family_code = self._last_wafer_info.get("familyCode", "") + + self._set_operation_progress(True) + self._connection_status = "Reading..." + self.portsUpdated.emit(self._service.enumerate_ports()) + self._append_log( + f"Reading memory on {port} (family={family_code or 'auto'}) ..." + ) + + threading.Thread( + target=self._read_worker, + args=(port, family_code), + daemon=True, + ).start() + + def _read_worker(self, port: str, family_code: str) -> None: + """Background-thread body for readMemoryAsync.""" + try: + data = self._service.read_wafer_data(port, family_code) + except Exception as exc: + log.exception("Read worker crashed: %s", exc) + data = None + self._readFinished.emit(port, family_code, data) + + @Slot(str, str, object) + def _handle_read_finished( + self, port: str, family_code: str, data: Optional[bytes] + ) -> None: + """Main-thread completion handler for readMemoryAsync.""" + if data is not None: + self._connection_status = "Connected" + self._append_log(f"Read {len(data)} bytes") + self._raw_bytes = data + self.readResult.emit({"success": True, "bytes": len(data)}) + self._set_operation_progress(False) + # Auto-chain: read → parse → save (matches C# behavior). + # Parse runs on main thread — it's CPU-bound but bounded (~1s). + self.parseAndSaveData(family_code, port) + self._save_status() + return + self._connection_status = "Disconnected" + self._append_log("Read returned no data") + self._raw_bytes = None + self.readResult.emit({"error": "Read returned no data"}) + self._set_operation_progress(False) + self._save_status() + + @Slot(str) + def eraseMemory(self, port: str) -> None: + """Send p1 erase command.""" + self._set_operation_progress(True) + self._connection_status = "Erasing..." + self.portsUpdated.emit(self._service.enumerate_ports()) + self._append_log(f"Erase command sent to {port} (wafer takes ~15s) ...") + + ok = self._service.erase_wafer(port) + if ok: + self._connection_status = "Connected" + self._append_log("Erase command accepted — wait ~15 seconds") + else: + self._connection_status = "Disconnected" + self._append_log("Erase command failed") + self.eraseResult.emit({"success": ok}) + self._set_operation_progress(False) + + @Slot(str) + def readDebug(self, port: str) -> None: + """Read both D1 (sensor) and F1 (debug/cold-junction) data. + + Emits debugResult with counts on success. + """ + self._set_operation_progress(True) + self._connection_status = "Reading debug..." + self.portsUpdated.emit(self._service.enumerate_ports()) + self._append_log(f"Reading debug data on {port} ...") + + # Step 1: D1 sensor data + sensor_data = self._service.read_wafer_data(port, family_code="") + if sensor_data is None: + self._connection_status = "Disconnected" + self._append_log("D1 read failed — aborting debug read") + self.debugResult.emit({"error": "D1 read failed"}) + self._set_operation_progress(False) + return + + self._append_log(f"D1 read: {len(sensor_data)} bytes") + + # Step 2: F1 debug data + debug_data = self._service.read_wafer_data(port, family_code="", cmd="F1") + if debug_data is None: + self._connection_status = "Disconnected" + self._append_log("F1 read failed") + self.debugResult.emit({"error": "F1 read failed"}) + self._set_operation_progress(False) + return + + self._connection_status = "Connected" + self._append_log( + f"Debug read complete: {len(sensor_data)} sensor bytes, " + f"{len(debug_data)} debug bytes" + ) + self.debugResult.emit({ + "success": True, + "sensor_bytes": len(sensor_data), + "debug_bytes": len(debug_data), + }) + self._set_operation_progress(False) + + @Slot(str, str) + def parseAndSaveData(self, family_code: str = "", port: str = "") -> None: + """Parse raw bytes into temperatures and save to CSV. + + Uses the bytes from the most recent readMemoryAsync call. + Emits parsedDataReady with the parsed result. + """ + if not self._raw_bytes: + self._append_log("No raw bytes to parse (read first)") + self.parsedDataReady.emit({ + "success": False, + "error": "No raw bytes available", + }) + return + + if not self._save_data_dir: + self._append_log("No save data directory set") + self.parsedDataReady.emit({ + "success": False, + "error": "No save data directory set", + }) + return + + fc = family_code or ( + self._last_wafer_info.get("familyCode", "") + if self._last_wafer_info + else "" + ) + serial = ( + self._last_wafer_info.get("serialNumber", "UNKNOWN") + if self._last_wafer_info + else "UNKNOWN" + ) + + self._append_log(f"Parsing {len(self._raw_bytes)} bytes (family={fc or 'auto'}) ...") + + # Step 1: Parse binary → hex strings + hex_data = parse_binary_data(self._raw_bytes, fc) + if hex_data is None: + self._append_log("Binary parse failed") + self.parsedDataReady.emit({ + "success": False, + "error": "Binary parse failed", + }) + return + + rows = len(hex_data) + cols = len(hex_data[0]) if hex_data else 0 + self._append_log(f"Parsed: {rows} rows × {cols} sensors") + + # Step 2: Convert hex → temperatures + temp_data = convert_to_temperatures(hex_data, fc) + + # Step 3: Remove trailing zero rows + remove_trailing_zeros(temp_data) + self._append_log(f"After trim: {len(temp_data)} rows") + + # Step 4: Save to CSV + csv_path = save_to_csv(temp_data, fc, serial, self._save_data_dir) + if csv_path is None: + self._append_log("CSV save failed") + self.parsedDataReady.emit({ + "success": False, + "error": "CSV save failed", + }) + return + + self._append_log(f"Saved CSV: {csv_path}") + + # Load data into the QAbstractTableModel + self._data_model.load_data(temp_data, cols) + self._data_row_count = len(temp_data) + self._data_col_count = cols + + # Emit parsed data to QML + self.parsedDataReady.emit({ + "success": True, + "rows": len(temp_data), + "cols": cols, + "data": temp_data, + "csv_path": csv_path, + "familyCode": fc, + "serialNumber": serial, + }) + self._save_status() + + # ---- Status Persistence ---- + + def _save_status(self) -> None: + """Persist current operational status to settings.""" + # Update settings with current status values + self._settings.connection_status = self._connection_status + self._settings.selected_port = self._selected_port + self._settings.last_wafer_info = self._last_wafer_info.copy() + self._settings.save_data_dir = self._save_data_dir + self._settings.activity_log = self._activity_log.copy() + self._settings.data_row_count = self._data_row_count + self._settings.data_col_count = self._data_col_count + self._settings.last_csv_path = self._last_csv_path + + # Save to disk + LocalSettings.save_settings(LocalSettings._settings_path(self._data_dir), self._settings) + + # ---- Helpers ---- + + @staticmethod + def _wafer_info_to_dict(info: WaferInfo) -> dict[str, Any]: + """Convert WaferInfo dataclass to JSON-serialisable dict for QML.""" + return { + "familyCode": info.family_code, + "serialNumber": info.serial_number, + "sensorCount": info.sensor_count, + "mfgDateHex": info.mfg_date_hex, + "runtime": info.runtime, + "cycleCount": info.cycle_count, + } diff --git a/backend/graph_view.py b/backend/graph_view.py new file mode 100644 index 0000000..eac5bfb --- /dev/null +++ b/backend/graph_view.py @@ -0,0 +1,169 @@ +"""pyqtgraph PlotWidget wrapper for embedding in QML. + +Exposes a QWidget with a pyqtgraph PlotWidget that can be displayed +via QML's `import QtWidgets` or `QtWidgets.QWidget` integration. +""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +from PySide6.QtCore import QObject, Property, Signal, Slot +from PySide6.QtWidgets import QWidget + +log = logging.getLogger(__name__) + +# Import pyqtgraph after Qt is initialized +import pyqtgraph as pg +from pyqtgraph import PlotWidget + + +class GraphView(QObject): + """QML-exposed controller for a pyqtgraph line chart. + + Accepts sensor temperature data (list of lists) and renders + each sensor as a separate line series. + """ + + # ---- signals ---- + dataReady = Signal(object) # {"success": bool, "sensory_names": list, "series": list} + + def __init__(self, parent: Optional[QObject] = None) -> None: + super().__init__(parent) + self._plot_widget: Optional[PlotWidget] = None + self._plot_window: Optional[QWidget] = None + self._series: list[Any] = [] + self._sensor_names: list[str] = [] + + @Property(object, notify=dataReady) + def plotWidget(self) -> Any: + """Return the QWidget hosting the pyqtgraph PlotWidget for QML embedding.""" + return self._plot_window + + @Slot() + def createPlotWidget(self, parent_widget: Optional[QWidget] = None) -> None: + """Create and return a QWidget containing a pyqtgraph PlotWidget. + + Args: + parent_widget: Optional parent widget. + """ + pg.setConfigOption("background", "default") + pg.setConfigOption("foreground", "default") + + self._plot_window = QWidget(parent=parent_widget) + self._plot_widget = PlotWidget() + self._plot_widget.setBackground("default") + + from PySide6.QtWidgets import QVBoxLayout + layout = QVBoxLayout(self._plot_window) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + layout.addWidget(self._plot_widget) + + # Set axis labels + self._plot_widget.setLabel("left", "Temperature", units="°C") + self._plot_widget.setLabel("bottom", "Measurement Interval") + self._plot_widget.setTitle("Sensor Temperature Over Time") + + @Slot(str, str) + def updateChart(self, sensor_names_str: str, series_data_str: str) -> None: + """Update the chart with sensor data. + + Args: + sensor_names_str: Comma-separated sensor names (e.g. "Sensor1,Sensor2"). + series_data_str: JSON-like string of nested lists for each sensor's values. + """ + import json + + if not self._plot_widget: + log.warning("PlotWidget not created yet") + return + + try: + sensor_names = [s.strip() for s in sensor_names_str.split(",") if s.strip()] + series_data = json.loads(series_data_str) + except (json.JSONDecodeError, AttributeError) as exc: + log.error("Failed to parse chart data: %s", exc) + return + + # Clear existing series + self._plot_widget.clear() + self._series = [] + + if not series_data: + return + + # Determine Y-axis range from all data + all_values = [] + for sensor_values in series_data: + for v in sensor_values: + try: + all_values.append(float(v)) + except (ValueError, TypeError): + pass + + if all_values: + y_min = min(all_values) + y_max = max(all_values) + y_range = y_max - y_min + buffer = max(y_range * 0.1, 1.0) # At least 1 degree buffer + self._plot_widget.setYRange(y_min - buffer, y_max + buffer) + else: + self._plot_widget.setYRange(-50, 150) + + # X-axis: measurement intervals (0-based index) + num_points = len(series_data[0]) if series_data else 0 + x_axis = list(range(num_points)) + + # Define a set of distinct colors for series + colors = [ + (255, 87, 87), # Red + (66, 165, 245), # Blue + (102, 187, 106), # Green + (255, 167, 38), # Orange + (171, 71, 188), # Purple + (0, 188, 212), # Cyan + (255, 112, 67), # Deep Orange + (121, 85, 72), # Brown + (92, 107, 192), # Indigo + (48, 125, 117), # Teal + ] + + # Add each sensor as a line series + for i, sensor_name in enumerate(sensor_names): + if i >= len(series_data): + break + + sensor_values = series_data[i] + y_values = [] + for v in sensor_values: + try: + y_values.append(float(v)) + except (ValueError, TypeError): + y_values.append(0.0) + + color = colors[i % len(colors)] + pen = pg.mkPen(color=color, width=1) + + curve = self._plot_widget.plot(x_axis, y_values, name=sensor_name, pen=pen) + self._series.append(curve) + + self._sensor_names = sensor_names + + @Slot() + def resetChart(self) -> None: + """Clear the chart.""" + if self._plot_widget: + self._plot_widget.clear() + self._series = [] + self._sensor_names = [] + + @Slot() + def destroyPlotWidget(self) -> None: + """Destroy the plot widget.""" + if self._plot_window: + self._plot_window.deleteLater() + self._plot_window = None + self._plot_widget = None + self._series = [] diff --git a/backend/local_settings.py b/backend/local_settings.py index d89be31..640d2b1 100644 --- a/backend/local_settings.py +++ b/backend/local_settings.py @@ -7,6 +7,7 @@ from pathlib import Path class LocalSettings: # ===== Defaults ===== def __init__(self): + # Configuration settings self.chamber_id = "" self.reverse_z_wafer = False self.master = {} # Dict[str, str] @@ -18,6 +19,16 @@ class LocalSettings: self.wafer_detect_timeout = 5000 self.split_threshold = 40.0 + # Status persistence (operational state) + self.connection_status = "Disconnected" + self.selected_port = "" + self.last_wafer_info = {} # Dict[str, Any] + self.save_data_dir = "" + self.activity_log = [] # List[str] + self.data_row_count = 0 + self.data_col_count = 0 + self.last_csv_path = "" + # ===== File Path Helpers ===== @classmethod def _settings_path(cls, directory: str) -> Path: diff --git a/main.py b/main.py index 7a54800..18f4954 100644 --- a/main.py +++ b/main.py @@ -5,6 +5,8 @@ from PySide6.QtQml import QQmlApplicationEngine from PySide6.QtQuickControls2 import QQuickStyle from PySide6.QtWidgets import QApplication +from backend.device_controller import DeviceController +from backend.local_settings import LocalSettings from backend.local_settings_model import LocalSettingsModel from backend.file_browser import FileBrowser @@ -26,6 +28,12 @@ if __name__ == "__main__": engine.rootContext().setContextProperty("settingsModel", settings_model) engine.rootContext().setContextProperty("file_browser", select_file_dialog_model) + # ===== Device Controller (serial comm) ===== + data_dir = str(settings_model._data_dir) + raw_settings = LocalSettings.read_settings(data_dir) + device_controller = DeviceController(raw_settings, data_dir) + engine.rootContext().setContextProperty("deviceController", device_controller) + # ===== QML Startup ===== engine.addImportPath(Path(__file__).parent) engine.loadFromModule("ISC", "Main") diff --git a/pyproject.toml b/pyproject.toml index 27a3e10..33774a0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,16 @@ # ===== Project Metadata ===== [project] -name = "PySide QtQuick Project" +name = "pygui" +version = "0.1.0" + +[project.optional-dependencies] +dev = ["pytest"] # ===== PySide Build Inputs ===== [tool.pyside6-project] -files = ["ISC/HomePage.qml", "ISC/Main.qml", "ISC/Tabs/SelectFileDialog.qml", "ISC/Tabs/SettingsTab.qml", "ISC/Tabs/qmldir", "ISC/Theme.qml", "ISC/qmldir", "backend/file_browser.py", "backend/local_settings.py", "backend/local_settings_model.py", "main.py"] +files = ["ISC/HomePage.qml", "ISC/Main.qml", "ISC/Tabs/DataTab.qml", "ISC/Tabs/GraphTab.qml", "ISC/Tabs/SelectFileDialog.qml", "ISC/Tabs/SettingsTab.qml", "ISC/Tabs/StatusTab.qml", "ISC/Tabs/qmldir", "ISC/Theme.qml", "ISC/qmldir", "backend/data_model.py", "backend/device_controller.py", "backend/graph_view.py", "backend/file_browser.py", "backend/local_settings.py", "backend/local_settings_model.py", "main.py", "serialcomm/__init__.py", "serialcomm/data_parser.py", "serialcomm/serial_port.py", "serialcomm/device_service.py"] + +[dependency-groups] +dev = [ + "pytest>=9.0.3", +] diff --git a/requirements.txt b/requirements.txt index 4e03671..fff2e63 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ PySide6>=6.6.0 pyqtgraph>=0.13.0 numpy>=1.24.0 +pyserial>=3.5 diff --git a/serialcomm/__init__.py b/serialcomm/__init__.py new file mode 100644 index 0000000..3c4c280 --- /dev/null +++ b/serialcomm/__init__.py @@ -0,0 +1,6 @@ +"""Serial port communication layer for the temperature-sensing wafer.""" + +from serialcomm.device_service import DeviceService +from serialcomm.serial_port import SerialPort, WaferInfo + +__all__ = ["DeviceService", "SerialPort", "WaferInfo"] diff --git a/serialcomm/data_parser.py b/serialcomm/data_parser.py new file mode 100644 index 0000000..4ad9bca --- /dev/null +++ b/serialcomm/data_parser.py @@ -0,0 +1,292 @@ +"""Binary data parser and temperature converter for wafer data. + +Mirrors the C# Form1.cs binary parsing pipeline: + 1. Read raw bytes → strip per-block overhead → 1D hex array + 2. Chunk 1D array into rows × sensors → List[List[str]] (hex values) + 3. Convert hex values → float temperatures (family-dependent) + 4. Remove trailing zero rows + 5. Save to CSV with Sensor1, Sensor2, ... headers +""" + +import logging +import os +from typing import Optional + +log = logging.getLogger(__name__) + +# Max DUTs (sensors) per row before overhead bytes are stripped +# P wafer: 244 valid readings per 256-block (12 overhead bytes) +# X wafer: 80 valid readings per 256-block (14 overhead bytes) +MAXDUT_P = 244 +MAXDUT_X = 80 + + +def csv_column_count(family_code: str) -> int: + """Return the number of columns to display for a family code.""" + mapping = { + "A": 48, + "E": 48, + "P": 48, + "B": 29, + "C": 29, + "D": 29, + "F": 22, + "X": 80, + } + return mapping.get(family_code, 0) + + +def _hex_to_binary(hex_str: str) -> list[int]: + """Convert a 4-char hex string to a 16-bit binary list (MSB first).""" + value = int(hex_str, 16) + return [(value >> (15 - i)) & 1 for i in range(16)] + + +def _twos_complement_excluding_msb(bits: list[int]) -> list[int]: + """Invert bits 1-15, add 1 (2's complement excluding MSB).""" + bits = list(bits) # copy + for i in range(1, len(bits)): + bits[i] = 1 - bits[i] + carry = 1 + for i in range(len(bits) - 1, 0, -1): + s = bits[i] + carry + bits[i] = s % 2 + carry = s // 2 + return bits + + +def _binary_subsequence_to_int(bits: list[int], start: int, end: int) -> int: + """Convert bits[start:end+1] to integer.""" + result = 0 + for i in range(start, end + 1): + result = (result << 1) | bits[i] + return result + + +def _binary_fraction_to_double(bits: list[int], start: int) -> float: + """Convert bits[start:] to a fractional value (0 < frac < 1).""" + result = 0.0 + divisor = 2.0 + for i in range(start, len(bits)): + result += bits[i] / divisor + divisor *= 2.0 + return result + + +def _convert_standard(binary_bits: list[int]) -> float: + """Convert 16-bit binary using standard formula (B/C/D/F families). + + Bit layout: + bit 0 : sign + bits 1-11 : integer part (11 bits, 2^10 .. 2^0) + bits 12-13 : fractional part (2 bits, 2^-1, 2^-2) + bits 14-15 : unused + """ + value = 0.0 + for i in range(1, 14): + if binary_bits[i]: + value += 2.0 ** (11 - i) + if binary_bits[0]: + value = -value + return value + + +def _convert_aep(binary_bits: list[int]) -> float: + """Convert 16-bit binary using AEP formula. + + Bit layout: + bit 0 : sign + bits 1-8 : integer part (8 bits) + bits 9-15 : fractional part (7 bits) + """ + bits = binary_bits + if bits[0] == 1: + bits = _twos_complement_excluding_msb(bits) + integer_part = _binary_subsequence_to_int(bits, 1, 8) + fractional_part = _binary_fraction_to_double(bits, 9) + result = integer_part + fractional_part + if binary_bits[0] == 1: + result = -result + return result + + +def _convert_hex_to_temp(hex_str: str, family_code: str) -> float: + """Convert a singi hale 4-char hex string to a float temperature.""" + bits = _hex_to_binary(hex_str) + if family_code in ("A", "E", "P"): + result = _convert_aep(bits) + elif family_code in ("B", "C", "D"): + result = _convert_standard(bits) + if result < -2000: + result = 0.0 + elif family_code == "X": + # X wafer uses AEP-style conversion + result = _convert_aep(bits) + else: + # Unknown family — try standard + result = _convert_standard(bits) + return round(result, 2) + + +def parse_binary_data(data_bytes: bytes, family_code: str) -> Optional[list[list[str]]]: + """Parse raw wafer bytes into a 2D array of hex strings. + + Strips per-block overhead bytes (12 for P, 14 for X) and chunks + the remaining readings into rows. + + Args: + data_bytes: Raw binary data from the wafer. + family_code: Wafer family code ("P", "X", "A", "B", "C", "D", "F"). + + Returns: + List of rows, each row is a list of 4-char hex strings. + Returns None on failure. + """ + try: + if family_code == "X": + return _parse_x_binary(data_bytes) + else: + return _parse_p_binary(data_bytes) + except Exception as exc: + log.error("Binary parse failed: %s", exc) + return None + + +def _parse_p_binary(data_bytes: bytes) -> list[list[str]]: + """Parse P-family (and A/B/C/D/E/F) binary data. + + Each block of 256 readings has 244 valid + 12 overhead. + Valid readings are chunked into rows of MAXDUT_P (244). + """ + readings: list[str] = [] + + # Read 2 bytes at a time (UInt16 little-endian) + # Each 256-word block has MAXDUT_P valid readings (first N words), rest are overhead + num_words = len(data_bytes) // 2 + for i in range(num_words): + value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little") + if i % 256 < MAXDUT_P: + readings.append(f"{value:04X}") + + # Chunk into rows of MAXDUT_P + result: list[list[str]] = [] + idx = 0 + while idx + MAXDUT_P <= len(readings): + result.append(readings[idx : idx + MAXDUT_P]) + idx += MAXDUT_P + + log.info("Parsed P-family: %d rows × %d cols", len(result), MAXDUT_P) + return result + + +def _parse_x_binary(data_bytes: bytes) -> list[list[str]]: + """Parse X-family binary data. + + Each block of 256 readings has 80 valid + 14 overhead. + Valid readings are chunked into rows of MAXDUT_X (80). + """ + readings: list[str] = [] + + num_words = len(data_bytes) // 2 + for i in range(num_words): + value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little") + if i % 256 < MAXDUT_X: + readings.append(f"{value:04X}") + + # Chunk into rows of MAXDUT_X + result: list[list[str]] = [] + idx = 0 + while idx + MAXDUT_X <= len(readings): + result.append(readings[idx : idx + MAXDUT_X]) + idx += MAXDUT_X + + log.info("Parsed X-family: %d rows × %d cols", len(result), MAXDUT_X) + return result + + +def convert_to_temperatures( + hex_data: list[list[str]], family_code: str +) -> list[list[str]]: + """Convert hex string values to temperature strings. + + Args: + hex_data: 2D array of 4-char hex strings from parse_binary_data. + family_code: Wafer family code. + + Returns: + Same structure with temperature strings (e.g. "25.37"). + """ + temp_value: list[list[str]] = [] + for row in hex_data: + temp_row: list[str] = [] + for hex_val in row: + temp = _convert_hex_to_temp(hex_val, family_code) + temp_row.append(str(temp)) + temp_value.append(temp_row) + return temp_value + + +def remove_trailing_zeros(data: list[list[str]]) -> None: + """Remove rows of all-zero values from the end of data (in-place). + + A value is considered zero if its float representation is < 0.01. + """ + while data: + last_row = data[-1] + is_all_zero = True + for val in last_row: + try: + fval = float(val) + except ValueError: + fval = 99.9 + if fval >= 0.01: + is_all_zero = False + break + if is_all_zero: + data.pop() + else: + break + + +def save_to_csv( + data: list[list[str]], + family_code: str, + serial_number: str, + output_dir: str, +) -> Optional[str]: + """Save parsed temperature data to a CSV file. + + Args: + data: 2D array of temperature strings. + family_code: Wafer family code. + serial_number: Wafer serial number (e.g. "P00001"). + output_dir: Directory to save the CSV file. + + Returns: + Full file path on success, None on failure. + """ + try: + os.makedirs(output_dir, exist_ok=True) + # Build filename: P00001-20260505_133045.csv + from datetime import datetime + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"{serial_number}-{timestamp}.csv" + filepath = os.path.join(output_dir, filename) + + num_cols = csv_column_count(family_code) + with open(filepath, "w", encoding="utf-8") as f: + # Header row: Sensor1,Sensor2,... + headers = [f"Sensor{j + 1}" for j in range(num_cols)] + f.write(",".join(headers) + "\n") + # Data rows + for row in data: + # Pad or trim to match header count + padded = row[:num_cols] + ["0"] * max(0, num_cols - len(row)) + f.write(",".join(padded) + "\n") + + log.info("Saved %d rows × %d cols to %s", len(data), num_cols, filepath) + return filepath + except Exception as exc: + log.error("CSV save failed: %s", exc) + return None diff --git a/serialcomm/device_service.py b/serialcomm/device_service.py new file mode 100644 index 0000000..b4a5a8b --- /dev/null +++ b/serialcomm/device_service.py @@ -0,0 +1,152 @@ +"""High-level wafer device communication service. + +Coordinates port enumeration, detect, read, and erase operations +using settings from LocalSettingsModel. Designed to be called from +QML via @Slot methods on a QObject controller. +""" + +import logging +from pathlib import Path +from typing import Optional + +import serial.tools.list_ports + +from backend.local_settings import LocalSettings +from serialcomm.serial_port import SerialPort, WaferInfo + +log = logging.getLogger(__name__) + + +class DeviceService: + """High-level wafer device communication.""" + + # Expected hex string lengths for known wafer families + # P wafer: 393216 hex chars (196608 bytes) + # X wafer: 1310720 hex chars (655360 bytes) + EXPECTED_HEX_LENGTHS = { + "P": 393216, + "X": 1310720, + } + + def __init__(self, settings: LocalSettings) -> None: + self._settings = settings + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def enumerate_ports(self) -> list[str]: + """Return list of available serial port names.""" + return [p.device for p in serial.tools.list_ports.comports()] + + def detect_wafer(self, port: str) -> Optional[WaferInfo]: + """Try to detect a wafer on the given port. + + Opens the port, sends s0 command, parses response. + Returns WaferInfo if detected, None otherwise. + """ + try: + sp = SerialPort(port) + info = sp.detect(self._settings.wafer_detect_timeout) + return info + except Exception as exc: + log.error("Detect failed on %s: %s", port, exc) + return None + + SCAN_PER_PORT_TIMEOUT_MS = 1500 + """Per-port timeout while scanning. Wafer responds fast or not at all.""" + + def detect_all_ports(self) -> tuple[Optional[str], Optional[WaferInfo]]: + """Scan every available serial port until one responds to s0. + + Mirrors C# chkConnection(): enumerates all ports, tries each one, + returns the first that responds with a valid 1024-byte payload. + + Uses a short per-port timeout (SCAN_PER_PORT_TIMEOUT_MS) so a + machine with many serial devices (Bluetooth, debug consoles, etc.) + doesn't take 30+ seconds to scan. The configured wafer_detect_timeout + is reserved for confirmed-port operations. + + Returns: + (port_name, WaferInfo) on success, (None, None) if nothing found. + """ + ports = self.enumerate_ports() + log.info("Scanning %d port(s) for wafer ...", len(ports)) + for port in ports: + log.info(" trying %s ...", port) + try: + sp = SerialPort(port) + info = sp.detect(self.SCAN_PER_PORT_TIMEOUT_MS) + except Exception as exc: + log.warning(" %s: %s", port, exc) + info = None + if info is not None: + log.info(" found wafer on %s (family=%s)", port, info.family_code) + return port, info + log.info(" no wafer detected on any port") + return None, None + + def read_wafer_data( + self, + port: str, + family_code: str = "", + cmd: str = "D1", + timeout_ms: int | None = None, + retries: int | None = None, + ) -> Optional[bytes]: + """Full read workflow: open → send command → retry → close. + + Args: + port: Serial port name (e.g. "/dev/ttyUSB0"). + family_code: Wafer family ("P", "X", etc.) for size validation. + cmd: Command string ("D1" for sensor data, "F1" for debug data). + timeout_ms: Read timeout in ms (uses settings default if None). + retries: Max retries on bad read (uses settings default if None). + + Returns: + Raw bytes from the wafer, or None on failure. + """ + if timeout_ms is None: + timeout_ms = self._settings.wafer_read_timeout + if retries is None: + retries = self._settings.wafer_read_retries + + # Determine expected hex string length. + # get_wafer_data_size returns BYTES; hex string is 2× that. + if family_code: + expected_hex_len = self._settings.get_wafer_data_size(family_code) * 2 + else: + expected_hex_len = 0 # no validation + + try: + sp = SerialPort(port) + data = sp.read_memory( + cmd=cmd, + expected_hex_len=expected_hex_len, + timeout_ms=timeout_ms, + retries=retries, + ) + if data is not None: + log.info("Read %d bytes from wafer on %s", len(data), port) + else: + log.warning("Read returned no data from %s", port) + return data + except Exception as exc: + log.error("Read failed on %s: %s", port, exc) + return None + + def erase_wafer(self, port: str) -> bool: + """Send p1 erase command. Wafer takes ~15s to erase. + + Returns True if command was sent successfully. + """ + try: + sp = SerialPort(port) + return sp.erase_memory() + except Exception as exc: + log.error("Erase failed on %s: %s", port, exc) + return False + + def get_expected_hex_length(self, family_code: str) -> int: + """Return expected hex string length for a wafer family.""" + return self._settings.get_wafer_data_size(family_code) diff --git a/serialcomm/serial_port.py b/serialcomm/serial_port.py new file mode 100644 index 0000000..d49d9ed --- /dev/null +++ b/serialcomm/serial_port.py @@ -0,0 +1,203 @@ +"""Low-level serial port communication with the wafer device. + +Mirrors the C# Form1.cs wafer protocol: + - Detect : s0 + 510×'F' → 1024 hex chars → WaferInfo + - ReadMemory: D1 + 510×'F' → N hex chars (retry on bad length) + - Erase : p1 + 510×'F' → no response (~15s blocking) + +All commands are 512-char strings: 2-char command + 510×'F' padding. +""" + +import logging +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Iterator, Optional + +import serial as pyserial + +log = logging.getLogger(__name__) + + +@dataclass +class WaferInfo: + """Parsed wafer identification from the s0 detect response.""" + + family_code: str # "P", "X", "A", "B", "C", "D", "E", "F" + serial_number: str # e.g. "P00001" + sensor_count: int + mfg_date_hex: str # raw 8 hex chars + sensor_assigned_hex: str # raw 2 hex chars + locked_hex: str # raw 2 hex chars + runtime: int # seconds + cycle_count: int + + +class SerialPort: + """Low-level serial transport for the temperature-sensing wafer. + + The serial port is opened per-operation (matching C# behaviour) + rather than kept open persistently. + """ + + BAUDRATE = 888888 + COMMAND_PAD = "F" * 510 # pad to 512 chars total + + def __init__(self, port_name: str) -> None: + self._port_name = port_name + + # ------------------------------------------------------------------ + # Context manager for scoped port access + # ------------------------------------------------------------------ + + @contextmanager + def _open(self, timeout: float | None = None) -> Iterator[pyserial.Serial]: + """Open the serial port, yield it, then close on exit.""" + port = pyserial.Serial( + self._port_name, + self.BAUDRATE, + parity=pyserial.PARITY_NONE, + bytesize=8, + stopbits=pyserial.STOPBITS_ONE, + xonxoff=False, + rtscts=False, + dsrdtr=False, + ) + if timeout is not None: + port.timeout = timeout + try: + yield port + finally: + port.close() + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def detect(self, timeout_ms: int = 5000) -> Optional[WaferInfo]: + """Try s0 command. Returns WaferInfo if response is 1024 hex chars.""" + with self._open(timeout_ms / 1000) as port: + port.write(("s0" + self.COMMAND_PAD).encode()) + response = port.readline().decode().strip() + if len(response) == 1024: + return self._parse_wafer_info(response) + return None + + def read_memory( + self, + cmd: str = "D1", + expected_hex_len: int = 393216, + timeout_ms: int = 120000, + retries: int = 8, + ) -> Optional[bytes]: + """Send read command, retry on bad length or odd-length string. + + The wafer sends hex-encoded bytes. A dropped byte corrupts the + alignment, so the entire buffer must be retried. + + Returns raw bytes on success, None on failure. + """ + with self._open(timeout_ms / 1000) as port: + port.write((cmd + self.COMMAND_PAD).encode()) + + for attempt in range(retries, 0, -1): + data_string = port.readline().decode(errors="ignore").strip() + + # Trim trailing odd nibble — wafer sometimes drops 1 char + # at the tail; one stray nibble is harmless, the rest is good. + if len(data_string) % 2 != 0: + data_string = data_string[:-1] + + # Defensively cap at expected length if the wafer over-sends. + if expected_hex_len > 0 and len(data_string) > expected_hex_len: + data_string = data_string[:expected_hex_len] + + # Accept if we got at least 99% of expected (matches C# tolerance). + threshold = int(expected_hex_len * 0.99) if expected_hex_len > 0 else 1 + if len(data_string) >= threshold: + break + + log.warning( + "Short hex string (len=%d, expected~%d), retries left=%d", + len(data_string), + expected_hex_len, + attempt - 1, + ) + + if attempt > 1: + port.reset_input_buffer() + # Re-issue the command — reset_input_buffer alone leaves + # the wafer waiting for nothing. + port.write((cmd + self.COMMAND_PAD).encode()) + continue + + log.error("Retries exhausted after bad read") + return None + + try: + return bytes.fromhex(data_string) + except ValueError as exc: + log.error("Failed to decode hex: %s", exc) + return None + + def erase_memory(self, timeout_ms: int = 5000) -> bool: + """Send p1 erase command. Wafer takes ~15s to erase. + + Returns True if command was sent successfully. + """ + with self._open(timeout_ms / 1000) as port: + try: + port.write(("p1" + self.COMMAND_PAD).encode()) + return True + except Exception as exc: + log.error("Erase command failed: %s", exc) + return False + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + @staticmethod + def _hex_to_ascii(hex_str: str) -> str: + """Convert hex string (e.g. '50') to ASCII char ('P').""" + hex_str = hex_str.replace(" ", "") + result = [] + for i in range(0, len(hex_str), 2): + hex_char = hex_str[i : i + 2] + if len(hex_char) == 2: + result.append(chr(int(hex_char, 16))) + return "".join(result) + + def _parse_wafer_info(self, hex_response: str) -> WaferInfo: + """Parse the 1024-char hex response into WaferInfo. + + Layout (from C# checkPort): + bytes 0-1 : FamilyCode (2 hex chars → ASCII) + bytes 2-5 : Serial number (3 bytes hex → decimal, 5 digits) + bytes 6-7 : Sensor count (1 byte hex) + bytes 8-15 : Mfg date (4 bytes hex) + bytes 16-17 : Sensor assigned (1 byte hex) + bytes 18-19 : Locked (1 byte hex) + bytes 1016-1019 : Runtime (2 bytes hex → decimal seconds) + bytes 1020-1023 : Cycle time (2 bytes hex → decimal cycles) + """ + raw = bytes.fromhex(hex_response) + + family_code = self._hex_to_ascii(hex_response[0:2]) + serial_num = f"{family_code}{int(hex_response[2:6], 16):05d}" + sensor_count = int(hex_response[6:8], 16) + mfg_date_hex = hex_response[8:16] + sensor_assigned_hex = hex_response[16:18] + locked_hex = hex_response[18:20] + runtime = int(hex_response[1016:1020], 16) + cycle_count = int(hex_response[1020:1024], 16) + + return WaferInfo( + family_code=family_code, + serial_number=serial_num, + sensor_count=sensor_count, + mfg_date_hex=mfg_date_hex, + sensor_assigned_hex=sensor_assigned_hex, + locked_hex=locked_hex, + runtime=runtime, + cycle_count=cycle_count, + ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_data_parser.py b/tests/test_data_parser.py new file mode 100644 index 0000000..5dc64cc --- /dev/null +++ b/tests/test_data_parser.py @@ -0,0 +1,299 @@ +"""Tests for serialcomm/data_parser.py binary parsing pipeline.""" + +import pytest +from serialcomm.data_parser import ( + csv_column_count, + parse_binary_data, + convert_to_temperatures, + remove_trailing_zeros, + save_to_csv, + _convert_hex_to_temp, + MAXDUT_P, + MAXDUT_X, +) + + +# ── csv_column_count ────────────────────────────────────────────────────────── + + +class TestCsvColumnCount: + def test_a_family(self): + assert csv_column_count("A") == 48 + + def test_e_family(self): + assert csv_column_count("E") == 48 + + def test_p_family(self): + assert csv_column_count("P") == 48 + + def test_b_family(self): + assert csv_column_count("B") == 29 + + def test_c_family(self): + assert csv_column_count("C") == 29 + + def test_d_family(self): + assert csv_column_count("D") == 29 + + def test_f_family(self): + assert csv_column_count("F") == 22 + + def test_x_family(self): + assert csv_column_count("X") == 80 + + def test_unknown_family_returns_zero(self): + assert csv_column_count("Z") == 0 + + def test_empty_string_returns_zero(self): + assert csv_column_count("") == 0 + + +# ── Temperature conversion ──────────────────────────────────────────────────── + + +class TestConvertHexToTemp: + """Spot-check known hex → temperature values. + + Reference values derived from the C# ConvertBinaryArrayToDecimal logic. + """ + + def test_zero_hex_gives_zero_standard(self): + # 0x0000 → all bits 0 → temperature 0.0 + result = _convert_hex_to_temp("0000", "B") + assert result == 0.0 + + def test_zero_hex_gives_zero_aep(self): + result = _convert_hex_to_temp("0000", "A") + assert result == 0.0 + + def test_standard_positive_small(self): + # 0x0050 = 0000 0000 0101 0000 → bits 1-11 = 000000000101 = 4+1 = 5, fraction bits = 00 → 5.0°C + result = _convert_hex_to_temp("0050", "B") + assert result == pytest.approx(5.0, abs=0.1) + + def test_standard_negative(self): + # Sign bit set → negative temperature + result = _convert_hex_to_temp("8050", "B") + assert result < 0 + + def test_aep_positive(self): + # 0x6400 = 0110 0100 0000 0000 + # sign=0, int bits 1-8 = 11001000 = ... + # Just check it's in a reasonable range + result = _convert_hex_to_temp("6400", "A") + assert -300 < result < 300 + + def test_x_family_uses_aep(self): + # X uses same AEP formula as A + result_x = _convert_hex_to_temp("0100", "X") + result_a = _convert_hex_to_temp("0100", "A") + assert result_x == result_a + + def test_unknown_family_falls_back_to_standard(self): + # Should not raise; falls back to standard formula + result = _convert_hex_to_temp("0050", "Z") + assert isinstance(result, float) + + +# ── Binary parsing ──────────────────────────────────────────────────────────── + + +def _make_p_block(num_blocks: int = 1, value: int = 0x0100) -> bytes: + """Build synthetic P-family binary data. + + Each block is 256 words (512 bytes). Words 0-243 hold `value`, + words 244-255 are zero (overhead). + """ + data = bytearray() + for _ in range(num_blocks): + for i in range(256): + word = value if i < MAXDUT_P else 0 + data += word.to_bytes(2, byteorder="little") + return bytes(data) + + +def _make_x_block(num_blocks: int = 1, value: int = 0x0100) -> bytes: + """Build synthetic X-family binary data. + + Each block is 256 words (512 bytes). Words 0-79 hold `value`, + words 80-255 are zero (overhead). + """ + data = bytearray() + for _ in range(num_blocks): + for i in range(256): + word = value if i < MAXDUT_X else 0 + data += word.to_bytes(2, byteorder="little") + return bytes(data) + + +class TestParseBinaryData: + def test_p_family_single_block_returns_one_row(self): + data = _make_p_block(1) + result = parse_binary_data(data, "P") + assert result is not None + assert len(result) == 1 + + def test_p_family_single_block_row_has_244_sensors(self): + data = _make_p_block(1) + result = parse_binary_data(data, "P") + assert result is not None + assert len(result[0]) == MAXDUT_P # 244 + + def test_p_family_two_blocks_returns_two_rows(self): + data = _make_p_block(2) + result = parse_binary_data(data, "P") + assert result is not None + assert len(result) == 2 + + def test_p_family_hex_values_are_4_chars(self): + data = _make_p_block(1) + result = parse_binary_data(data, "P") + assert result is not None + for hex_val in result[0]: + assert len(hex_val) == 4 + + def test_x_family_single_block_returns_one_row(self): + data = _make_x_block(1) + result = parse_binary_data(data, "X") + assert result is not None + assert len(result) == 1 + + def test_x_family_single_block_row_has_80_sensors(self): + data = _make_x_block(1) + result = parse_binary_data(data, "X") + assert result is not None + assert len(result[0]) == MAXDUT_X # 80 + + def test_a_family_uses_p_parser(self): + # A/B/C/D/E/F all route to _parse_p_binary + data = _make_p_block(1) + result = parse_binary_data(data, "A") + assert result is not None + assert len(result[0]) == MAXDUT_P + + def test_empty_bytes_returns_empty_list(self): + result = parse_binary_data(b"", "P") + # No full block → no rows + assert result is not None + assert result == [] + + +# ── convert_to_temperatures ─────────────────────────────────────────────────── + + +class TestConvertToTemperatures: + def test_returns_same_shape(self): + hex_data = [["0000", "0000"], ["0000", "0000"]] + result = convert_to_temperatures(hex_data, "B") + assert len(result) == 2 + assert len(result[0]) == 2 + + def test_zero_hex_gives_zero_string(self): + hex_data = [["0000"]] + result = convert_to_temperatures(hex_data, "B") + assert result[0][0] == "0.0" + + def test_values_are_strings(self): + hex_data = [["0050"]] + result = convert_to_temperatures(hex_data, "B") + assert isinstance(result[0][0], str) + # Must be parseable as float + float(result[0][0]) + + def test_p_family_single_block(self): + data = _make_p_block(1, value=0x0100) + hex_data = parse_binary_data(data, "P") + result = convert_to_temperatures(hex_data, "P") + assert len(result) == 1 + assert all(isinstance(v, str) for v in result[0]) + + +# ── remove_trailing_zeros ───────────────────────────────────────────────────── + + +class TestRemoveTrailingZeros: + def test_removes_all_zero_last_row(self): + data = [["25.0", "24.5"], ["0.0", "0.0"]] + remove_trailing_zeros(data) + assert len(data) == 1 + + def test_preserves_non_zero_rows(self): + data = [["25.0", "24.5"], ["23.0", "22.0"]] + remove_trailing_zeros(data) + assert len(data) == 2 + + def test_removes_multiple_trailing_zero_rows(self): + data = [["25.0"], ["0.0"], ["0.0"]] + remove_trailing_zeros(data) + assert len(data) == 1 + + def test_all_zero_data_becomes_empty(self): + data = [["0.0", "0.0"], ["0.0", "0.0"]] + remove_trailing_zeros(data) + assert data == [] + + def test_empty_list_stays_empty(self): + data = [] + remove_trailing_zeros(data) + assert data == [] + + def test_mixed_keeps_first_nonzero(self): + data = [["25.0"], ["1.0"], ["0.0"]] + remove_trailing_zeros(data) + assert len(data) == 2 + + +# ── save_to_csv ─────────────────────────────────────────────────────────────── + + +class TestSaveToCsv: + def test_creates_file(self, tmp_path): + data = [["25.0", "24.5"], ["23.0", "22.0"]] + result = save_to_csv(data, "P", "P00001", str(tmp_path)) + assert result is not None + from pathlib import Path + + assert Path(result).exists() + + def test_filename_contains_serial(self, tmp_path): + data = [["25.0"]] + result = save_to_csv(data, "P", "P12345", str(tmp_path)) + assert result is not None + assert "P12345" in result + + def test_csv_has_sensor_headers(self, tmp_path): + # Header count is driven by csv_column_count(family), not data width. + # P family → 48 sensors; F family → 22 sensors; X family → 80 sensors. + cases = [("P", 48), ("F", 22), ("X", 80), ("B", 29)] + for family, expected_cols in cases: + data = [["25.0", "24.5"]] + result = save_to_csv(data, family, f"{family}00001", str(tmp_path)) + assert result is not None, f"save_to_csv returned None for {family}" + headers = open(result).readline().strip().split(",") + assert len(headers) == expected_cols, ( + f"{family}: expected {expected_cols} headers, got {len(headers)}" + ) + assert headers[0] == "Sensor1" + assert headers[-1] == f"Sensor{expected_cols}" + + def test_csv_row_count_matches_data(self, tmp_path): + data = [["25.0"], ["24.5"], ["23.0"]] + result = save_to_csv(data, "P", "P00001", str(tmp_path)) + assert result is not None + lines = open(result).readlines() + # 1 header + 3 data rows + assert len(lines) == 4 + + def test_creates_output_dir_if_missing(self, tmp_path): + nested = str(tmp_path / "a" / "b" / "c") + data = [["25.0"]] + result = save_to_csv(data, "P", "P00001", nested) + assert result is not None + from pathlib import Path + + assert Path(result).exists() + + def test_returns_none_on_invalid_path(self): + data = [["25.0"]] + result = save_to_csv(data, "P", "P00001", "/\x00invalid\x00path") + assert result is None diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..3f3c4d0 --- /dev/null +++ b/uv.lock @@ -0,0 +1,86 @@ +version = 1 +revision = 3 +requires-python = ">=3.11" + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + +[[package]] +name = "packaging" +version = "26.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "pygui" +version = "0.1.0" +source = { virtual = "." } + +[package.optional-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [{ name = "pytest", marker = "extra == 'dev'" }] +provides-extras = ["dev"] + +[package.metadata.requires-dev] +dev = [{ name = "pytest", specifier = ">=9.0.3" }] + +[[package]] +name = "pytest" +version = "9.0.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165, upload-time = "2026-04-07T17:16:18.027Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, +]