def setup_device(cls, ctrl_connection: Connection, device_identifier):
     ret = usrp.open(device_identifier)
     ctrl_connection.send("OPEN:" + str(ret))
     success = ret == 0
     if success:
         device_repr = usrp.get_device_representation()
         ctrl_connection.send(device_repr)
     return success

    def setup_device(cls, ctrl_connection: Connection, device_identifier):
        ret = hackrf.setup(device_identifier)
        msg = "SETUP"
        if device_identifier:
            msg += " ({})".format(device_identifier)
        msg += ": "+str(ret)
        ctrl_connection.send(msg)

        return ret == 0

    def shutdown_device(cls, ctrl_conn: Connection):
        logger.debug("HackRF: closing device")
        ret = hackrf.close()
        ctrl_conn.send("CLOSE:" + str(ret))

        ret = hackrf.exit()
        ctrl_conn.send("EXIT:" + str(ret))

        return True

    class RawSocketHandler(socketserver.BaseRequestHandler):
        def setup(self):
            registry.append(self)

        def handle(self):
            self.conn = Connection(self.request.detach())
            while self.conn._handle:
                try:
                    obj = json.loads(self.conn.recv_bytes().decode('utf-8'))
                except EOFError:
                    break
                if obj['type'] == 'message':
                    bus.post(nt_from_dict(Message, obj['message'], None))
                    self.conn.send_bytes(json.dumps({'ret': True}).encode('utf-8'))
                elif obj['type'] == 'request':
                    m = bus.post_sync(nt_from_dict(Message, obj['message'], None))
                    if m:
                        ret = {"ret": True, "response": m._asdict()}
                    else:
                        ret = {"ret": False, "response": None}
                    self.conn.send_bytes(json.dumps(ret).encode('utf-8'))

        def send(self, msg):
            if isinstance(msg, Message):
                ret = {"type": "message", "message": msg._asdict()}
            else:
                ret = {"type": "response", "response": msg._asdict()}
            self.conn.send_bytes(json.dumps(ret).encode('utf-8'))

        def finish(self):
            registry.remove(self)

        def close(self):
            self.conn.close()

 def prepare_sync_send(cls, ctrl_connection: Connection):
     try:
         cls.pyaudio_stream = cls.pyaudio_handle.open(format=pyaudio.paFloat32,
                                                      channels=2,
                                                      rate=cls.SAMPLE_RATE,
                                                      output=True)
         ctrl_connection.send("Successfully started pyaudio stream")
         return 0
     except Exception as e:
         logger.exception(e)
         ctrl_connection.send("Failed to start pyaudio stream")

    def setup_device(cls, ctrl_connection: Connection, device_identifier):
        ret = limesdr.open()
        ctrl_connection.send("OPEN:" + str(ret))
        limesdr.disable_all_channels()
        if ret != 0:
            return False

        ret = limesdr.init()
        ctrl_connection.send("INIT:" + str(ret))

        return ret == 0

 def prepare_sync_receive(cls, ctrl_connection: Connection):
     try:
         cls.pyaudio_stream = cls.pyaudio_handle.open(format=pyaudio.paFloat32,
                                                      channels=2,
                                                      rate=cls.SAMPLE_RATE,
                                                      input=True,
                                                      frames_per_buffer=cls.CHUNK_SIZE)
         ctrl_connection.send("Successfully started pyaudio stream")
         return 0
     except Exception as e:
         logger.exception(e)
         ctrl_connection.send("Failed to start pyaudio stream")

