can udp server and udp clients be used in same python file?
Yes, it is indeed possible to have both a UDP server and UDP client in the same Python file. You can accomplish this using Python's socket module, which provides the necessary functionality for creating both server and client UDP sockets.
Here's a simplified example of how you can achieve this:
import socket # Creating a UDP server server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) server_address = ('localhost', 12345) server.bind(server_address) # Creating a UDP client client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Now you can implement the server and client logic, handling sending and receiving data as needed. # Remember to close the sockets when you're done server.close() client.close()
In this example, both the UDP server and UDP client logic would exist within the same Python file. However, you would typically separate the server and client behavior into different functions or classes for better organization and readability.
import sys from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QTextEdit from PyQt5.QtNetwork import QUdpSocket
class UDPServer(QWidget): def init(self): super().init()
self.socket = QUdpSocket(self) self.socket.bind(12345) # Bind to a specific port self.layout = QVBoxLayout(self) self.text_edit = QTextEdit(self) self.layout.addWidget(self.text_edit) self.start_button = QPushButton("Start Server", self) self.start_button.clicked.connect(self.start_server) self.layout.addWidget(self.start_button) self.socket.readyRead.connect(self.read_pending_datagrams) def start_server(self): self.text_edit.append("Server started on port 12345") def read_pending_datagrams(self): while self.socket.hasPendingDatagrams(): datagram, host, port = self.socket.readDatagram(self.socket.pendingDatagramSize()) message = f"Received from {host.toString()}:{port} - {datagram.data().decode('utf-8')}" self.text_edit.append(message)
class UDPClient(QWidget): def init(self): super().init()
self.socket = QUdpSocket(self) self.layout = QVBoxLayout(self) self.text_edit = QTextEdit(self) self.layout.addWidget(self.text_edit) self.send_button = QPushButton("Send Message", self) self.send_button.clicked.connect(self.send_message) self.layout.addWidget(self.send_button) def send_message(self): message = "Hello, Server!" self.socket.writeDatagram(message.encode('utf-8'), "127.0.0.1", 12345) self.text_edit.append(f"Sent: {message}")
class MainWindow(QWidget): def init(self): super().init()
self.server = UDPServer() self.client = UDPClient() self.layout = QVBoxLayout(self) self.layout.addWidget(self.server) self.layout.addWidget(self.client)
if name == "main": app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec_())
how to add udp server and client concepts in above pyqt5 program
The provided PyQt5 program already includes the concepts of a UDP server and client.
In the given code:
UDPServer class represents the UDP server, which binds to a specific port and listens for incoming datagrams. When a datagram is received, it appends the received message to the text edit.UDPClient class represents the UDP client, which allows the user to send a message to the server.Both the server and client functionalities are incorporated into the MainWindow class, where instances of UDPServer and UDPClient are added as widgets.
It's important to note that your code is already correctly demonstrating the concepts of a UDP server and client using PyQt5 and QUdpSocket. The QUdpSocket class provided by PyQt5 allows handling both the server and client functionalities for UDP communication. When you run the program and interact with the widgets, you can observe the behavior of a UDP server and client based on the defined logic.
If there are specific modifications or additional functionalities you'd like to implement, please feel free to let me know!
import struct
import threading
import time
import sys
import os
import backend
from PyQt5 import QtCore, QtWidgets, QtGui
from PyQt5.QtWidgets import QApplication, QGridLayout, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton,
QTableWidget, QTableWidgetItem, QStackedWidget, QLabel, QSpacerItem, QSizePolicy
from PyQt5.QtGui import QPixmap, QRegion, QIcon
from PyQt5.QtCore import Qt, QSize, QObject, pyqtSignal, QThread
from PyQt5.QtWidgets import QFileDialog
from datetime import datetime
import win32event
import win32api
import tempfile
import winerror
import platform
from collections import OrderedDict
import serial, serial.tools.list_ports as list_ports
import subprocess
import msvcrt
from sensirion_gas_index_algorithm.voc_algorithm import VocAlgorithm
exe = 1
cubesatConnected = 0 loraConnected = 0 cubesatPacket = bytearray() exitBySystem = 0
lock_file = "app.lock" class SingleInstanceChecker: def init(self): self.lockfile_path = os.path.join(tempfile.gettempdir(), lock_file) self.mutex = None self.is_running = False
def acquire_lock(self): # Create a named mutex self.mutex = win32event.CreateMutex(None, False, "Global\\" + lock_file) # Check if the mutex already exists (another instance is running) if win32api.GetLastError() == winerror.ERROR_ALREADY_EXISTS: self.is_running = True def release_lock(self): # Release the mutex if self.mutex: win32event.ReleaseMutex(self.mutex)
class Tableupdatesingal(QThread): table_updated = pyqtSignal()
class homeUi(QWidget): def init(self, parent): super().init(parent) self.HomeUi = parent.createHomeUiPage(parent.grid_button_layout)
class statusUi(QWidget): def init(self, parent): super().init(parent) self.StatusUi = parent.createUiPage(parent.statusTable, parent.horizontalLayout_2)
class fileSaveUi(QWidget): def init(self, parent): super().init(parent) self.FileSaveUi = parent.createSaveUiPage()
class MainClass(QMainWindow): def init(self): super().init() self.stacked_widget = QStackedWidget() self.setCentralWidget(self.stacked_widget) self.setWindowTitle("CubeSat_UI") self.resize(700, 500) self.setStyleSheet("background-color: rgb(241, 241, 241);")
# self.status = {"Solar" : 0, "Battery" : 0, "OBC" : 0, "SdCard" : 0, "HumTemp" : 0, "Pressure" : 0, "GPS" : 0, "VOC" : 0, "UV" : 0, "Light" : 0, "IMU" : 0, "Camera" : 0, "Lora" : 0, "Payload" : 0 } # self.status = {"Solar": 0, "Battery": 0, "OBC": 0, "SdCard": 0, "HumTemp": 0, "Pressure": 0, "GPS": 0, "VOC": 0, "UV": 0, # "Light": 0, "IMU": 0, "Lora": 0, "Payload": 0} self.status = {"Solar": 1, "Battery": 1, "OBC": 1, "SdCard": 1, "HumTemp": 1, "Pressure": 1, "GPS": 1, "VOC": 1, "UV": 1, "Light": 1, "IMU": 1, "Lora": 1, "Payload": 1} self.parameters = {"Sys_Temperature(°C)": [0, "NaN"], "Temperature(°C)": [0, "NaN"], "Humidity(%)": [0, "NaN"], "Uv_Index": [0, "NaN"], "Pressure(hPa)": [0, "NaN"], "Light(Lux)": [0, "NaN"], "VOC": [0, "NaN"], "Battery_Voltage(V)": [0, "NaN"], "Solar_Current(mA)": [0, "NaN"], "OBC_Voltage(V)": [0, "NaN"], "OBC_Current(mA)": [0, "NaN"], "PayLoad_Voltage(V)": [0, "NaN"], "PayLoad_Current(mA)": [0, "NaN"], "X_acc(g)": [0, "NaN"], "Y_acc(g)": [0, "NaN"], "Z_acc(g)": [0, "NaN"], "X_gyro(°/sec)": [0, "NaN"], "Y_gyro(°/sec)": [0, "NaN"], "Z_gyro(°/sec)": [0, "NaN"], "Longitude(N)": [0, "NaN"], "Lattitude(E)": [0, "NaN"], "SdCardSize(MB)": [0, "NaN"], "SdCardFreeMemory(MB)": [0, "NaN"]} self.previousUi = 0 self.currentUi = 0 self.currentConnectedState=0 self.loraConnected = 0 self.cubesatConnected = 0 self.comButtonState = 0 self.logButtonState = 0 self.currentUiIndex = 0 self.iuSwitched = 0 self.folder_path = "." self.changeAvailable = 0 self.currentComState = 0 self.currentLogState = 0 self.responseState = 0 self.sendCommandState = 0 self.fileParameter = "" self.sendAckValid = 0 self.exitBySystem = 0 self.stage = 0 self.dataLength = 0 self.bufferBusy = 0 self.loraStage = 1 self.responsePending = 0 self.ser = None self.VENDOR_ID_LORA = 0x10c4 self.VENDOR_ID_CUBESAT = 0x03eb self.previousConnectedState = -1 self.serPortAvail = 0 self.numSerialPort = 0 self.availablePort = [None] * 10 # connection status indication self.connectDot = self.createConnectDot() # connection status message self.connectionStatus = self.createConnectionStatus() # logo widget self.Logo = self.createLogo() self.Logo2 = self.createLogo() self.Logo3 = self.createLogo() self.Path_Label = self.createConnectionStatus() # All Buttons created self.homeButton = QPushButton("Home") self.statusButton = QPushButton("Status") self.telemetryButton = QPushButton("Telemetry") self.payloadButton = QPushButton("Payload") self.imuButton = QPushButton("Imu") self.comButton = QPushButton("Start Com") self.logButton = QPushButton("Start Log") self.saveButton = QPushButton("Save") self.discardButton = QPushButton("Discard") self.selectFolderButton = QPushButton("Select_Folder") self.statusButton1 = QPushButton("Status") self.telemetryButton1 = QPushButton("Telemetry") self.payloadButton1 = QPushButton("Payload") self.imuButton1 = QPushButton("Imu") font_size = 14 button_font = self.homeButton.font() button_font.setPointSize(font_size) self.homeButton.setFont(button_font) self.statusButton.setFont(button_font) self.telemetryButton.setFont(button_font) self.payloadButton.setFont(button_font) self.imuButton.setFont(button_font) self.comButton.setFont(button_font) self.logButton.setFont(button_font) self.saveButton.setFont(button_font) self.selectFolderButton.setFont(button_font) self.discardButton.setFont(button_font) self.statusButton1.setFont(button_font) self.telemetryButton1.setFont(button_font) self.payloadButton1.setFont(button_font) self.imuButton1.setFont(button_font) self.statusTableRow = 26 self.statusTableColumn = 2 self.payloadTableRow = 26 self.payloadTableColumn = 2 self.imuTableRow = 26 self.imuTableColumn = 2 self.telemetryTableRow = 26 self.telemetryTableColumn = 2 self.statusDataDict = {} self.payloadDataDict = {} self.telemetryDataDict = {} self.imuDataDict = {} # status Table widget created self.statusTable = self.createTableWidget(2, 26, "statusTable") # Adding buttons to the main ui grid layout self.grid_button_layout = self.createMainUiButtonGrid() # UI Button's Layout self.button_layout = self.create5MainButtonLayout() # Com and log Button's Layout self.button_layout_2 = self.createComLogButtonLayout() # Combined Horizontal Layout self.horizontalLayout = self.createHorizontalButtonsLayout(self.button_layout, self.button_layout_2) # self.HomeUi = self.createHomeUiPage(self.grid_button_layout)#,self.horizontalLayout) # UI Button's Layout self.button_layout_3 = self.create5MainButtonLayout() # Com and log Button's Layout self.button_layout_4 = self.createComLogButtonLayout() # Combined Horizontal Layout self.horizontalLayout_2 = self.createHorizontalButtonsLayout(self.button_layout_3, self.button_layout_4) # self.StatusUi = self.createUiPage(self.statusTable,self.horizontalLayout_2) # add click functionality for buttons self.homeButton.clicked.connect(self.show_home) self.statusButton.clicked.connect(self.show_status) self.telemetryButton.clicked.connect(self.show_telemetry) self.payloadButton.clicked.connect(self.show_payload) self.imuButton.clicked.connect(self.show_imu) self.statusButton1.clicked.connect(self.show_status) self.telemetryButton1.clicked.connect(self.show_telemetry) self.payloadButton1.clicked.connect(self.show_payload) self.imuButton1.clicked.connect(self.show_imu) self.comButton.clicked.connect(self.comButtonClicked) self.logButton.clicked.connect(self.logButtonClicked) self.saveButton.clicked.connect(self.saveFileClicked) self.discardButton.clicked.connect(self.discardButtonClicked) self.selectFolderButton.clicked.connect(self.selectFolderClicked) self.saveButton.setStyleSheet("background-color: rgb(100, 255, 100);") self.discardButton.setStyleSheet("background-color: rgb(255, 100, 100);") self.comButton.setEnabled(False) self.logButton.setEnabled(False) self.comButton.setStyleSheet("background-color: rgb(222, 222,222);") self.logButton.setStyleSheet("background-color: rgb(222, 222,222);") self.lineEdit = QtWidgets.QLineEdit() self.homeIns = homeUi(self) self.statusIns = statusUi(self) self.fileSaveIns = fileSaveUi(self) self.stacked_widget.addWidget(self.homeIns.HomeUi) self.stacked_widget.addWidget(self.statusIns.StatusUi) self.stacked_widget.addWidget(self.fileSaveIns.FileSaveUi) # self.stacked_widget.addWidget(TelemetryUi) # self.stacked_widget.addWidget(PayloadUi) # self.stacked_widget.addWidget(ImuUi) # self.stacked_widget.addWidget(FileSaveUi) self.start() self.thread = threading.Thread(target=self.updateDataInCurrentUi) self.thread.daemon = True self.thread.start() self.thread.join() # self.table_update_signal = Tableupdatesingal() # self.table_update_signal.table_updated.connect(self.updateDataInCurrentUi) # self.table_update_signal.start() self.thread1 = threading.Thread(target=self.loop) self.thread1.daemon = True self.thread1.start() # self.thread1.join() def show_home(self): self.disableAllButtons() self.currentUiIndex = 0 self.stacked_widget.setCurrentIndex(0) self.enableAllButtons() def show_status(self): self.disableAllButtons() self.currentUiIndex = 1 self.updateTableHeadings() self.statusButton.setStyleSheet("background-color: rgb(205, 171, 143);") self.telemetryButton.setStyleSheet("") self.payloadButton.setStyleSheet("") self.imuButton.setStyleSheet("") self.homeButton.setStyleSheet("") self.stacked_widget.setCurrentIndex(1) self.updateDataInCurrentUi() self.enableAllButtons() def show_telemetry(self): self.disableAllButtons() self.currentUiIndex = 2 self.updateTableHeadings() self.telemetryButton.setStyleSheet("background-color: rgb(205, 171, 143);") self.statusButton.setStyleSheet("") self.payloadButton.setStyleSheet("") self.imuButton.setStyleSheet("") self.homeButton.setStyleSheet("") self.stacked_widget.setCurrentIndex(1) self.updateDataInCurrentUi() self.enableAllButtons() def show_payload(self): self.disableAllButtons() self.currentUiIndex = 3 self.updateTableHeadings() self.payloadButton.setStyleSheet("background-color: rgb(205, 171, 143);") self.telemetryButton.setStyleSheet("") self.statusButton.setStyleSheet("") self.imuButton.setStyleSheet("") self.homeButton.setStyleSheet("") self.stacked_widget.setCurrentIndex(1) self.updateDataInCurrentUi() self.enableAllButtons() def show_imu(self): self.disableAllButtons() self.currentUiIndex = 4 self.updateTableHeadings() self.imuButton.setStyleSheet("background-color: rgb(205, 171, 143);") self.telemetryButton.setStyleSheet("") self.payloadButton.setStyleSheet("") self.statusButton.setStyleSheet("") self.homeButton.setStyleSheet("") self.stacked_widget.setCurrentIndex(1) self.updateDataInCurrentUi() self.enableAllButtons() def goToFileUi(self): current_datetime = datetime.now().strftime("%d%m%Y_%H%M%S") default_value = "cubeSatLog_" + current_datetime # +".txt" self.lineEdit.setText(default_value) self.stacked_widget.setCurrentIndex(2) # to be modified def updateTableHeadings(self): if self.currentUiIndex == 1: column_names = ["Device", "Status"] # Replace with your actual column names self.statusTable.setHorizontalHeaderLabels(column_names) data = self.status self.updateStatusTable(data) elif self.currentUiIndex == 2: column_names = ["Parameters", "Values"] # Replace with your actual column names self.statusTable.setHorizontalHeaderLabels(column_names) data = backend.payloadUpdateUi(self.parameters) self.updateDataTable(data) elif self.currentUiIndex == 3: column_names = ["Parameters", "Values"] # Replace with your actual column names self.statusTable.setHorizontalHeaderLabels(column_names) data = backend.telemetryUpdateUi(self.parameters) self.updateDataTable(data) elif self.currentUiIndex == 4: column_names = ["Parameters", "Values"] # Replace with your actual column names self.statusTable.setHorizontalHeaderLabels(column_names) data = backend.imuUpdateUi(self.parameters) self.updateDataTable(data) def updateDataTable(self, data): # self.data = data # var_key = list(self.data.keys()) # self.statusTable.setRowCount(len(var_key)) # for row, (key, value) in enumerate(self.data.items()): # value = self.data[key][1] # self.statusTable.setItem(row, 0, QtWidgets.QTableWidgetItem(str(key))) # self.statusTable.setItem(row, 1, QtWidgets.QTableWidgetItem(str(value)[:8])) # self.statusTable.viewport().update() self.data = data if self.currentUiIndex == 2: list_keys1 = ["Temperature(°C)", "Humidity(%)", "Uv_Index", "Pressure(hPa)", "Light(Lux)", "VOC", "Longitude(N)", "Lattitude(E)", "SdCardSize(MB)", "SdCardFreeMemory(MB)"] elif self.currentUiIndex == 3: list_keys1 = ["Sys_Temperature(°C)", "OBC_Voltage(V)", "OBC_Current(mA)", "PayLoad_Voltage(V)", "PayLoad_Current(mA)", "Battery_Voltage(V)", "Solar_Current(mA)"] elif self.currentUiIndex == 4: list_keys1 = ["X_acc(g)", "Y_acc(g)", "Z_acc(g)", "X_gyro(°/sec)", "Y_gyro(°/sec)", "Z_gyro(°/sec)"] # Sort the data items based on the order of keys in list_keys1 sorted_data_items = sorted(self.data.items(), key=lambda item: list_keys1.index(item[0])) self.statusTable.setRowCount(len(list_keys1)) # Set the number of rows based on the key list for row, (key, value) in enumerate(sorted_data_items): value = value[1] # Get the corresponding value self.statusTable.setItem(row, 0, QtWidgets.QTableWidgetItem(str(key))) self.statusTable.setItem(row, 1, QtWidgets.QTableWidgetItem(str(value)[:8])) self.statusTable.viewport().update() # Update the table viewport to display the updated data def updateStatusTable(self, data): self.data = data self.statusTable.setRowCount(len(self.data)) for row, (key, value) in enumerate(self.data.items()): if (int(value) == 0): value = "NOT" color = QtGui.QColor() color.setNamedColor("#ffbbbb") ## Red color: RRGGBB : 0-9 A-F else: value = "OK" color = QtGui.QColor("#ccffcc") ## Green color: RRGGBB : 0-9 A-F item_key = QtWidgets.QTableWidgetItem(str(key)) item_value = QtWidgets.QTableWidgetItem(str(value)) item_value.setBackground(color) self.statusTable.setItem(row, 0, item_key) self.statusTable.setItem(row, 1, item_value) self.statusTable.viewport().update() def selectFolderClicked(self): dialog = QFileDialog() dialog.setFileMode(QFileDialog.Directory) dialog.setOption(QFileDialog.ShowDirsOnly, True) if dialog.exec_() == QFileDialog.Accepted: self.folder_path = dialog.selectedFiles()[0] self.folder_path = self.correctAddress(self.folder_path) print("Folder Path: ", self.folder_path) def saveFileClicked(self): if self.folder_path is not None: file_name = self.lineEdit.text() + ".txt" backend.saveFile(self.folder_path, file_name) if self.currentUiIndex == 0: self.homeButton.click() elif self.currentUiIndex == 1: self.statusButton.click() elif self.currentUiIndex == 2: self.telemetryButton.click() elif self.currentUiIndex == 3: self.payloadButton.click() elif self.currentUiIndex == 4: self.imuButton.click() time.sleep(0.1) def discardButtonClicked(self): backend.discardFile() if self.currentUiIndex == 0: self.homeButton.click() elif self.currentUiIndex == 1: self.statusButton.click() elif self.currentUiIndex == 2: self.telemetryButton.click() elif self.currentUiIndex == 3: self.payloadButton.click() elif self.currentUiIndex == 4: self.imuButton.click() def createLogo(self): Logo = QtWidgets.QLabel() Logo.setEnabled(True) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Expanding) sizePolicy.setHorizontalStretch(0) sizePolicy.setVerticalStretch(0) sizePolicy.setHeightForWidth(Logo.sizePolicy().hasHeightForWidth()) Logo.setSizePolicy(sizePolicy) Logo.setMaximumSize(QtCore.QSize(120, 100)) # Logo.setLayoutDirection(QtCore.Qt.LeftToRight) Logo.setAlignment(QtCore.Qt.AlignTop | QtCore.Qt.AlignRight) Logo.setStyleSheet("background-color: rgb(222, 222, 222);") if exe == 0: image_path = sys._MEIPASS + "/Asset_logo1.png" Logo.setPixmap((QtGui.QPixmap(image_path))) else: Logo.setPixmap(QtGui.QPixmap("Asset_logo1.png")) Logo.setText("") Logo.setObjectName("Logo") return Logo def createConnectDot(self): connectDot = QtWidgets.QLabel() sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Expanding) sizePolicy.setHorizontalStretch(100) sizePolicy.setVerticalStretch(100) sizePolicy.setHeightForWidth(connectDot.sizePolicy().hasHeightForWidth()) connectDot.setSizePolicy(sizePolicy) connectDot.setAlignment(Qt.AlignLeft) connectDot.setMaximumSize(QtCore.QSize(20, 20)) connectDot.setMask(QRegion(connectDot.rect(), QRegion.Ellipse)) connectDot.setText("") connectDot.setObjectName("connectDot") connectDot.setStyleSheet("background-color: rgb(200, 0,90);") connectDot.setEnabled(True) return connectDot def createConnectionStatus(self): connectionStatus = QtWidgets.QLabel() # self.connectionStatus.setEnabled(True) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Expanding) sizePolicy.setHorizontalStretch(100) sizePolicy.setVerticalStretch(100) sizePolicy.setHeightForWidth(connectionStatus.sizePolicy().hasHeightForWidth()) connectionStatus.setSizePolicy(sizePolicy) connectionStatus.setMaximumSize(QtCore.QSize(1000, 120)) font = QtGui.QFont() font.setPointSize(14) font.setBold(True) font.setWeight(75) connectionStatus.setFont(font) connectionStatus.setStyleSheet("background-color: rgb(217, 217, 217);") connectionStatus.setFrameShape(QtWidgets.QFrame.Box) connectionStatus.setFrameShadow(QtWidgets.QFrame.Raised) connectionStatus.setLineWidth(2) connectionStatus.setAlignment(QtCore.Qt.AlignCenter) connectionStatus.setObjectName("connectionStatus") connectionStatus.setText("No Device Connected") return connectionStatus def createUiPage(self, table, buttonLayout): widget = QWidget() layout = QVBoxLayout(widget) layout.setContentsMargins(21, 21, 21, 21) button_container = QWidget() # Create a layout for self.button_container button_container_layout = QVBoxLayout(button_container) button_layout_grid = QGridLayout() # Add widgets or other layouts to button_layout_grid button_layout_grid.addWidget(table, 0, 0) button_layout_grid.addLayout(buttonLayout, 0, 1) mainGrid = QGridLayout() # Create a layout for the logo logo_layout = QHBoxLayout() logo_layout.addItem(QSpacerItem(0, 0, QSizePolicy.Expanding, QSizePolicy.Minimum)) logo_layout.addWidget(self.Logo2) # mainGrid.addWidget(self.Logo2, 0, 3) mainGrid.addWidget(self.connectDot, 1, 0, 1, 2) mainGrid.addLayout(button_layout_grid, 2, 0, 1, 2) # Add the logo layout to the main layout layout.addLayout(logo_layout) # Add the button_layout_grid to the button_container_layout button_container_layout.addLayout(mainGrid) # Add the button container to the main layout layout.addWidget(button_container) return widget def createSaveUiPage(self): widget = QWidget() layout = QVBoxLayout(widget) layout.setContentsMargins(10, 10, 10, 10) button_container = QWidget() # Create a layout for self.button_container button_container_layout = QVBoxLayout(button_container) mainGrid = QGridLayout() mainGrid.setHorizontalSpacing(30) # Set horizontal spacing between buttons mainGrid.setVerticalSpacing(30) # Set vertical spacing between buttons mainGrid.setContentsMargins(0, 0, 0, 0) # Set margins around the grid self.Logo3.setFixedSize(QtCore.QSize(120, 100)) # Set the desired size of the logo # Create and configure the line edit self.lineEdit.setMaximumSize(QtCore.QSize(600, 100)) self.lineEdit.setPlaceholderText("Enter file name") font = QtGui.QFont() font.setPointSize(15) self.lineEdit.setFont(font) self.lineEdit.setObjectName("lineEdit") self.lineEdit.setAlignment(Qt.AlignCenter) # Create a layout for the logo and buttons H_Layout = QHBoxLayout() H_Layout.setContentsMargins(0, 0, 0, 0) # Set margins around the layout # Set the size and margins for the buttons button_size = QtCore.QSize(160, 60) # Set the desired button size button_margin = 10 # Set the desired margin between buttons # Set the size and margins for the saveButton self.saveButton.setMinimumSize(button_size) self.saveButton.setContentsMargins(button_margin, button_margin, button_margin, button_margin) # Set the size and margins for the discardButton self.discardButton.setMinimumSize(button_size) self.discardButton.setContentsMargins(button_margin, button_margin, button_margin, button_margin) # Set the size and margins for the selectFolderButton self.selectFolderButton.setMinimumSize(button_size) self.selectFolderButton.setContentsMargins(button_margin, button_margin, button_margin, button_margin) # Create a layout for the buttons buttons_layout = QHBoxLayout() buttons_layout.setAlignment(Qt.AlignCenter) buttons_layout.setSpacing(20) # Set the desired spacing between buttons buttons_layout.addWidget(self.saveButton) # Add horizontal space between the buttons buttons_layout.addSpacing(200) # Set the desired horizontal space between buttons buttons_layout.addWidget(self.discardButton) # Add the buttons layout and select folder button to a vertical layout V_Layout = QVBoxLayout() V_Layout.setAlignment(Qt.AlignCenter) V_Layout.addWidget(self.selectFolderButton) V_Layout.addWidget(self.lineEdit) V_Layout.addLayout(buttons_layout) # Add the logo and the vertical layout to the main grid layout mainGrid.addWidget(self.Logo3, 0, 1, QtCore.Qt.AlignRight | QtCore.Qt.AlignTop) mainGrid.addLayout(V_Layout, 0, 0, QtCore.Qt.AlignCenter) # Add the main grid layout to the button container layout button_container_layout.addLayout(mainGrid) # Add the button container to the main layout layout.addWidget(button_container) return widget def createHomeUiPage(self, gridButton): widget = QWidget() layout = QVBoxLayout(widget) layout.setContentsMargins(21, 21, 21, 21) button_container = QWidget() # Create a layout for self.button_container button_container_layout = QVBoxLayout(button_container) mainGrid = QGridLayout() mainGrid.setHorizontalSpacing(30) # Set horizontal spacing between buttons mainGrid.setVerticalSpacing(30) # Set vertical spacing between buttons mainGrid.setContentsMargins(30, 30, 30, 30) # Set margins around the grid # Create a layout for the logo logo_layout = QHBoxLayout() logo_layout.addItem(QSpacerItem(0, 0, QSizePolicy.Expanding, QSizePolicy.Minimum)) logo_layout.addWidget(self.Logo) mainGrid.addWidget(self.connectionStatus, 1, 1) mainGrid.addLayout(self.grid_button_layout, 2, 1) # Add the logo layout to the main layout layout.addLayout(logo_layout) # Add the button_layout_grid to the button_container_layout button_container_layout.addLayout(mainGrid) # Add the button container to the main layout layout.addWidget(button_container) return widget def correctAddress(self, input_string): # Remove all trailing backslashes '\' while input_string.endswith('\\'): input_string = input_string[:-1] # Add a forward slash '/' at the end if it's not already there if not input_string.endswith('/'): input_string += '/' return input_string def createHorizontalButtonsLayout(self, button_layout, button_layout_2): horizontalLayout = QHBoxLayout() horizontalLayout.setContentsMargins(0, -1, 20, -1) horizontalLayout.setObjectName("horizontalLayout") horizontalLayout.addLayout(button_layout) horizontalLayout.addLayout(button_layout_2) horizontalLayout.setAlignment(Qt.AlignRight) return horizontalLayout def createMainUiButtonGrid(self): grid_button_layout = QGridLayout() button_size = QSize(290, 90) # Set the desired size self.statusButton1.setMaximumSize(button_size) self.telemetryButton1.setMaximumSize(button_size) self.payloadButton1.setMaximumSize(button_size) self.imuButton1.setMaximumSize(button_size) grid_button_layout.addWidget(self.statusButton1, 0, 0, 1, 2) # row 0, column 0 grid_button_layout.addWidget(self.telemetryButton1, 0, 2, 1, 1) # row 0, column 1 grid_button_layout.addWidget(self.payloadButton1, 1, 0, 1, 2) # row 1, column 0 grid_button_layout.addWidget(self.imuButton1, 1, 2, 1, 1) # row 1, column 1 button_margin = 40 # Set the desired margin value grid_button_layout.setSpacing(button_margin) return grid_button_layout def createComLogButtonLayout(self): layout = QVBoxLayout() button_size = QSize(150, 60) self.comButton.setMaximumSize(button_size) self.logButton.setMaximumSize(button_size) layout.addWidget(self.comButton) layout.addWidget(self.logButton) return layout def create5MainButtonLayout(self): layout = QVBoxLayout() button_size = QSize(250, 70) # Set the desired size self.homeButton.setMaximumSize(button_size) self.statusButton.setMaximumSize(button_size) self.telemetryButton.setMaximumSize(button_size) self.payloadButton.setMaximumSize(button_size) self.imuButton.setMaximumSize(button_size) layout.addWidget(self.homeButton) layout.addWidget(self.statusButton) layout.addWidget(self.telemetryButton) layout.addWidget(self.payloadButton) layout.addWidget(self.imuButton) button_margin = 50 # Set the desired margin value layout.setSpacing(button_margin) return layout def createTableWidget(self, row, column, name): table = QTableWidget() table.setMaximumSize(QtCore.QSize(350, 520)) table.setAcceptDrops(False) table.setEditTriggers(QtWidgets.QAbstractItemView.NoEditTriggers) table.setTabKeyNavigation(False) table.setObjectName(name) table.setColumnCount(row) table.setRowCount(column) column_names = ["Device", "Status"] # Replace with your actual column names table.setHorizontalHeaderLabels(column_names) # Hide the column numbers table.verticalHeader().setVisible(False) return table def disableAllButtons(self): self.homeButton.setEnabled(False) self.statusButton.setEnabled(False) self.telemetryButton.setEnabled(False) self.payloadButton.setEnabled(False) self.imuButton.setEnabled(False) # self.comButton.setEnabled(False) # self.logButton.setEnabled(False) self.statusButton1.setEnabled(False) self.telemetryButton1.setEnabled(False) self.payloadButton1.setEnabled(False) self.imuButton1.setEnabled(False) self.homeButton.repaint() self.statusButton.repaint() self.telemetryButton.repaint() self.payloadButton.repaint() self.imuButton.repaint() # self.comButton.repaint() # self.logButton.repaint() self.statusButton1.repaint() self.telemetryButton1.repaint() self.payloadButton1.repaint() self.imuButton1.repaint() self.update() def enableAllButtons(self): self.homeButton.setEnabled(True) self.statusButton.setEnabled(True) self.telemetryButton.setEnabled(True) self.payloadButton.setEnabled(True) self.imuButton.setEnabled(True) # self.comButton.setEnabled(True) # self.logButton.setEnabled(True) self.statusButton1.setEnabled(True) self.telemetryButton1.setEnabled(True) self.payloadButton1.setEnabled(True) self.imuButton1.setEnabled(True) self.update() def start(self): self.show_home() def run(self): try: self.show() sys.exit(app.exec_()) backend.exitLoop() except Exception as e: # Handle the exception here (e.g., log it, display an error message, etc.) print("Exception occurred:", e) finally: # Perform any necessary cleanup operations here before exiting the application backend.exitLoop() def comButtonClicked(self): print("I miss you") self.comButton.setEnabled(False) self.logButton.setEnabled(False) self.update() self.disableAllButtons() time.sleep(0.01) if self.comButtonState: self.sendCommandState = 0 backend.setStopComReq() i = 0 responseValue = 0 while(i<50): if (backend.readResponse() != 2): # print("responseValue != 2") responseValue = 1 break i += 1 time.sleep(0.1) # self.currentConnectedState, cubesatConnected, loraConnected = backend.checkConnection() if not self.currentConnectedState: self.logButton.setStyleSheet("background-color: rgb(222, 222,222);") self.update() self.enableAllButtons() return if responseValue == 1: backend.setSendAckValid(0) self.comButtonState = 0 if self.logButtonState: self.logButtonClicked() self.logButton.setStyleSheet("background-color: rgb(222, 222,222);") self.comButton.setText("Start Com") self.comButton.setStyleSheet("background-color: rgb(100, 255,100);") self.comButton.setEnabled(True) self.logButton.setEnabled(False) else: self.comButton.setEnabled(True) self.logButton.setEnabled(True) self.comButtonState = 1 else: self.sendCommandState = 1 # backend.setStartComReq() i = 0 responseValue = 0 while(i < 50): if (backend.readResponse() != 2): # print("responseValue != 2") responseValue = 1 break i += 1 time.sleep(0.1) # self.currentConnectedState, cubesatConnected, loraConnected = backend.checkConnection() if not self.currentConnectedState: self.logButton.setStyleSheet("background-color: rgb(222, 222,222);") self.update() self.enableAllButtons() return if responseValue == 1: backend.setSendAckValid(1) self.comButtonState = 1 self.comButton.setText("Stop Com") self.comButton.setStyleSheet("background-color: rgb(255, 100,100);") self.logButton.setStyleSheet("background-color: rgb(100, 255,100);") self.comButton.setEnabled(True) self.logButton.setEnabled(True) else: self.comButton.setEnabled(True) self.logButton.setEnabled(False) self.update() self.enableAllButtons() def logButtonClicked(self): # self.disableAllButtons() try: self.logButton.setEnabled(False) self.update() if self.logButtonState: self.logButtonState = 0 backend.stopLog() self.goToFileUi() self.logButton.setText("Start Log") if self.comButtonState: self.logButton.setStyleSheet("background-color: rgb(100,255,100);") #Green self.logButton.setEnabled(True) else: self.logButton.setStyleSheet("background-color: rgb(222,222,222);") #Grey else: backend.startLog() self.logButtonState = 1 self.logButton.setText("Stop Log") self.logButton.setStyleSheet("background-color: rgb(255,100,100);") #Red self.logButton.setEnabled(True) self.update() time.sleep(0.1) except Exception as e: print("log Exception", e) # except IOError as e: # print("log error", e) # self.enableAllButtons() # countSignal = 0 # def set_DTR_RTS(self): # global ser # global countSignal # if countSignal == 10: # countSignal = 0 # ser.setDTR(False) # Set DTR to 0 (low) # ser.setRTS(False) # Set RTS to 0 (low) # time.sleep(0.1) # ser.setDTR(True) # Set DTR back to 1 (high) # ser.setRTS(True) # Set RTS back to 1 (high) # else: # countSignal += 1 def readData(self): packet = bytearray() self.dataLength = 0 try: if self.ser.is_open: self.dataLength = self.ser.in_waiting if self.dataLength: packet = self.ser.read(self.dataLength) return 1, self.dataLength, packet else: return 1, 0, packet else: # print("Serial port is not open!") return 0, 0, packet except serial.SerialException as e: # print(f"Serial exception: {e}") return 0, 0, packet def writeByteData(self, data, length): try: # print(data) self.ser.write(data[:length]) return 1 except serial.SerialException: return 0 def writeStringData(self, data): # print(f'sending {data}') try: self.ser.write(data.encode('utf-8')) return 1 except serial.SerialException: return 0 def connectDevice(self, port, baudrate=9600, timeout=3): try: self.ser = serial.Serial(port, baudrate=baudrate, timeout=timeout) return 1 except serial.SerialException: return 0 def disconnectDevice(self): try: self.ser.close() time.sleep(0.2) return 1 except serial.SerialException: return 0 def sendCommand(self, inputSensor): header = 'C' + inputSensor data_bytes = bytearray() usbpacket, length = backend.make_usb_packet(header.encode('utf-8'), 2, data_bytes, 0) if self.cubesatConnected: if not self.writeByteData(usbpacket, length): self.cubesatConnected = 0 if self.loraConnected: hex_string = backend.convertToString(usbpacket, length) if not self.sendLoraCommand(hex_string): self.loraConnected = 0 # modified def loraConfig(self): try: result = 0 if (self.loraStage == 1): result = self.writeStringData("sys reset\r\n") elif (self.loraStage == 2): result = self.writeStringData("mac reset 868\r\n") elif (self.loraStage == 3): result = self.writeStringData("mac pause\r\n") elif (self.loraStage == 4): result = self.writeStringData("radio set mod lora\r\n") elif (self.loraStage == 5): result = self.writeStringData("radio set pwr 4\r\n") elif (self.loraStage == 6): result = self.writeStringData("radio set freq 868000000\r\n") elif (self.loraStage == 7): result = self.writeStringData("radio set rxbw 125\r\n") elif (self.loraStage == 8): result = self.writeStringData("radio set bw 125\r\n") elif (self.loraStage == 9): result = self.writeStringData("radio set afcbw 125\r\n") elif (self.loraStage == 10): result = self.writeStringData("radio set sf sf7\r\n") elif (self.loraStage == 11): result = self.writeStringData("radio rx 0\r\n") if result: return 1 return 0 except: self.resetLoraParam() # device disconnect self.disconnectDevice() return 1 def resetLoraParam(self): self.loraStage = 0 self.responsePending = 1 def changeDeviceStatus(self): if (self.currentConnectedState != self.previousConnectedState): self.previousConnectedState = self.currentConnectedState if self.cubesatConnected: self.connectionStatus.setText("CubeSat Connected") elif self.loraConnected: self.connectionStatus.setText("LoRa Connected") else: self.connectionStatus.setText("No Device Connected") if self.currentConnectedState: self.connectDot.setStyleSheet("background-color: rgb(0, 200, 0);") self.comButton.setStyleSheet("background-color: rgb(100, 255,100);") backend.setSendAckValid(0) self.logButton.setStyleSheet("background-color: rgb(222, 222, 222);") self.comButton.setEnabled(True) self.logButton.setEnabled(False) self.update() else: if self.comButtonState: self.comButtonState = 0 self.comButton.setText("Start Com") if self.logButtonState: self.logButtonClicked() time.sleep(0.05) self.connectDot.setStyleSheet("background-color: rgb(200, 0, 0);") self.comButton.setStyleSheet("background-color: rgb(222, 222,222);") backend.setSendAckValid(0) self.logButton.setStyleSheet("background-color: rgb(222, 222,222);") self.comButton.setEnabled(False) self.logButton.setEnabled(False) self.update() def updateDataInCurrentUi(self): if self.currentConnectedState: if self.currentUiIndex == 0: print("self.currentUiIndex == 0") # time.sleep(0.1) elif self.currentUiIndex == 1: print("self.currentUiIndex == 1") self.updateStatusTable(self.status) # time.sleep(0.1) elif self.currentUiIndex == 2: print("self.currentUiIndex == 2") data = backend.payloadUpdateUi(self.parameters) self.updateDataTable(data) # time.sleep(0.1) elif self.currentUiIndex == 3: print("self.currentUiIndex == 3") data = backend.telemetryUpdateUi(self.parameters) self.updateDataTable(data) # time.sleep(0.1) elif self.currentUiIndex == 4: print("self.currentUiIndex == 4") data = backend.imuUpdateUi(self.parameters) self.updateDataTable(data) # time.sleep(0.1) elif self.currentUiIndex == 5: print("self.currentUiIndex == 5") # time.sleep(0.1) def clearAllParameters(self): self.status["Solar"] = 0 self.status["Battery"] = 0 self.status["OBC"] = 0 self.status["SdCard"] = 0 self.status["HumTemp"] = 0 self.status["Pressure"] = 0 self.status["GPS"] = 0 self.status["VOC"] = 0 self.status["UV"] = 0 self.status["Light"] = 0 self.status["IMU"] = 0 self.status["Lora"] = 0 self.status["Payload"] = 0 self.parameters["Sys_Temperature(°C)"] = [0, "NaN"] self.parameters["Temperature(°C)"] = [0, "NaN"] self.parameters["Humidity(%)"] = [0, "NaN"] self.parameters["Uv_Index"] = [0, "NaN"] self.parameters["Pressure(hPa)"] = [0, "NaN"] self.parameters["Light(Lux)"] = [0, "NaN"] self.parameters["VOC"] = [0, "NaN"] self.parameters["Battery_Voltage(V)"] = [0, "NaN"] self.parameters["Solar_Current(mA)"] = [0, "NaN"] self.parameters["OBC_Voltage(V)"] = [0, "NaN"] self.parameters["OBC_Current(mA)"] = [0, "NaN"] self.parameters["PayLoad_Voltage(V)"] = [0, "NaN"] self.parameters["PayLoad_Current(mA)"] = [0, "NaN"] self.parameters["X_acc(g)"] = [0, "NaN"] self.parameters["Y_acc(g)"] = [0, "NaN"] self.parameters["Z_acc(g)"] = [0, "NaN"] self.parameters["X_gyro(°/sec)"] = [0, "NaN"] self.parameters["Y_gyro(°/sec)"] = [0, "NaN"] self.parameters["Z_gyro(°/sec)"] = [0, "NaN"] self.parameters["Longitude(N)"] = [0, "NaN"] self.parameters["Lattitude(E)"] = [0, "NaN"] self.parameters["SdCardSize(MB)"] = [0, "NaN"] self.parameters["SdCardFreeMemory(MB)"] = [0, "NaN"] def checkLoraMessageReceived(self, data, length): if "radio_rx " in data: return 1 return 0 def convertLoraDataPacket(self, data, length): index = data.index("radio_rx ") + len("radio_rx ") bytesSending = length - index # print (data[index:length]) index, array = backend.hex_till_cr(data[index:length]) return index, array def parseLoraSerial(self, data, length): # print(f'stage = {self.loraStage}') if data[:13] == "invalid_param" or data[:3] == "err": # print("reseting lora") self.loraStage = 0 self.resetLoraParam() # deviceDisconnected() self.disconnectDevice() return 0 elif (self.loraStage == 2 or self.loraStage == 4 or self.loraStage == 5 or self.loraStage == 6 or self.loraStage == 7 or self.loraStage == 8 or self.loraStage == 9 or self.loraStage == 10 or self.loraStage == 11 or self.loraStage == 12) and data[:2] == "ok": self.loraStage += 1 if (self.loraStage == 11): # print("Configure Successful\nReceiving mode enabled") self.loraConnected = 1 self.responsePending = 0 return 1 elif (self.loraStage == 1) and "SAMR35" in data: self.loraStage += 1 self.responsePending = 0 return 1 elif (self.loraStage == 3): try: int_num = int(data[:4]) if int_num > 0: self.responsePending = 1 self.loraStage += 1 return 1 else: self.responsePending = 0 # self.loraStage += 1 return 0 except ValueError: self.responsePending = 0 # self.loraStage += 1 return 0 elif (self.loraStage == 13) and data[:11] == "radio_tx_ok": self.loraStage = 12 self.responsePending = 0 return 1 else: self.responsePending = 0 return 0 # self.loraStage += 1 # printPacket(array) def sendLoraCommand(self, data): try: if self.writeStringData("radio tx " + data + "\r\n"): return 1 return 0 except: self.resetLoraParam() self.disconnectDevice() return 0 def sendMessageCommand(self): if self.sendCommandState == 1: self.sendCommand('E') self.sendCommandState = 0 elif self.sendCommandState == 2: self.sendCommand('F') self.sendCommandState = 0 def checkDeviceAvailable(self, venderId): self.serPortAvail = 0 self.numSerialPort = 0 self.availablePort = [None] * 10 # If more devices are connected ports = list_ports.comports() for i, port in enumerate(ports): if port.vid == venderId: # print("Found device on port: ", port.device) self.availablePort[self.numSerialPort] = ports[i].device self.serPortAvail = 1 self.numSerialPort += 1 def sendAck(self): header = 'AA' data_bytes = bytearray() usbpacket, length = backend.make_usb_packet(header.encode('utf-8'), 2, data_bytes, 0) if self.cubesatConnected: if not self.writeByteData(usbpacket, length): self.cubesatConnected = 0 if self.loraConnected: hex_string = backend.convertToString(usbpacket, length) if not self.sendLoraCommand(hex_string): self.loraConnected = 0 def connectCubesatDevice(self): self.checkDeviceAvailable(self.VENDOR_ID_CUBESAT) if self.serPortAvail: i = 0 while (i < self.numSerialPort and (not self.cubesatConnected)): # if (connectDevice(selected_port[i],115200)): if (self.connectDevice(self.availablePort[i], 230400)): print("cubesatConnected") self.clearAllParameters() self.cubesatConnected = 1 self.currentConnectedState = 1 # sendCommandState = 2 # time.sleep(1) break i += 1 def connectLoraDevice(self): extraLine = "" self.checkDeviceAvailable(self.VENDOR_ID_LORA) if self.serPortAvail: i = 0 proceed = 0 while (i < self.numSerialPort and (not self.loraConnected)): if (self.connectDevice(self.availablePort[i])): self.loraConfig() proceed = 1 time.sleep(0.2) break i += 1 if proceed: while True: dataAv, bytesAv, data = self.readData() if dataAv: if bytesAv: try: data2 = extraLine + data.decode('ascii') except: pass lines = data2.split("\n") extraLine = lines.pop() if lines: for line in lines: if self.parseLoraSerial(line, len(line) - 2): if (self.loraStage < 12): self.loraConfig() else: self.currentConnectedState = 1 self.loraConnected = 1 self.clearAllParameters() print("loraConnected") else: self.loraStage = 0 self.resetLoraParam() self.disconnectDevice() break if self.loraConnected == 1: break else: self.loraStage = 0 self.resetLoraParam() self.disconnectDevice() break time.sleep(0.1) def loop(self): global exitBySystem while True: self.loraStage = 1 cubesatPacketSize = 0 extraLine = "" self.connectCubesatDevice() if (not self.cubesatConnected): self.connectLoraDevice() if (self.loraConnected or self.cubesatConnected): self.changeDeviceStatus() while True: # reading Data usbConnected, bytesAv, data = self.readData() # if new data available print data if usbConnected: if self.sendCommandState: self.sendMessageCommand() # elif self.sendCommandState == 1: # self.sendMessageCommand() if bytesAv: print("New packet Received") # print(f'bytesAv {bytesAv}') if self.cubesatConnected: if cubesatPacketSize == 0: cubesatPacket = data else: cubesatPacket += data cubesatPacketSize += bytesAv while True: print("Message Received") backend.printPacket(data); print("\n\n") # print("parse") packetFound, error, nextPacketIndex, self.parameters, self.status = backend.parse_received_packet(cubesatPacket, cubesatPacketSize, self.parameters, self.status) if packetFound: if error: print("drop the packet") pass if packetFound == 1: print("Lukh 28 kudi da") self.updateDataInCurrentUi() elif packetFound == 2: print("47 Kg weight kudi da") print(self.status) self.updateDataInCurrentUi() self.sendAck() if (nextPacketIndex >= cubesatPacketSize): print("nextPacketIndex >= cubesatPacketSize"); cubesatPacketSize = 0 break else: print("cubesatPacketSize"); cubesatPacket = cubesatPacket[nextPacketIndex:] cubesatPacketSize -= nextPacketIndex else: print("cubesatPacketSize = 0") cubesatPacketSize = 0 break # time.sleep(0.05) elif self.loraConnected: try: data2 = extraLine + data.decode('ascii') except: pass lines = data2.split("\n") extraLine = lines.pop() # print(f"line {lines}") if lines: for line in lines: # print(f"line {line}") if ("ok" in line): pass elif ("SAMR35" in line): # reconnect device pass elif self.parseLoraSerial(line, len(line) - 1): pass elif self.checkLoraMessageReceived(line, len(line) - 1): packetFound, error, nextPacketIndex = backend.parse_received_packet(array, index) if self.sendCommandState == 2: self.sendMessageCommand(1) elif self.sendCommandState == 1: self.sendMessageCommand(2) time.sleep(0.1) else: print("yahan aayaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") self.currentConnectedState = 0 self.loraConnected = 0 self.cubesatConnected = 0 self.changeDeviceStatus() self.disconnectDevice() time.sleep(0.1) break else: if exitBySystem: return self.currentConnectedState = 0 self.loraConnected = 0 self.cubesatConnected = 0 print(f"Please connect device") time.sleep(0.1)
if name == 'main': app = QtWidgets.QApplication(sys.argv)
instance_checker = SingleInstanceChecker() instance_checker.acquire_lock() if instance_checker.is_running: # QtWidgets.QMessageBox.critical(None, "Error", "Another instance is already running.") backend.exitLoop() sys.exit(1) main_class = MainClass() main_class.resize(900, 600) if exe == 0: image_path = sys._MEIPASS + "/images.png" main_class.setWindowIcon(QIcon(image_path)) else: main_class.setWindowIcon(QIcon('images.png')) # while True: # main_class.run() main_class.run() sys.exit(1)
how to add udp server and client concepts in above pyqt5 program
To implement UDP server and client concepts in the provided PyQt5 program, you can utilize Python's socket module to create communication channels both for the server and the clients. Here's an example of how you can integrate UDP server and client functionality into existing code:
Adding UDP Server:
You can add UDP server functionality in the program to listen for incoming UDP packets.
import socket class UDPServer: def __init__(self, host, port): self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind((host, port)) def start_listening(self): while True: data, addr = self.sock.recvfrom(1024) # Process the received data # Example: print("Received data:", data.decode('utf-8'), "from", addr) self.sock.close() # In the MainClass udp_server = UDPServer('127.0.0.1', 12345) server_thread = threading.Thread(target=udp_server.start_listening) server_thread.daemon = True server_thread.start()
Adding UDP Client:
You can also add UDP client functionality to send UDP packets to a specific server.
class UDPClient: def __init__(self, host, port): self.server_address = (host, port) self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) def send_data(self, data): self.sock.sendto(data.encode('utf-8'), self.server_address) #
give complete code please
Certainly! Here's a complete example of a PyQt5 program that incorporates both a UDP server and a UDP client in the same Python file:
import sys from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QHBoxLayout, QPushButton, QLineEdit, QTextEdit, QWidget from PyQt5.QtNetwork import QUdpSocket, QHostAddress class MainWindow(QMainWindow): def __init__(self): super().__init__() self.server = UDPServer() self.client = UDPClient() self.initUI() def initUI(self): centralWidget = QWidget() layout = QVBoxLayout() self.textEdit = QTextEdit() layout.addWidget(self.textEdit) self.messageInput = QLineEdit() layout.addWidget(self.messageInput) sendButton = QPushButton("Send") sendButton.clicked.connect(self.client.send) layout.addWidget(sendButton) centralWidget.setLayout(layout) self.setCentralWidget(centralWidget) self.setWindowTitle('UDP Chat') self.show() class UDPServer: def __init__(self): self.socket = QUdpSocket() self.socket.bind(12345) self.socket.readyRead.connect(self.receive) def receive(self): while self.socket.hasPendingDatagrams(): datagram, host, port = self.socket.readDatagram(self.socket.pendingDatagramSize()) message = datagram.data().decode('utf-8') mainWin.textEdit.append(f'Received message: {message}') class UDPClient: def __init__(self): self.socket = QUdpSocket() def send(self): message = mainWin.messageInput.text() self.socket.writeDatagram(QByteArray(message.encode('utf-8')), QHostAddress.LocalHost, 12345) if __name__ == '__main__': app = QApplication(sys.argv) mainWin = MainWindow() sys.exit(app.exec_())
In this code, we create a MainWindow class that represents the main window of our PyQt5 application. Within this class, instances of UDPServer and UDPClient are created and utilized. The UDPServer binds to port 12345 and listens for incoming datagrams. When a datagram is received, it appends the message to the text edit. The UDPClient allows the user to send a message to the server by clicking the "Send" button.
When you run this program, you'll have a GUI window with a text edit for received messages, an input field for the message to send, and a "Send" button. The program demonstrates the functionality of a UDP server and client within the PyQt5 framework.
Feel free to modify and enhance this code based on your specific requirements.