pymonoprice/__init__.py
def get_monoprice(port_url):
    """
    Return synchronous version of Monoprice interface
    :param port_url: serial port, i.e. '/dev/ttyUSB0'
    :return: synchronous implementation of Monoprice interface
    """

    lock = RLock()

    def synchronized(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            with lock:
                return func(*args, **kwargs)
        return wrapper

    class MonopriceSync(Monoprice):
        def __init__(self, port_url):
            self._port = serial.serial_for_url(port_url, do_not_open=True)
            self._port.baudrate = 9600
            self._port.stopbits = serial.STOPBITS_ONE
            self._port.bytesize = serial.EIGHTBITS
            self._port.parity = serial.PARITY_NONE
            self._port.timeout = TIMEOUT
            self._port.write_timeout = TIMEOUT
            self._port.open()

        def _process_request(self, request: bytes, skip=0):
            """
            :param request: request that is sent to the monoprice
            :param skip: number of bytes to skip for end of transmission decoding
            :return: ascii string returned by monoprice
            """
            _LOGGER.debug('Sending "%s"', request)
            # clear
            self._port.reset_output_buffer()
            self._port.reset_input_buffer()
            # send
            self._port.write(request)
            self._port.flush()
            # receive
            result = bytearray()
            while True:
                c = self._port.read(1)
                if not c:
                    raise serial.SerialTimeoutException(
                        'Connection timed out! Last received bytes {}'.format([hex(a) for a in result]))
                result += c
                if len(result) > skip and result[-LEN_EOL:] == EOL:
                    break
            ret = bytes(result)
            _LOGGER.debug('Received "%s"', ret)
            return ret.decode('ascii')

        @synchronized
        def zone_status(self, zone: int):
            # Ignore first 6 bytes as they will contain 3 byte command and 3 bytes of EOL
            return ZoneStatus.from_string(self._process_request(_format_zone_status_request(zone), skip=6))

        @synchronized
        def set_power(self, zone: int, power: bool):
            self._process_request(_format_set_power(zone, power))

        @synchronized
        def set_mute(self, zone: int, mute: bool):
            self._process_request(_format_set_mute(zone, mute))

        @synchronized
        def set_volume(self, zone: int, volume: int):
            self._process_request(_format_set_volume(zone, volume))

        @synchronized
        def set_treble(self, zone: int, treble: int):
            self._process_request(_format_set_treble(zone, treble))

        @synchronized
        def set_bass(self, zone: int, bass: int):
            self._process_request(_format_set_bass(zone, bass))

        @synchronized
        def set_balance(self, zone: int, balance: int):
            self._process_request(_format_set_balance(zone, balance))

        @synchronized
        def set_source(self, zone: int, source: int):
            self._process_request(_format_set_source(zone, source))

        @synchronized
        def restore_zone(self, status: ZoneStatus):
            self.set_power(status.zone, status.power)
            self.set_mute(status.zone, status.mute)
            self.set_volume(status.zone, status.volume)
            self.set_treble(status.zone, status.treble)
            self.set_bass(status.zone, status.bass)
            self.set_balance(status.zone, status.balance)
            self.set_source(status.zone, status.source)

    return MonopriceSync(port_url)
def get_monoprice(port_url):
    """
    Return synchronous version of Monoprice interface
    :param port_url: serial port, i.e. '/dev/ttyUSB0'
    :return: synchronous implementation of Monoprice interface
    """

    lock = RLock()

    def synchronized(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            with lock:
                return func(*args, **kwargs)
        return wrapper

    class MonopriceSync(Monoprice):
        def __init__(self, port_url):
            self._port = serial.serial_for_url(port_url, do_not_open=True)
            self._port.baudrate = 9600
            self._port.stopbits = serial.STOPBITS_ONE
            self._port.bytesize = serial.EIGHTBITS
            self._port.parity = serial.PARITY_NONE
            self._port.timeout = TIMEOUT
            self._port.write_timeout = TIMEOUT
            self._port.open()

        def _process_request(self, request: bytes, skip=0):
            """
            :param request: request that is sent to the monoprice
            :param skip: number of bytes to skip for end of transmission decoding
            :return: ascii string returned by monoprice
            """
            _LOGGER.debug('Sending "%s"', request)
            # clear
            self._port.reset_output_buffer()
            self._port.reset_input_buffer()
            # send
            self._port.write(request)
            self._port.flush()
            # receive
            result = bytearray()
            while True:
                c = self._port.read(1)
                if not c:
                    raise serial.SerialTimeoutException(
                        'Connection timed out! Last received bytes {}'.format([hex(a) for a in result]))
                result += c
                if len(result) > skip and result[-LEN_EOL:] == EOL:
                    break
            ret = bytes(result)
            _LOGGER.debug('Received "%s"', ret)
            return ret.decode('ascii')

        @synchronized
        def zone_status(self, zone: int):
            # Ignore first 6 bytes as they will contain 3 byte command and 3 bytes of EOL
            return ZoneStatus.from_string(self._process_request(_format_zone_status_request(zone), skip=6))

        @synchronized
        def set_power(self, zone: int, power: bool):
            self._process_request(_format_set_power(zone, power))

        @synchronized
        def set_mute(self, zone: int, mute: bool):
            self._process_request(_format_set_mute(zone, mute))

        @synchronized
        def set_volume(self, zone: int, volume: int):
            self._process_request(_format_set_volume(zone, volume))

        @synchronized
        def set_treble(self, zone: int, treble: int):
            self._process_request(_format_set_treble(zone, treble))

        @synchronized
        def set_bass(self, zone: int, bass: int):
            self._process_request(_format_set_bass(zone, bass))

        @synchronized
        def set_balance(self, zone: int, balance: int):
            self._process_request(_format_set_balance(zone, balance))

        @synchronized
        def set_source(self, zone: int, source: int):
            self._process_request(_format_set_source(zone, source))

        @synchronized
        def restore_zone(self, status: ZoneStatus):
            self.set_power(status.zone, status.power)
            self.set_mute(status.zone, status.mute)
            self.set_volume(status.zone, status.volume)
            self.set_treble(status.zone, status.treble)
            self.set_bass(status.zone, status.bass)
            self.set_balance(status.zone, status.balance)
            self.set_source(status.zone, status.source)

    return MonopriceSync(port_url)
def get_monoprice ( port_url ) : lock = RLock ( ) def synchronized ( func ) : @ wraps ( func ) def wrapper ( * args , * * kwargs ) : with lock : return func ( * args , * * kwargs ) return wrapper class MonopriceSync ( Monoprice ) : def __init__ ( self , port_url ) : self . _port = serial . serial_for_url ( port_url , do_not_open = True ) self . _port . baudrate = 9600 self . _port . stopbits = serial . STOPBITS_ONE self . _port . bytesize = serial . EIGHTBITS self . _port . parity = serial . PARITY_NONE self . _port . timeout = TIMEOUT self . _port . write_timeout = TIMEOUT self . _port . open ( ) def _process_request ( self , request : bytes , skip = 0 ) : """
            :param request: request that is sent to the monoprice
            :param skip: number of bytes to skip for end of transmission decoding
            :return: ascii string returned by monoprice
            """ _LOGGER . debug ( 'Sending "%s"' , request ) # clear self . _port . reset_output_buffer ( ) self . _port . reset_input_buffer ( ) # send self . _port . write ( request ) self . _port . flush ( ) # receive result = bytearray ( ) while True : c = self . _port . read ( 1 ) if not c : raise serial . SerialTimeoutException ( 'Connection timed out! Last received bytes {}' . format ( [ hex ( a ) for a in result ] ) ) result += c if len ( result ) > skip and result [ - LEN_EOL : ] == EOL : break ret = bytes ( result ) _LOGGER . debug ( 'Received "%s"' , ret ) return ret . decode ( 'ascii' ) @ synchronized def zone_status ( self , zone : int ) : # Ignore first 6 bytes as they will contain 3 byte command and 3 bytes of EOL return ZoneStatus . from_string ( self . _process_request ( _format_zone_status_request ( zone ) , skip = 6 ) ) @ synchronized def set_power ( self , zone : int , power : bool ) : self . _process_request ( _format_set_power ( zone , power ) ) @ synchronized def set_mute ( self , zone : int , mute : bool ) : self . _process_request ( _format_set_mute ( zone , mute ) ) @ synchronized def set_volume ( self , zone : int , volume : int ) : self . _process_request ( _format_set_volume ( zone , volume ) ) @ synchronized def set_treble ( self , zone : int , treble : int ) : self . _process_request ( _format_set_treble ( zone , treble ) ) @ synchronized def set_bass ( self , zone : int , bass : int ) : self . _process_request ( _format_set_bass ( zone , bass ) ) @ synchronized def set_balance ( self , zone : int , balance : int ) : self . _process_request ( _format_set_balance ( zone , balance ) ) @ synchronized def set_source ( self , zone : int , source : int ) : self . _process_request ( _format_set_source ( zone , source ) ) @ synchronized def restore_zone ( self , status : ZoneStatus ) : self . set_power ( status . zone , status . power ) self . set_mute ( status . zone , status . mute ) self . set_volume ( status . zone , status . volume ) self . set_treble ( status . zone , status . treble ) self . set_bass ( status . zone , status . bass ) self . set_balance ( status . zone , status . balance ) self . set_source ( status . zone , status . source ) return MonopriceSync ( port_url )
Return synchronous version of Monoprice interface
    :param port_url: serial port, i.e. '/dev/ttyUSB0'
    :return: synchronous implementation of Monoprice interface
Return synchronous version of Monoprice interface : param port_url : serial port i . e . / dev / ttyUSB0 : return : synchronous implementation of Monoprice interface