def __init__(self,port=None,logFileName=False):
print("\n### TENMA MULTIMETER INTERFACE ###")
devList=self.device_list()
self.port=port
self.logFileName=logFileName
if self.logFileName:
self.log_init()
if port is None:
self.port=self.guessPort
if not self.port in devList:
print("!!! %s isn't in the list of serial ports !!!"%self.port)
print("Preparing to use",self.port)
self.ser = serial.Serial()
self.ser.port = self.port
self.ser.baudrate = 19230
self.ser.bytesize = serial.SEVENBITS
self.ser.parity = serial.PARITY_ODD
self.ser.stopbits = serial.STOPBITS_ONE
self.ser.timeout = 3
self.ser.xonxoff = False
self.ser.rtscts = False
self.ser.dsrdtr = False
self.ser.writeTimeout = 3
python类SEVENBITS的实例源码
def get_dsmr_connection_parameters():
""" Returns the communication settings required for the DSMR version set. """
DSMR_VERSION_MAPPING = {
DataloggerSettings.DSMR_VERSION_2: {
'baudrate': 9600,
'bytesize': serial.SEVENBITS,
'parity': serial.PARITY_EVEN,
'crc': False,
},
DataloggerSettings.DSMR_VERSION_4_PLUS: {
'baudrate': 115200,
'bytesize': serial.EIGHTBITS,
'parity': serial.PARITY_NONE,
'crc': True,
},
}
datalogger_settings = DataloggerSettings.get_solo()
connection_parameters = DSMR_VERSION_MAPPING[datalogger_settings.dsmr_version]
connection_parameters['com_port'] = datalogger_settings.com_port
return connection_parameters
def test_init_sets_the_correct_attrs(self):
"""__init__ sets the fields that get_settings reads"""
for setting, value in (
('baudrate', 57600),
('timeout', 7),
('write_timeout', 12),
('inter_byte_timeout', 15),
('stopbits', serial.STOPBITS_TWO),
('bytesize', serial.SEVENBITS),
('parity', serial.PARITY_ODD),
('xonxoff', True),
('rtscts', True),
('dsrdtr', True)):
kwargs = {'do_not_open': True, setting: value}
ser = serial.serial_for_url(PORT, **kwargs)
d = ser.get_settings()
self.assertEqual(getattr(ser, setting), value)
self.assertEqual(d[setting], value)
def log(options):
# configure the serial connections (the parameters differs on the device you are connecting to)
ser = serial.Serial(
port=options.device,
baudrate=9600,
parity=serial.PARITY_ODD,
stopbits=serial.STOPBITS_TWO,
bytesize=serial.SEVENBITS
)
prog = Progress()
with open(options.outfile, 'w') as the_file:
the_file.write('timestamp;value\n')
while True:
# \r\n is for device terminators set to CR LF
ser.write((':FETCh?\r\n'))
# wait one second before reading output.
time.sleep(options.interval)
out = ''
while ser.inWaiting() > 0:
out += ser.read(1)
if out != '':
out = out.rstrip()
res = "%s;%s\n" % (time.time(), float(out))
the_file.write(res)
the_file.flush()
prog.show()
def __init__(self,
path,
baudrate=9600,
timeout=1,
open_on_create=True,
debugging=False):
super(TeliumNativeSerial, self).__init__(
path,
baudrate=baudrate,
bytesize=SEVENBITS,
parity=PARITY_EVEN,
stopbits=STOPBITS_ONE,
timeout=timeout,
open_on_create=open_on_create,
debugging=debugging)
def connect(
self, baudrate=None, parity=None, databits=None, stopbits=None):
"""Open a serial port for communication with barcode reader device
Connection settings are taken from three possible locations in the
following order:
- method arguments
- object properties set directly or in constructor or
detect_connection_settings()
- default device settings according to device documentation
If connection settings (baud rate, parity, etc.) are neither provided
as arguments nor previously set as object properties, for example by
a search with the detect_connection_settings() method, the default
values specified in the device documentation are used. For example,
page 3-1 of the MS3 user manual specifies:
- baud rate: 9600
- parity: even
- data bits: 7
- stop bits: 1
- flow control: none
"""
baudrate = baudrate or self.baudrate or 9600
parity = parity or self.parity or serial.PARITY_EVEN
bytesize = databits or self.databits or serial.SEVENBITS
stopbits = stopbits or self.stopbits or serial.STOPBITS_ONE
self.port = serial.Serial(
self.portname,
baudrate=baudrate,
parity=parity,
bytesize=bytesize,
stopbits=stopbits,
timeout=1,
xonxoff=False,
rtscts=False,
dsrdtr=False,
)
self._config = self.read_config()
def __init__(self, tty, timeout=2):
self.conn = serial.Serial(tty,
baudrate=19200,
bytesize=serial.SEVENBITS,
parity=serial.PARITY_ODD,
stopbits=serial.STOPBITS_ONE,
timeout=timeout,
xonxoff=1,
rtscts=0,
dsrdtr=0
)
# switch on power for hardware RS232
# transmitter via handshake-signals
self.conn.dtr = True
self.conn.rts = False
def test_dsmr_version_2(self):
""" Test connection parameters for DSMR v2. """
datalogger_settings = DataloggerSettings.get_solo()
datalogger_settings.dsmr_version = DataloggerSettings.DSMR_VERSION_2
datalogger_settings.save()
self.assertEqual(DataloggerSettings.get_solo().dsmr_version, DataloggerSettings.DSMR_VERSION_2)
connection_parameters = dsmr_datalogger.services.get_dsmr_connection_parameters()
self.assertEqual(connection_parameters['baudrate'], 9600)
self.assertEqual(connection_parameters['bytesize'], serial.SEVENBITS)
self.assertEqual(connection_parameters['parity'], serial.PARITY_EVEN)
def __init__(self,
port: str,
baudrate: int,
bytesize: int,
stopbits: float,
parity: str,
timeout: float,
xonxoff: bool,
rtscts: bool):
"""Converts data from JSON format to serial.Serial format."""
self._port = port
self._baudrate = baudrate
self._bytesize = {
5: serial.FIVEBITS,
6: serial.SIXBITS,
7: serial.SEVENBITS,
8: serial.EIGHTBITS
}[bytesize]
self._stopbits = {
1: serial.STOPBITS_ONE,
1.5: serial.STOPBITS_ONE_POINT_FIVE,
2: serial.STOPBITS_TWO
}[stopbits]
self._parity = {
'none': serial.PARITY_NONE,
'even': serial.PARITY_EVEN,
'odd': serial.PARITY_ODD,
'mark': serial.PARITY_MARK,
'space': serial.PARITY_SPACE
}[parity]
self._timeout = timeout
self._xonxoff = xonxoff
self._rtscts = rtscts
def openSerial(self):
#Set up the relationship between what the user enters and what the API calls for
bytedic = {'5':serial.FIVEBITS,
'6':serial.SIXBITS,
'7':serial.SEVENBITS,
'8':serial.EIGHTBITS}
bytesize = bytedic[str(self.root.variables['databits'])]
paritydict = {'None':serial.PARITY_NONE,
'Even':serial.PARITY_EVEN,
'Odd' :serial.PARITY_ODD,
'Mark':serial.PARITY_MARK,
'Space':serial.PARITY_SPACE}
parity=paritydict[self.root.variables['parity']]
stopbitsdict = {'1':serial.STOPBITS_ONE,
'2':serial.STOPBITS_TWO}
stopbits = stopbitsdict[str(self.root.variables['stopbits'])]
#Open the serial port given the settings, store under the root
if os.name == 'nt':
port = self.root.variables['COMport'][0:5].strip()
self.root.ser = serial.Serial(\
port=port,\
baudrate=str(self.root.variables['baud']),\
bytesize=bytesize, parity=parity, stopbits=stopbits, timeout=0.5)
else:
first_space = self.root.variables['COMport'].index(' ')
port = self.root.variables['COMport'][0:first_space].strip()
# Parameters necessary due to https://github.com/pyserial/pyserial/issues/59
self.root.ser = serial.Serial(\
port=port,\
baudrate=str(self.root.variables['baud']),\
bytesize=bytesize, parity=parity, stopbits=stopbits, timeout=0.5, rtscts=True, dsrdtr=True)
io.DEFAULT_BUFFER_SIZE = 5000
#Purge the buffer of any previous data
if float(serial.VERSION[0:3]) < 3:
#If we're executing with pySerial 2.x
serial.Serial.flushInput(self.root.ser)
serial.Serial.flushOutput(self.root.ser)
else:
#Otherwise we're using pySerial 3.x
serial.Serial.reset_input_buffer(self.root.ser)
serial.Serial.reset_output_buffer(self.root.ser)
def predict_result():
ser = serial.Serial( # ????????????
port='COM6',
baudrate=1200,
parity=serial.PARITY_ODD,
stopbits=serial.STOPBITS_TWO,
bytesize=serial.SEVENBITS
)
serial_data = []
plt.xlim(0, 100)
plt.ylim(300, 700)
plt.title('GSR')
plt.ion()
i = 0
j = 0
id = 0
while True:
line = ser.readline()
line = int(line)
serial_data.append(line)
if i > 100:
plt.xlim(i - 100, i)
plt.plot(serial_data)
i += 1
j += 1
if j >= 50:
clf = joblib.load('model\\happy_model.m')
select = joblib.load('model\\vector_select.m')
vector = getattr.get_vector(serial_data)
new_vector = select.transform(vector)
print(new_vector)
result = clf.predict(new_vector)
if result[0] == '2':
clf = joblib.load('model\\sad_model.m')
result = clf.predict(new_vector)
j = 0
plt.plot([i, i], [300, 700], 'r--')
if result[0] == '1':
plt.annotate('happy', xy=(i, 600), xytext=(i - 10, 600), arrowprops=dict(facecolor='red', shrink=0.05))
res = 1
database.insert(id, res)
elif result[0] == '2':
plt.annotate('normal', xy=(i, 600), xytext=(i - 10, 600), arrowprops=dict(facecolor='blue', shrink=0.05))
res = 0
database.insert(id, res)
else:
plt.annotate('sad', xy=(i, 600), xytext=(i - 10, 600),arrowprops=dict(facecolor='black', shrink=0.05))
res = 2
database.insert(id, res)
print(result)
id += 1
plt.pause(0.001)
