/usr/lib/python3/dist-packages/serial/threaded/__init__.py is in python3-serial 3.0.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 | #!/usr/bin/env python3
#
# Working with threading and pySerial
#
# This file is part of pySerial. https://github.com/pyserial/pyserial
# (C) 2015 Chris Liechti <cliechti@gmx.net>
#
# SPDX-License-Identifier: BSD-3-Clause
"""\
Support threading with serial ports.
"""
import serial
import threading
class Protocol(object):
"""\
Protocol as used by the ReaderThread. This base class provides empty
implementations of all methods.
"""
def connection_made(self, transport):
"""Called when reader thread is started"""
def data_received(self, data):
"""Called with snippets received from the serial port"""
def connection_lost(self, exc):
"""\
Called when the serial port is closed or the reader loop terminated
otherwise.
"""
class Packetizer(Protocol):
"""
Read binary packets from serial port. Packets are expected to be terminated
with a TERMINATOR byte (null byte by default).
The class also keeps track of the transport.
"""
TERMINATOR = b'\0'
def __init__(self):
self.buffer = bytearray()
self.transport = None
def connection_made(self, transport):
"""Store transport"""
self.transport = transport
def connection_lost(self, exc):
"""Forget transport"""
self.transport = None
def data_received(self, data):
"""Buffer received data, find TERMINATOR, call handle_packet"""
self.buffer.extend(data)
while self.TERMINATOR in self.buffer:
packet, self.buffer = self.buffer.split(self.TERMINATOR)
self.handle_packet(packet)
def handle_packet(self, packet):
"""Process packets - to be overridden by subclassing"""
raise NotImplementedError('please implement functionality in handle_packet')
class LineReader(Packetizer):
"""
Read and write (Unicode) lines from/to serial port.
The encoding is applied.
"""
TERMINATOR = b'\r\n'
ENCODING = 'utf-8'
UNICODE_HANDLING = 'replace'
def handle_packet(self, packet):
self.handle_line(packet.decode(self.ENCODING, self.UNICODE_HANDLING))
def handle_line(self, line):
"""Process one line - to be overridden by subclassing"""
raise NotImplementedError('please implement functionality in handle_line')
def write_line(self, text):
"""
Write text to the transport. ``text`` is a Unicode string and the encoding
is applied before sending ans also the newline is append.
"""
# + is not the best choice but bytes does not support % or .format in py3 and we want a single write call
self.transport.write(text.encode(self.ENCODING, self.UNICODE_HANDLING) + self.TERMINATOR)
class ReaderThread(threading.Thread):
"""\
Implement a serial port read loop and dispatch to a Protocol instance (like
the asyncio.Protocol) but do it with threads.
Calls to close() will close the serial port but it is also possible to just
stop() this thread and continue the serial port instance otherwise.
"""
def __init__(self, serial_instance, protocol_factory):
"""\
Initialize thread.
Note that the serial_instance' timeout is set to one second!
Other settings are not changed.
"""
super(ReaderThread, self).__init__()
self.daemon = True
self.serial = serial_instance
self.protocol_factory = protocol_factory
self.alive = True
self._lock = threading.Lock()
self._connection_made = threading.Event()
self.protocol = None
def stop(self):
"""Stop the reader thread"""
self.alive = False
self.join(2)
def run(self):
"""Reader loop"""
self.serial.timeout = 1
self.protocol = self.protocol_factory()
try:
self.protocol.connection_made(self)
except Exception as e:
self.alive = False
self.protocol.connection_lost(e)
self._connection_made.set()
return
error = None
self._connection_made.set()
while self.alive and self.serial.is_open:
try:
# read all that is there or wait for one byte (blocking)
data = self.serial.read(self.serial.in_waiting or 1)
except serial.SerialException as e:
# probably some I/O problem such as disconnected USB serial
# adapters -> exit
error = e
break
else:
if data:
# make a separated try-except for called used code
try:
self.protocol.data_received(data)
except Exception as e:
error = e
break
self.alive = False
self.protocol.connection_lost(error)
self.protocol = None
def write(self, data):
"""Thread safe writing (uses lock)"""
with self._lock:
self.serial.write(data)
def close(self):
"""Close the serial port and exit reader thread (uses lock)"""
# use the lock to let other threads finish writing
with self._lock:
# first stop reading, so that closing can be done on idle port
self.stop()
self.serial.close()
def connect(self):
"""
Wait until connection is set up and return the transport and protocol
instances.
"""
if self.alive:
self._connection_made.wait()
if not self.alive:
raise RuntimeError('connection_lost already called')
return (self, self.protocol)
else:
raise RuntimeError('already stopped')
# - - context manager, returns protocol
def __enter__(self):
"""\
Enter context handler. May raise RuntimeError in case the connection
could not be created.
"""
self.start()
self._connection_made.wait()
if not self.alive:
raise RuntimeError('connection_lost already called')
return self.protocol
def __exit__(self, exc_type, exc_val, exc_tb):
"""Leave context: close port"""
self.close()
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# test
if __name__ == '__main__':
import sys
import time
import traceback
class PrintLines(LineReader):
def connection_made(self, transport):
super(PrintLines, self).connection_made(transport)
sys.stdout.write('port opened\n')
self.write_line('hello world')
def handle_line(self, data):
sys.stdout.write('line received: {}\n'.format(repr(data)))
def connection_lost(self, exc):
if exc:
traceback.print_exc(exc)
sys.stdout.write('port closed\n')
ser = serial.serial_for_url('loop://', baudrate=115200, timeout=1)
with ReaderThread(ser, PrintLines) as protocol:
protocol.write_line('hello')
time.sleep(2)
# alternative usage
ser = serial.serial_for_url('loop://', baudrate=115200, timeout=1)
t = ReaderThread(ser, PrintLines)
t.start()
transport, protocol = t.connect()
protocol.write_line('hello')
time.sleep(2)
t.close()
|