simple_client.py 文件源码

def write(self, data):
        """Writes samples of the given SampleSpec to the PulseAudio server."""

        self._fail_if_not_connected()

        # TODO: Use self.create_data_array...
        if self._sample_format == SampleFormat.PA_SAMPLE_U8:
            size = len(data)
            c_data = (ctypes.c_uint8 * size)(*data)
        elif self._sample_format == SampleFormat.PA_SAMPLE_S16LE:
            raise Exception('Not implemented yet.')
        elif self._sample_format == SampleFormat.PA_SAMPLE_FLOAT32LE:
            raise Exception('Not implemented yet.')

        error = ctypes.c_int(0)
        result = _libpulse_simple.pa_simple_write(self._client, c_data, size, error)
        if result < 0:
            raise PulseAudioException('Could not write data.', error.value)