<s> import logging <EOL> import numbers <EOL> import time <EOL> try : <EOL> import urlparse <EOL> except ImportError : <EOL> import urllib . parse as urlparse <EOL> try : <EOL> import queue <EOL> except ImportError : <EOL> import Queue as queue <EOL> from serial . serialutil import SerialBase , SerialException , to_bytes , iterbytes , writeTimeoutError , portNotOpenError <EOL> LOGGER_LEVELS = { <EOL> '<STR_LIT>' : logging . DEBUG , <EOL> '<STR_LIT:info>' : logging . INFO , <EOL> '<STR_LIT>' : logging . WARNING , <EOL> '<STR_LIT:error>' : logging . ERROR , <EOL> } <EOL> class Serial ( SerialBase ) : <EOL> """<STR_LIT>""" <EOL> BAUDRATES = ( <NUM_LIT:50> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:200> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <EOL> <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( Serial , self ) . __init__ ( * args , ** kwargs ) <EOL> self . buffer_size = <NUM_LIT> <EOL> self . queue = None <EOL> self . logger = None <EOL> def open ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . is_open : <EOL> raise SerialException ( "<STR_LIT>" ) <EOL> self . logger = None <EOL> self . queue = queue . Queue ( self . buffer_size ) <EOL> if self . _port is None : <EOL> raise SerialException ( "<STR_LIT>" ) <EOL> self . from_url ( self . port ) <EOL> self . _reconfigure_port ( ) <EOL> self . is_open = True <EOL> if not self . _dsrdtr : <EOL> self . _update_dtr_state ( ) <EOL> if not self . _rtscts : <EOL> self . _update_rts_state ( ) <EOL> self . reset_input_buffer ( ) <EOL> self . reset_output_buffer ( ) <EOL> def close ( self ) : <EOL> if self . is_open : <EOL> self . is_open = False <EOL> try : <EOL> self . queue . put_nowait ( None ) <EOL> except queue . Full : <EOL> pass <EOL> super ( Serial , self ) . close ( ) <EOL> def _reconfigure_port ( self ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( self . _baudrate , numbers . Integral ) or not <NUM_LIT:0> < self . _baudrate < <NUM_LIT:2> ** <NUM_LIT:32> : <EOL> raise ValueError ( "<STR_LIT>" % ( self . _baudrate ) ) <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> def from_url ( self , url ) : <EOL> """<STR_LIT>""" <EOL> parts = urlparse . urlsplit ( url ) <EOL> if parts . scheme != "<STR_LIT>" : <EOL> raise SerialException ( '<STR_LIT>' % ( parts . scheme , ) ) <EOL> try : <EOL> for option , values in urlparse . parse_qs ( parts . query , True ) . items ( ) : <EOL> if option == '<STR_LIT>' : <EOL> logging . basicConfig ( ) <EOL> self . logger = logging . getLogger ( '<STR_LIT>' ) <EOL> self . logger . setLevel ( LOGGER_LEVELS [ values [ <NUM_LIT:0> ] ] ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % ( option , ) ) <EOL> except ValueError as e : <EOL> raise SerialException ( '<STR_LIT>' % e ) <EOL> @ property <EOL> def in_waiting ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . debug ( '<STR_LIT>' % ( self . queue . qsize ( ) , ) ) <EOL> return self . queue . qsize ( ) <EOL> def read ( self , size = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . _timeout is not None and self . _timeout != <NUM_LIT:0> : <EOL> timeout = time . time ( ) + self . _timeout <EOL> else : <EOL> timeout = None <EOL> data = bytearray ( ) <EOL> while size > <NUM_LIT:0> and self . is_open : <EOL> try : <EOL> b = self . queue . get ( timeout = self . _timeout ) <EOL> except queue . Empty : <EOL> if self . _timeout == <NUM_LIT:0> : <EOL> break <EOL> else : <EOL> if data is not None : <EOL> data += b <EOL> size -= <NUM_LIT:1> <EOL> else : <EOL> break <EOL> if timeout and time . time ( ) > timeout : <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> break <EOL> return bytes ( data ) <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> data = to_bytes ( data ) <EOL> time_used_to_send = <NUM_LIT> * len ( data ) / self . _baudrate <EOL> if self . _write_timeout is not None and time_used_to_send > self . _write_timeout : <EOL> time . sleep ( self . _write_timeout ) <EOL> raise writeTimeoutError <EOL> for byte in iterbytes ( data ) : <EOL> self . queue . put ( byte , timeout = self . _write_timeout ) <EOL> return len ( data ) <EOL> def reset_input_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> try : <EOL> while self . queue . qsize ( ) : <EOL> self . queue . get_nowait ( ) <EOL> except queue . Empty : <EOL> pass <EOL> def reset_output_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> try : <EOL> while self . queue . qsize ( ) : <EOL> self . queue . get_nowait ( ) <EOL> except queue . Empty : <EOL> pass <EOL> def _update_break_state ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' % ( self . _break_state , ) ) <EOL> def _update_rts_state ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' % ( self . _rts_state , ) ) <EOL> def _update_dtr_state ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' % ( self . _dtr_state , ) ) <EOL> @ property <EOL> def cts ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' % ( self . _rts_state , ) ) <EOL> return self . _rts_state <EOL> @ property <EOL> def dsr ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' % ( self . _dtr_state , ) ) <EOL> return self . _dtr_state <EOL> @ property <EOL> def ri ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> return False <EOL> @ property <EOL> def cd ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise portNotOpenError <EOL> if self . logger : <EOL> self . logger . info ( '<STR_LIT>' ) <EOL> return True <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import sys <EOL> s = Serial ( '<STR_LIT>' ) <EOL> sys . stdout . write ( '<STR_LIT>' % s ) <EOL> sys . stdout . write ( "<STR_LIT>" ) <EOL> s . write ( "<STR_LIT>" ) <EOL> s . flush ( ) <EOL> sys . stdout . write ( "<STR_LIT>" % s . read ( <NUM_LIT:5> ) ) <EOL> s . close ( ) </s>
