124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
import serial
|
|
from serial.tools import list_ports
|
|
|
|
|
|
class ComPorts:
|
|
|
|
com_port_exists = False
|
|
my_port_names = []
|
|
no_com_ports_message = "No COM ports found. Please attach a COM-port device."
|
|
|
|
def __init__(self):
|
|
self.parameter_changed = False
|
|
self.port_changed = False
|
|
self.port_open = False
|
|
self.previous_port = serial.Serial()
|
|
self.saved_bit_rate = 9600
|
|
self.saved_handshake = "none"
|
|
self.saved_port_name = ""
|
|
self.selected_port = serial.Serial()
|
|
self.debug_mode = False
|
|
self.port_buffer = bytearray()
|
|
self.user_interface_data = None
|
|
|
|
def _emit_ui(self, action, text, color):
|
|
if self.user_interface_data is not None:
|
|
self.user_interface_data(action, text, color)
|
|
|
|
@property
|
|
def read_port_buffer(self):
|
|
if len(self.port_buffer) <= 0:
|
|
return b""
|
|
data = bytes(self.port_buffer)
|
|
del self.port_buffer[:]
|
|
return data
|
|
|
|
@property
|
|
def received_data_length(self):
|
|
return len(self.port_buffer)
|
|
|
|
@staticmethod
|
|
def find_com_ports():
|
|
ComPorts.my_port_names = sorted(p.device for p in list_ports.comports())
|
|
if len(ComPorts.my_port_names) <= 0:
|
|
ComPorts.com_port_exists = False
|
|
else:
|
|
ComPorts.com_port_exists = True
|
|
return ComPorts.my_port_names
|
|
|
|
def _display_exception(self, module_name, ex):
|
|
text = "Exception: {} Module: {}. Method: {}".format(ex, module_name, getattr(ex, "__class__").__name__)
|
|
self._emit_ui("DisplayStatus", text, "red")
|
|
|
|
def close_com_port(self, port_to_close):
|
|
try:
|
|
self._emit_ui("SisplayStatus", "", "black")
|
|
if port_to_close is not None and port_to_close.is_open:
|
|
port_to_close.close()
|
|
self._emit_ui("DisplayCurrentSettings", "", "black")
|
|
self.port_open = False
|
|
except (OSError, serial.SerialException) as ex:
|
|
self.parameter_changed = True
|
|
self.port_changed = True
|
|
self._display_exception("ComPorts.vb", ex)
|
|
|
|
def data_received(self):
|
|
try:
|
|
if self.selected_port.in_waiting > 0:
|
|
data = self.selected_port.read(self.selected_port.in_waiting)
|
|
self.port_buffer.extend(data)
|
|
except Exception as ex:
|
|
self._display_exception("ComPorts.vb", ex)
|
|
|
|
@staticmethod
|
|
def error_received(event_type):
|
|
messages = {
|
|
"rxover": "Input buffer overflow",
|
|
"overrun": "Character buffer overrun",
|
|
"rxparity": "Parity error",
|
|
"frame": "Framing error",
|
|
"txfull": "Output buffer full",
|
|
}
|
|
message = messages.get(str(event_type).lower())
|
|
if message:
|
|
print(message)
|
|
|
|
def open_com_port(self):
|
|
flag = False
|
|
try:
|
|
if ComPorts.com_port_exists:
|
|
if self.previous_port.is_open:
|
|
self.close_com_port(self.previous_port)
|
|
if (not self.selected_port.is_open) or self.port_changed:
|
|
self.selected_port.open()
|
|
if self.selected_port.is_open:
|
|
self.selected_port.timeout = 5
|
|
self._emit_ui("DisplayCurrentSettings", "", "black")
|
|
self._emit_ui("DisplayStatus", "", "black")
|
|
flag = True
|
|
self.port_changed = False
|
|
except (OSError, serial.SerialException) as ex:
|
|
self.parameter_changed = True
|
|
self.port_changed = True
|
|
self._display_exception("ComPorts.vb", ex)
|
|
self.port_open = flag
|
|
return flag
|
|
|
|
def write_to_com_port(self, data_to_write):
|
|
flag = False
|
|
try:
|
|
if (self.selected_port is None or not self.selected_port.is_open) or self.port_changed:
|
|
self.port_open = self.open_com_port()
|
|
if self.selected_port.is_open:
|
|
self.selected_port.write(data_to_write)
|
|
flag = True
|
|
except serial.SerialTimeoutException as ex:
|
|
self._display_exception("ComPorts.vb", ex)
|
|
except (OSError, serial.SerialException) as ex:
|
|
self._display_exception("ComPorts.vb", ex)
|
|
if isinstance(ex, PermissionError):
|
|
self.close_com_port(self.selected_port)
|
|
self.parameter_changed = True
|
|
self._emit_ui("DisplayCurrentSettings", "", "black")
|
|
return flag
|