def __init__(self, serial_name):
""" Initialize the power supply.
Each children an here define non-default values for his specific case
Input : serial_name, String, is the serial port name (e.g. COM2)
"""
self.name = "Generic Power Supply"
self.port = serial_name
self.baudrate = 9600 # Default baud rate
""" Possible baudrate values :
50, 75, 110, 134, 150, 200, 300, 600, 1200, 1800, 2400, 4800, 9600, 19200, 38400, 57600, 115200,
230400, 460800, 500000, 576000, 921600, 1000000, 1152000, 1500000, 2000000, 2500000, 3000000, 3500000, 4000000
"""
self.timeout = 1 # Default timeout, seconds
self.parity = serial.PARITY_NONE # Default parity
""" Possible parities :
PARITY_NONE, PARITY_EVEN, PARITY_ODD, PARITY_MARK, PARITY_SPACE
"""
self.stopbits = serial.STOPBITS_ONE # Default stop bits
""" Possible stopbits :
STOPBITS_ONE, STOPBITS_ONE_POINT_FIVE, STOPBITS_TWO
"""
self.bytesize = serial.EIGHTBITS
""" Possible bytesizes :
FIVEBITS, SIXBITS, SEVENBITS, EIGHTBITS
"""
self.ser = serial.Serial(self.port, self.baudrate, self.bytesize, self.parity, self.stopbits, timeout=self.timeout) # serial port
self.max_voltage = 0.0 # Volts
self.max_current = 0.0 # Amps
self.max_power = 0.0 # Watts
python类EIGHTBITS的实例源码
def __init__(self, serial_name):
self.port = serial_name
self.baudrate = 57600 # Default baud rate
self.timeout = 0.1 # Default timeout, seconds
self.parity = serial.PARITY_ODD # Default parity
self.stopbits = serial.STOPBITS_ONE # Default stop bits
self.bytesize = serial.EIGHTBITS
self.ser = serial.Serial(self.port, self.baudrate, self.bytesize, self.parity, self.stopbits, timeout=self.timeout) # serial port
self.max_voltage = 360.0 # Volts
self.max_current = 10.0 # Amps
self.max_power = 1000.0 # Watts
def connect(self):
"""
Open tty connection to sensor
"""
if self.link is not None:
self.disconnect()
self.link = serial.Serial(self.port, 9600, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE, dsrdtr=True, timeout=5, interCharTimeout=0.1)
def __init__(self, device='/dev/ttyAMA0', baud=115200,
parity_=serial.PARITY_NONE, stop_bits=serial.STOPBITS_ONE, byte_size=serial.EIGHTBITS, time_out=0):
# ACK header for data recive
self.Header = bytearray([0x00, 0x00, 0xff, 0x00, 0xff, 0x00, 0x00, 0x00, 0xff])
# time out.
if time_out != 0:
self.connectUART = serial.Serial(port=device, baudrate=baud, stopbits=stop_bits, bytesize=byte_size,
timeout=time_out)
else:
self.connectUART = serial.Serial(port=device, baudrate=baud, stopbits=stop_bits, bytesize=byte_size)
# manually close connection
def __init__(self, device='/dev/ttyAMA0', baud=115200,
parity_=serial.PARITY_NONE, stop_bits=serial.STOPBITS_ONE, byte_size=serial.EIGHTBITS, time_out=0):
super(PN532, self).__init__(device, baud, parity_, stop_bits, byte_size, time_out)
def __init__(self, port='/dev/serial0'):
self.ser = serial.Serial()
self.ser.port = port
self.ser.baudrate = 115200
self.ser.bytesize = serial.EIGHTBITS
self.ser.parity = serial.PARITY_NONE
self.ser.stopbits = serial.STOPBITS_ONE
self.open_serial()
self.rc_data = [0] * N_CHAN
def __init__(self, port="/dev/ttyACM0", bitrate=115200,
rx_protocol=msp.RX_MSP, firmware=msp.FIRMWARE_BF):
assert rx_protocol in msp.RX_OPTIONS, (
"unsupported rx protocol indicated")
self.rx_protocol = rx_protocol
self.rx_protocol_ch_count = len(
MSP_PAYLOAD_FMT[msp.MSP_RC][rx_protocol][1:])
assert firmware in msp.FIRMWARE_OPTIONS, (
"unsupported firmware indicated")
self.firmware = firmware
self.firmware_motor_count = len(
MSP_PAYLOAD_FMT[msp.MSP_MOTOR][firmware][1:])
self.ser = serial.Serial()
self.ser.port = port
self.ser.baudrate = bitrate
self.ser.bytesize = serial.EIGHTBITS
self.ser.parity = serial.PARITY_NONE
self.ser.stopbits = serial.STOPBITS_ONE
self.ser.timeout = None
self.ser.xonxoff = False
self.ser.rtscts = False
self.ser.dsrdtr = False
self.ser.writeTimeout = 2
self.open_serial()
def MakeSerialDevice(port="/dev/ttyUSB0"):
dev = serial.Serial(
port=port,
baudrate=115200,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS,
# blocking
timeout=5)
# dev.open()
return dev
def __init__(self,
serialdevice,
configuration,
power_control=DTR_ON,
scan_interval=0):
"""Initialize the data collector based on the given parameters."""
self.record_length = configuration[RECORD_LENGTH]
self.start_sequence = configuration[STARTBLOCK]
self.byte_order = configuration[BYTE_ORDER]
self.multiplier = configuration[MULTIPLIER]
self.timeout = configuration[TIMEOUT]
self.scan_interval = scan_interval
self.listeners = []
self.power_control = power_control
self.sensordata = {}
self.config = configuration
self.data = None
self.last_poll = None
self.start_func = None
self.stop_func = None
self.ser = serial.Serial(port=serialdevice,
baudrate=configuration[BAUD_RATE],
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS)
# Update date in using a background thread
if self.scan_interval > 0:
thread = threading.Thread(target=self.refresh, args=())
thread.daemon = True
thread.start()
def read_mh_z19_with_temperature(serial_device):
""" Read the CO2 PPM concenration and temperature from a MH-Z19 sensor"""
logger = logging.getLogger(__name__)
ser = serial.Serial(port=serial_device,
baudrate=9600,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS)
sbuf = bytearray()
starttime = time.time()
finished = False
timeout = 2
res = None
ser.write(MZH19_READ)
while not finished:
mytime = time.time()
if mytime - starttime > timeout:
logger.error("read timeout after %s seconds, read %s bytes",
timeout, len(sbuf))
return None
if ser.inWaiting() > 0:
sbuf += ser.read(1)
if len(sbuf) == MHZ19_SIZE:
# TODO: check checksum
res = (sbuf[2]*256 + sbuf[3], sbuf[4] - 40)
logger.debug("Finished reading data %s", sbuf)
finished = True
else:
time.sleep(.1)
logger.debug("Serial waiting for data, buffer length=%s",
len(sbuf))
return res
def __init__(self, protocol, deviceNameOrPortNumber, reactor,
baudrate = 9600, bytesize = EIGHTBITS, parity = PARITY_NONE,
stopbits = STOPBITS_ONE, xonxoff = 0, rtscts = 0):
self._serial = serial.Serial(deviceNameOrPortNumber, baudrate=baudrate,
bytesize=bytesize, parity=parity,
stopbits=stopbits, timeout=None,
xonxoff=xonxoff, rtscts=rtscts)
self.flushInput()
self.flushOutput()
self.reactor = reactor
self.protocol = protocol
self.outQueue = []
self.closed = 0
self.closedNotifies = 0
self.writeInProgress = 0
self.protocol = protocol
self._overlappedRead = win32file.OVERLAPPED()
self._overlappedRead.hEvent = win32event.CreateEvent(None, 1, 0, None)
self._overlappedWrite = win32file.OVERLAPPED()
self._overlappedWrite.hEvent = win32event.CreateEvent(None, 0, 0, None)
self.reactor.addEvent(self._overlappedRead.hEvent, self, 'serialReadEvent')
self.reactor.addEvent(self._overlappedWrite.hEvent, self, 'serialWriteEvent')
self.protocol.makeConnection(self)
flags, comstat = win32file.ClearCommError(self._serial.hComPort)
rc, self.read_buf = win32file.ReadFile(self._serial.hComPort,
win32file.AllocateReadBuffer(1),
self._overlappedRead)
def __init__(self, protocol, deviceNameOrPortNumber, reactor,
baudrate = 9600, bytesize = EIGHTBITS, parity = PARITY_NONE,
stopbits = STOPBITS_ONE, timeout = 0, xonxoff = 0, rtscts = 0):
abstract.FileDescriptor.__init__(self, reactor)
self._serial = serial.Serial(deviceNameOrPortNumber, baudrate = baudrate, bytesize = bytesize, parity = parity, stopbits = stopbits, timeout = timeout, xonxoff = xonxoff, rtscts = rtscts)
self.reactor = reactor
self.flushInput()
self.flushOutput()
self.protocol = protocol
self.protocol.makeConnection(self)
self.startReading()
def __init__(self, device='/dev/ttyUSB0', baudrate=9600, rst='-rts', debug=False):
self._sl = serial.Serial(
port = device,
parity = serial.PARITY_EVEN,
bytesize = serial.EIGHTBITS,
stopbits = serial.STOPBITS_TWO,
timeout = 1,
xonxoff = 0,
rtscts = 0,
baudrate = baudrate,
)
self._rst_pin = rst
self._debug = debug
def __init__(self,addr='COM1',ndacs=8,polarity=('BIP','BIP'),verb=True,timeout = 2,reset=False):
#Directly using pyserial interface and skipping pyvisa
self.serialport = serial.Serial(addr,baudrate=115200,bytesize=serial.EIGHTBITS, parity=serial.PARITY_ODD, stopbits=serial.STOPBITS_ONE,timeout=timeout)
if ndacs!=8 and ndacs!=16:
print('DAC WARNING, non-standard number of dacs. Should be 8 or 16 but %d was given' % ndacs)
self.ndacs = ndacs
self.verb = verb
self.SetPolarity(polarity)
if reset:
self.RampAllZero(tt=20.)
return
self.lastmessage = ()
#Function to set polarity. This just informs the driver what polarities are in use so it can correctly set the voltages.
#The driver cannot physically set the polarity. The real polarity of the DACs can only be set form the hardware switches.
ThreadlessPulseSpO2Plotter.py 文件源码
项目:CMS50DPlus-PulseOx
作者: jgerschler
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def connect(self):
if self.conn is None:
self.conn = serial.Serial(port = self.port,
baudrate = 19200,
parity = serial.PARITY_ODD,
stopbits = serial.STOPBITS_ONE,
bytesize = serial.EIGHTBITS,
timeout = 5,
xonxoff = 1)
elif not self.is_connected():
self.conn.open()
UNFINISHED CMS50DPlusPulseSpO2.py 文件源码
项目:CMS50DPlus-PulseOx
作者: jgerschler
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def connect(self):
if self.conn is None:
self.conn = serial.Serial(port = self.port,
baudrate = 19200,
parity = serial.PARITY_ODD,
stopbits = serial.STOPBITS_ONE,
bytesize = serial.EIGHTBITS,
timeout = 5,
xonxoff = 1)
elif not self.is_connected():
self.conn.open()
def connect(self):
if self.conn is None:
self.conn = serial.Serial(port = self.port,
baudrate = 19200,
parity = serial.PARITY_ODD,
stopbits = serial.STOPBITS_ONE,
bytesize = serial.EIGHTBITS,
timeout = 5,
xonxoff = 1)
elif not self.is_connected():
self.conn.open()
def connect(self):
if self.conn is None:
self.conn = serial.Serial(port = self.port,
baudrate = 19200,
parity = serial.PARITY_ODD,
stopbits = serial.STOPBITS_ONE,
bytesize = serial.EIGHTBITS,
timeout = 5,
xonxoff = 1)
elif not self.is_connected():
self.conn.open()
def __init__(self, port):
self.ser = serial.Serial(
port=port,
baudrate=baud_rate,
parity=serial.PARITY_ODD,
stopbits=serial.STOPBITS_ONE,
bytesize=serial.EIGHTBITS
)
self.ser.write(b'\xF6')
res = self.ser.read(2)
if res == b'e\xfb':
click.secho('nejemojo is go', fg='green')
else:
click.secho('bad mojo: {}'.format(res), fg='red')
def __init__(self, usb_pathname = '/dev/ttyUSB0', _read_timeout = 3.0):
# Open with Pyserial
self.read_timeout = _read_timeout
self.serial_port = serial.Serial(
port = usb_pathname,
baudrate = 2400,
parity = serial.PARITY_NONE,
stopbits = serial.STOPBITS_ONE,
bytesize = serial.EIGHTBITS,
timeout = self.read_timeout)