def generate_data(connection: Connection, num_samples=32768):
    frequency = 0.1
    divisor = 200
    pos = 0
    while True:
        result = np.zeros(num_samples, dtype=np.complex64)
        result.real = np.cos(2 * np.pi * frequency * np.arange(pos, pos + num_samples))
        result.imag = np.sin(2 * np.pi * frequency * np.arange(pos, pos + num_samples))
        pos += num_samples
        if pos / num_samples >= divisor:
            frequency *= 2
            if frequency >= 1:
                frequency = 0.1
            pos = 0
        connection.send(result)

    def device_send(cls, ctrl_connection: Connection, send_config: SendConfig, dev_parameters: OrderedDict):
        if not cls.init_device(ctrl_connection, is_tx=True, parameters=dev_parameters):
            ctrl_connection.send("failed to start tx mode")
            return False

        if cls.ASYNCHRONOUS:
            ret = cls.enter_async_send_mode(send_config.get_data_to_send)
        else:
            ret = cls.prepare_sync_send(ctrl_connection)

        if ret != 0:
            ctrl_connection.send("failed to start tx mode")
            return False

        exit_requested = False
        buffer_size = cls.CONTINUOUS_TX_CHUNK_SIZE if send_config.continuous else cls.SYNC_TX_CHUNK_SIZE
        if not cls.ASYNCHRONOUS and buffer_size == 0:
            logger.warning("Send buffer size is zero!")

        ctrl_connection.send("successfully started tx mode")

        while not exit_requested and not send_config.sending_is_finished():
            if cls.ASYNCHRONOUS:
                try:
                    time.sleep(0.5)
                except KeyboardInterrupt:
                    pass
            else:
                cls.send_sync(send_config.get_data_to_send(buffer_size))

            while ctrl_connection.poll():
                result = cls.process_command(ctrl_connection.recv(), ctrl_connection, is_tx=True)
                if result == cls.Command.STOP.name:
                    exit_requested = True
                    break

        if not cls.ASYNCHRONOUS:
            # Some Sync send calls (e.g. USRP) are not blocking, so we wait a bit here to ensure
            # that the send buffer on the SDR is cleared
            time.sleep(0.75)

        if exit_requested:
            logger.debug("{}: exit requested. Stopping sending".format(cls.__class__.__name__))
        if send_config.sending_is_finished():
            logger.debug("{}: sending is finished.".format(cls.__class__.__name__))

        cls.shutdown_device(ctrl_connection, is_tx=True)
        ctrl_connection.close()

    def setup_device(cls, ctrl_connection: Connection, device_identifier):
        device_identifier = device_identifier if isinstance(device_identifier, str) else ""
        try:
            device_identifier = re.search("\[.*\]", device_identifier).groups(0)
        except (IndexError, AttributeError):
            pass

        if not device_identifier:
            _, uris = plutosdr.scan_devices()
            try:
                device_identifier = uris[0]
            except IndexError:
                ctrl_connection.send("Could find a connected PlutoSDR")
                return False

        ret = plutosdr.open(device_identifier)
        ctrl_connection.send("OPEN ({}):{}".format(device_identifier, ret))
        return ret == 0

 def _process(self, conn_lru_dict: LRUCacheType[multiprocessing_connection.Connection, bool],
              handle_pool: ThreadPool,
              c_recv: multiprocessing_connection.Connection,
              c_send: multiprocessing_connection.Connection,
              wait=multiprocessing_connection.wait):
     while True:
         try:
             for conn in wait(list(conn_lru_dict.keys()) + [c_recv]):
                 if conn == c_recv:
                     c_recv.recv_bytes()
                     continue
                 del conn_lru_dict[conn]
                 if not conn.closed:
                     handle_pool.spawn(self._handle, conn_lru_dict, conn, c_send)
                 else:
                     self.logger.debug("conn %s was closed" % conn)
         except OSError as e:
             if getattr(e, "winerror", 0) == 6:
                 continue
             logging.exception("OSError")
         except:
             logging.exception("error")

    def shutdown_device(cls, ctrl_conn: Connection, is_tx: bool):
        if is_tx:
            result = hackrf.stop_tx_mode()
            ctrl_conn.send("STOP TX MODE:" + str(result))
        else:
            result = hackrf.stop_rx_mode()
            ctrl_conn.send("STOP RX MODE:" + str(result))

        result = hackrf.close()
        ctrl_conn.send("CLOSE:" + str(result))

        result = hackrf.exit()
        ctrl_conn.send("EXIT:" + str(result))

        return True

 def setup_device(cls, ctrl_connection: Connection, device_identifier):
     ctrl_connection.send("Initializing pyaudio...")
     try:
         cls.pyaudio_handle = pyaudio.PyAudio()
         ctrl_connection.send("Initialized pyaudio")
         return True
     except Exception as e:
         logger.exception(e)
         ctrl_connection.send("Failed to initialize pyaudio")

 def handle(self):
     self.conn = Connection(self.request.detach())
     while self.conn._handle:
         try:
             obj = json.loads(self.conn.recv_bytes().decode('utf-8'))
         except EOFError:
             break
         if obj['type'] == 'message':
             bus.post(nt_from_dict(Message, obj['message'], None))
             self.conn.send_bytes(json.dumps({'ret': True}).encode('utf-8'))
         elif obj['type'] == 'request':
             m = bus.post_sync(nt_from_dict(Message, obj['message'], None))
             if m:
                 ret = {"ret": True, "response": m._asdict()}
             else:
                 ret = {"ret": False, "response": None}
             self.conn.send_bytes(json.dumps(ret).encode('utf-8'))

    def init_device(cls, ctrl_connection: Connection, is_tx: bool, parameters: OrderedDict):
        if not cls.setup_device(ctrl_connection, device_identifier=parameters["identifier"]):
            return False

        limesdr.enable_channel(True, is_tx, parameters[cls.Command.SET_CHANNEL_INDEX.name])
        limesdr.set_tx(is_tx)

        for parameter, value in parameters.items():
            cls.process_command((parameter, value), ctrl_connection, is_tx)

        antennas = limesdr.get_antenna_list()
        ctrl_connection.send("Current normalized gain is {0:.2f}".format(limesdr.get_normalized_gain()))
        ctrl_connection.send("Current antenna is {0}".format(antennas[limesdr.get_antenna()]))
        ctrl_connection.send("Current chip temperature is {0:.2f}°C".format(limesdr.get_chip_temperature()))

        return True

 def prepare_sync_send(cls, ctrl_connection: Connection):
     ctrl_connection.send("Initializing stream...")
     limesdr.setup_stream(cls.SEND_FIFO_SIZE)
     ret = limesdr.start_stream()
     ctrl_connection.send("Initialize stream:{0}".format(ret))
     return ret

    def device_receive(cls, data_connection: Connection, ctrl_connection: Connection, dev_parameters: OrderedDict):
        if not cls.init_device(ctrl_connection, is_tx=False, parameters=dev_parameters):
            ctrl_connection.send("failed to start rx mode")
            return False

        try:
            cls.adapt_num_read_samples_to_sample_rate(dev_parameters[cls.Command.SET_SAMPLE_RATE.name])
        except NotImplementedError:
            # Many SDRs like HackRF or AirSpy do not need to calculate SYNC_RX_CHUNK_SIZE
            # as default values are either fine or given by the hardware
            pass

        if cls.ASYNCHRONOUS:
            ret = cls.enter_async_receive_mode(data_connection, ctrl_connection)
        else:
            ret = cls.prepare_sync_receive(ctrl_connection)

        if ret != 0:
            ctrl_connection.send("failed to start rx mode")
            return False

        exit_requested = False
        ctrl_connection.send("successfully started rx mode")

        while not exit_requested:
            if cls.ASYNCHRONOUS:
                time.sleep(0.25)
            else:
                cls.receive_sync(data_connection)
            while ctrl_connection.poll():
                result = cls.process_command(ctrl_connection.recv(), ctrl_connection, is_tx=False)
                if result == cls.Command.STOP.name:
                    exit_requested = True
                    break

        cls.shutdown_device(ctrl_connection, is_tx=False)
        data_connection.close()
        ctrl_connection.close()

 def receive_sync(cls, data_conn: Connection):
     data_conn.send_bytes(rtlsdr.read_sync())

 def prepare_sync_receive(cls, ctrl_connection: Connection):
     ret = rtlsdr.reset_buffer()
     ctrl_connection.send("RESET_BUFFER:" + str(ret))
     return ret

 def setup_device(cls, ctrl_connection: Connection, device_identifier):
     # identifier gets set in self.receive_process_arguments
     device_number = int(device_identifier)
     ret = rtlsdr.open(device_number)
     ctrl_connection.send("OPEN (#{}):{}".format(device_number, ret))
     return ret == 0