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
|
#! python
#
# Python Serial Port Extension for Win32, Linux, BSD, Jython
# see __init__.py
#
# This module implements a URL dummy handler for serial_for_url.
#
# (C) 2011 Chris Liechti <cliechti@gmx.net>
# this is distributed under a free software license, see license.txt
#
# URL format: test://
from serial.serialutil import *
import time
import socket
import logging
# map log level names to constants. used in fromURL()
LOGGER_LEVELS = {
'debug': logging.DEBUG,
'info': logging.INFO,
'warning': logging.WARNING,
'error': logging.ERROR,
}
class DummySerial(SerialBase):
"""Serial port implementation for plain sockets."""
def open(self):
"""Open port with current settings. This may throw a SerialException
if the port cannot be opened."""
self.logger = None
if self._port is None:
raise SerialException("Port must be configured before it can be used.")
# not that there anything to configure...
self._reconfigurePort()
# all things set up get, now a clean start
self._isOpen = True
def _reconfigurePort(self):
"""Set communication parameters on opened port. for the test://
protocol all settings are ignored!"""
if self.logger:
self.logger.info('ignored port configuration change')
def close(self):
"""Close port"""
if self._isOpen:
self._isOpen = False
def makeDeviceName(self, port):
raise SerialException("there is no sensible way to turn numbers into URLs")
def fromURL(self, url):
"""extract host and port from an URL string"""
if url.lower().startswith("test://"): url = url[7:]
try:
# is there a "path" (our options)?
if '/' in url:
# cut away options
url, options = url.split('/', 1)
# process options now, directly altering self
for option in options.split('/'):
if '=' in option:
option, value = option.split('=', 1)
else:
value = None
if option == 'logging':
logging.basicConfig() # XXX is that good to call it here?
self.logger = logging.getLogger('pySerial.test')
self.logger.setLevel(LOGGER_LEVELS[value])
self.logger.debug('enabled logging')
else:
raise ValueError('unknown option: {!r}'.format(option))
except ValueError as e:
raise SerialException('expected a string in the form "[test://][option[/option...]]": {}'.format(e))
return (host, port)
# - - - - - - - - - - - - - - - - - - - - - - - -
def inWaiting(self):
"""Return the number of characters currently in the input buffer."""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
# set this one to debug as the function could be called often...
self.logger.debug('WARNING: inWaiting returns dummy value')
return 0 # hmmm, see comment in read()
def read(self, size=1):
"""Read size bytes from the serial port. If a timeout is set it may
return less characters as requested. With no timeout it will block
until the requested number of bytes is read."""
if not self._isOpen: raise PortNotOpenError()
data = '123' # dummy data
return bytes(data)
def write(self, data):
"""Output the given string over the serial port. Can block if the
connection is blocked. May raise SerialException if the connection is
closed."""
if not self._isOpen: raise PortNotOpenError()
# nothing done
return len(data)
def flushInput(self):
"""Clear input buffer, discarding all that is in the buffer."""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored flushInput')
def flushOutput(self):
"""Clear output buffer, aborting the current output and
discarding all that is in the buffer."""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored flushOutput')
def sendBreak(self, duration=0.25):
"""Send break condition. Timed, returns to idle state after given
duration."""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored sendBreak({!r})'.format(duration))
def setBreak(self, level=True):
"""Set break: Controls TXD. When active, to transmitting is
possible."""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored setBreak({!r})'.format(level))
def setRTS(self, level=True):
"""Set terminal status line: Request To Send"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored setRTS({!r})'.format(level))
def setDTR(self, level=True):
"""Set terminal status line: Data Terminal Ready"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('ignored setDTR({!r})'.format(level))
def getCTS(self):
"""Read terminal status line: Clear To Send"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('returning dummy for getCTS()')
return True
def getDSR(self):
"""Read terminal status line: Data Set Ready"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('returning dummy for getDSR()')
return True
def getRI(self):
"""Read terminal status line: Ring Indicator"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('returning dummy for getRI()')
return False
def getCD(self):
"""Read terminal status line: Carrier Detect"""
if not self._isOpen: raise PortNotOpenError()
if self.logger:
self.logger.info('returning dummy for getCD()')
return True
# - - - platform specific - - -
# None so far
# assemble Serial class with the platform specific implementation and the base
# for file-like behavior. for Python 2.6 and newer, that provide the new I/O
# library, derive from io.RawIOBase
try:
import io
except ImportError:
# classic version with our own file-like emulation
class Serial(DummySerial, FileLike):
pass
else:
# io library present
class Serial(DummySerial, io.RawIOBase):
pass
# simple client test
if __name__ == '__main__':
import sys
s = Serial('test://logging=debug')
sys.stdout.write('{}\n'.format(s))
sys.stdout.write("write...\n")
s.write("hello\n")
s.flush()
sys.stdout.write("read: {}\n".format(s.read(5)))
s.close()
|
