first commit

opus/api/__init__.py

@@ -0,0 +1,9 @@
+import ctypes
+from ctypes.util import find_library
+
+
+libopus = ctypes.CDLL(find_library('opus'))
+
+c_int_pointer = ctypes.POINTER(ctypes.c_int)
+c_int16_pointer = ctypes.POINTER(ctypes.c_int16)
+c_float_pointer = ctypes.POINTER(ctypes.c_float)

opus/api/constants.py

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+
+"""Matches to `opus_defines.h`"""
+
+# No Error
+OK = 0
+
+# One or more invalid/out of range arguments
+BAD_ARG = -1
+
+# The mode struct passed is invalid
+BUFFER_TOO_SMALL = -2
+
+# The compressed data passed is corrupted
+INVALID_PACKET = -4
+
+# Invalid/unsupported request number
+UNIMPLEMENTED = -5
+
+
+# Pre-defined values for CTL interface
+
+APPLICATION_VOIP = 2048
+APPLICATION_AUDIO = 2049
+APPLICATION_RESTRICTED_LOWDELAY = 2051
+
+SIGNAL_MUSIC = 3002
+
+# Values for the various encoder CTLs
+
+SET_APPLICATION_REQUEST = 4000
+GET_APPLICATION_REQUEST = 4001
+SET_BITRATE_REQUEST = 4002
+GET_BITRATE_REQUEST = 4003
+SET_MAX_BANDWIDTH_REQUEST = 4004
+GET_MAX_BANDWIDTH_REQUEST = 4005
+SET_VBR_REQUEST = 4006
+GET_VBR_REQUEST = 4007
+SET_BANDWIDTH_REQUEST = 4008
+GET_BANDWIDTH_REQUEST = 4009
+SET_COMPLEXITY_REQUEST = 4010
+GET_COMPLEXITY_REQUEST = 4011
+SET_INBAND_FEC_REQUEST = 4012
+GET_INBAND_FEC_REQUEST = 4013
+SET_PACKET_LOSS_PERC_REQUEST = 4014
+GET_PACKET_LOSS_PERC_REQUEST = 4015
+SET_DTX_REQUEST = 4016
+GET_DTX_REQUEST = 4017
+SET_VBR_CONSTRAINT_REQUEST = 4020
+GET_VBR_CONSTRAINT_REQUEST = 4021
+SET_FORCE_CHANNELS_REQUEST = 4022
+GET_FORCE_CHANNELS_REQUEST = 4023
+SET_SIGNAL_REQUEST = 4024
+GET_SIGNAL_REQUEST = 4025
+GET_LOOKAHEAD_REQUEST = 4027
+RESET_STATE = 4028
+GET_SAMPLE_RATE_REQUEST = 4029
+GET_FINAL_RANGE_REQUEST = 4031
+GET_PITCH_REQUEST = 4033
+SET_GAIN_REQUEST = 4034
+GET_GAIN_REQUEST = 4045
+SET_LSB_DEPTH_REQUEST = 4036
+GET_LSB_DEPTH_REQUEST = 4037
+
+AUTO = -1000
+
+BANDWIDTH_NARROWBAND = 1101
+BANDWIDTH_MEDIUMBAND = 1102
+BANDWIDTH_WIDEBAND = 1103
+BANDWIDTH_SUPERWIDEBAND = 1104
+BANDWIDTH_FULLBAND = 1105

opus/api/ctl.py

@@ -0,0 +1,173 @@
+"""CTL macros rewritten to Python
+
+Usage example:
+
+    from opus.api import decoder, ctl
+
+    dec = decoder.create(48000, 2)
+    decoder.ctl(dec, ctl.set_gain, -15)
+    gain_value = decoder.ctl(dec, ctl.get_gain)
+
+"""
+
+import ctypes
+
+from opus.api import constants
+from opus.exceptions import OpusError
+
+
+def query(request):
+    """Query encoder/decoder with a request value"""
+
+    def inner(func, obj):
+        result_code = func(obj, request)
+
+        if result_code is not constants.OK:
+            raise OpusError(result_code)
+
+        return result_code
+
+    return inner
+
+
+def get(request, result_type):
+    """Get CTL value from a encoder/decoder"""
+
+    def inner(func, obj):
+        result = result_type()
+        result_code = func(obj, request, ctypes.byref(result))
+
+        if result_code is not constants.OK:
+            raise OpusError(result_code)
+
+        return result.value
+
+    return inner
+
+
+def set(request):
+    """Set new CTL value to a encoder/decoder"""
+
+    def inner(func, obj, value):
+        result_code = func(obj, request, value)
+        if result_code is not constants.OK:
+            raise OpusError(result_code)
+
+    return inner
+
+#
+# Generic CTLs
+#
+
+# Resets the codec state to be equivalent to a freshly initialized state
+reset_state = query(constants.RESET_STATE)
+
+# Gets the final state of the codec's entropy coder
+get_final_range = get(constants.GET_FINAL_RANGE_REQUEST, ctypes.c_uint)
+
+# Gets the encoder's configured bandpass or the decoder's last bandpass
+get_bandwidth = get(constants.GET_BANDWIDTH_REQUEST, ctypes.c_int)
+
+# Gets the pitch of the last decoded frame, if available
+get_pitch = get(constants.GET_PITCH_REQUEST, ctypes.c_int)
+
+# Configures the depth of signal being encoded
+set_lsb_depth = set(constants.SET_LSB_DEPTH_REQUEST)
+
+# Gets the encoder's configured signal depth
+get_lsb_depth = get(constants.GET_LSB_DEPTH_REQUEST, ctypes.c_int)
+
+#
+# Decoder related CTLs
+#
+
+# Gets the decoder's configured gain adjustment
+get_gain = get(constants.GET_GAIN_REQUEST, ctypes.c_int)
+
+# Configures decoder gain adjustment
+set_gain = set(constants.SET_GAIN_REQUEST)
+
+#
+# Encoder related CTLs
+#
+
+# Configures the encoder's computational complexity
+set_complexity = set(constants.SET_COMPLEXITY_REQUEST)
+
+# Gets the encoder's complexity configuration
+get_complexity = get(constants.GET_COMPLEXITY_REQUEST, ctypes.c_int)
+
+# Configures the bitrate in the encoder
+set_bitrate = set(constants.SET_BITRATE_REQUEST)
+
+# Gets the encoder's bitrate configuration
+get_bitrate = get(constants.GET_BITRATE_REQUEST, ctypes.c_int)
+
+# Enables or disables variable bitrate (VBR) in the encoder
+set_vbr = set(constants.SET_VBR_REQUEST)
+
+# Determine if variable bitrate (VBR) is enabled in the encoder
+get_vbr = get(constants.GET_VBR_REQUEST, ctypes.c_int)
+
+# Enables or disables constrained VBR in the encoder
+set_vbr_constraint = set(constants.SET_VBR_CONSTRAINT_REQUEST)
+
+# Determine if constrained VBR is enabled in the encoder
+get_vbr_constraint = get(constants.GET_VBR_CONSTRAINT_REQUEST, ctypes.c_int)
+
+# Configures mono/stereo forcing in the encoder
+set_force_channels = set(constants.SET_FORCE_CHANNELS_REQUEST)
+
+# Gets the encoder's forced channel configuration
+get_force_channels = get(constants.GET_FORCE_CHANNELS_REQUEST, ctypes.c_int)
+
+# Configures the maximum bandpass that the encoder will select automatically
+set_max_bandwidth = set(constants.SET_MAX_BANDWIDTH_REQUEST)
+
+# Gets the encoder's configured maximum allowed bandpass
+get_max_bandwidth = get(constants.GET_MAX_BANDWIDTH_REQUEST, ctypes.c_int)
+
+# Sets the encoder's bandpass to a specific value
+set_bandwidth = set(constants.SET_BANDWIDTH_REQUEST)
+
+# Configures the type of signal being encoded
+set_signal = set(constants.SET_SIGNAL_REQUEST)
+
+# Gets the encoder's configured signal type
+get_signal = get(constants.GET_SIGNAL_REQUEST, ctypes.c_int)
+
+# Configures the encoder's intended application
+set_application = set(constants.SET_APPLICATION_REQUEST)
+
+# Gets the encoder's configured application
+get_application = get(constants.GET_APPLICATION_REQUEST, ctypes.c_int)
+
+# Gets the sampling rate the encoder or decoder was initialized with
+get_sample_rate = get(constants.GET_SAMPLE_RATE_REQUEST, ctypes.c_int)
+
+# Gets the total samples of delay added by the entire codec
+get_lookahead = get(constants.GET_LOOKAHEAD_REQUEST, ctypes.c_int)
+
+# Configures the encoder's use of inband forward error correction (FEC)
+set_inband_fec = set(constants.SET_INBAND_FEC_REQUEST)
+
+# Gets encoder's configured use of inband forward error correction
+get_inband_fec = get(constants.GET_INBAND_FEC_REQUEST, ctypes.c_int)
+
+# Configures the encoder's expected packet loss percentage
+set_packet_loss_perc = set(constants.SET_PACKET_LOSS_PERC_REQUEST)
+
+# Gets the encoder's configured packet loss percentage
+get_packet_loss_perc = get(constants.GET_PACKET_LOSS_PERC_REQUEST, ctypes.c_int)
+
+# Configures the encoder's use of discontinuous transmission (DTX)
+set_dtx = set(constants.SET_DTX_REQUEST)
+
+# Gets encoder's configured use of discontinuous transmission
+get_dtx = get(constants.GET_DTX_REQUEST, ctypes.c_int)
+
+#
+# Other stuff
+#
+
+unimplemented = query(constants.UNIMPLEMENTED)

opus/api/decoder.py

@@ -0,0 +1,187 @@
+# -*- coding: utf-8 -*-
+
+import array
+import ctypes
+
+from opus.api import libopus, c_int_pointer, c_int16_pointer, c_float_pointer
+from opus.exceptions import OpusError
+
+
+class Decoder(ctypes.Structure):
+    """Opus decoder state.
+
+    This contains the complete state of an Opus decoder.
+    """
+
+    pass
+
+DecoderPointer = ctypes.POINTER(Decoder)
+
+
+get_size = libopus.opus_decoder_get_size
+get_size.argtypes = (ctypes.c_int,)
+get_size.restype = ctypes.c_int
+get_size.__doc__ = 'Gets the size of an OpusDecoder structure'
+
+
+_create = libopus.opus_decoder_create
+_create.argtypes = (ctypes.c_int, ctypes.c_int, c_int_pointer)
+_create.restype = DecoderPointer
+
+
+def create(fs, channels):
+    """Allocates and initializes a decoder state"""
+
+    result_code = ctypes.c_int()
+
+    result = _create(fs, channels, ctypes.byref(result_code))
+    if result_code.value is not 0:
+        raise OpusError(result_code.value)
+
+    return result
+
+
+_packet_get_bandwidth = libopus.opus_packet_get_bandwidth
+_packet_get_bandwidth.argtypes = (ctypes.c_char_p,)
+_packet_get_bandwidth.restype = ctypes.c_int
+
+
+def packet_get_bandwidth(data):
+    """Gets the bandwidth of an Opus packet."""
+
+    data_pointer = ctypes.c_char_p(data)
+
+    result = _packet_get_bandwidth(data_pointer)
+    if result < 0:
+        raise OpusError(result)
+
+    return result
+
+
+_packet_get_nb_channels = libopus.opus_packet_get_nb_channels
+_packet_get_nb_channels.argtypes = (ctypes.c_char_p,)
+_packet_get_nb_channels.restype = ctypes.c_int
+
+
+def packet_get_nb_channels(data):
+    """Gets the number of channels from an Opus packet"""
+
+    data_pointer = ctypes.c_char_p(data)
+
+    result = _packet_get_nb_channels(data_pointer)
+    if result < 0:
+        raise OpusError(result)
+
+    return result
+
+
+_packet_get_nb_frames = libopus.opus_packet_get_nb_frames
+_packet_get_nb_frames.argtypes = (ctypes.c_char_p, ctypes.c_int)
+_packet_get_nb_frames.restype = ctypes.c_int
+
+
+def packet_get_nb_frames(data, length=None):
+    """Gets the number of frames in an Opus packet"""
+
+    data_pointer = ctypes.c_char_p(data)
+    if length is None:
+        length = len(data)
+
+    result = _packet_get_nb_frames(data_pointer, ctypes.c_int(length))
+    if result < 0:
+        raise OpusError(result)
+
+    return result
+
+
+_packet_get_samples_per_frame = libopus.opus_packet_get_samples_per_frame
+_packet_get_samples_per_frame.argtypes = (ctypes.c_char_p, ctypes.c_int)
+_packet_get_samples_per_frame.restype = ctypes.c_int
+
+
+def packet_get_samples_per_frame(data, fs):
+    """Gets the number of samples per frame from an Opus packet"""
+
+    data_pointer = ctypes.c_char_p(data)
+
+    result = _packet_get_nb_frames(data_pointer, ctypes.c_int(fs))
+    if result < 0:
+        raise OpusError(result)
+
+    return result
+
+
+_get_nb_samples = libopus.opus_decoder_get_nb_samples
+_get_nb_samples.argtypes = (DecoderPointer, ctypes.c_char_p, ctypes.c_int32)
+_get_nb_samples.restype = ctypes.c_int
+
+
+def get_nb_samples(decoder, packet, length):
+    result = _get_nb_samples(decoder, packet, length)
+    if result < 0:
+        raise OpusError(result)
+
+    return result
+
+
+_decode = libopus.opus_decode
+_decode.argtypes = (DecoderPointer, ctypes.c_char_p, ctypes.c_int32, c_int16_pointer, ctypes.c_int, ctypes.c_int)
+_decode.restype = ctypes.c_int
+
+
+def decode(decoder, data, length, frame_size, decode_fec, channels=2):
+    """Decode an Opus frame
+
+    Unlike the `opus_decode` function , this function takes an additional parameter `channels`,
+    which indicates the number of channels in the frame
+    """
+
+    pcm_size = frame_size * channels * ctypes.sizeof(ctypes.c_int16)
+    pcm = (ctypes.c_int16 * pcm_size)()
+    pcm_pointer = ctypes.cast(pcm, c_int16_pointer)
+
+    # Converting from a boolean to int
+    decode_fec = int(bool(decode_fec))
+
+    result = _decode(decoder, data, length, pcm_pointer, frame_size, decode_fec)
+    if result < 0:
+        raise OpusError(result)
+
+    return array.array('h', pcm[ :result * channels ]).tostring()
+
+
+_decode_float = libopus.opus_decode_float
+_decode_float.argtypes = (DecoderPointer, ctypes.c_char_p, ctypes.c_int32, c_float_pointer, ctypes.c_int, ctypes.c_int)
+_decode_float.restype = ctypes.c_int
+
+
+def decode_float(decoder, data, length, frame_size, decode_fec, channels=2):
+    pcm_size = frame_size * channels * ctypes.sizeof(ctypes.c_float)
+    pcm = (ctypes.c_float * pcm_size)()
+    pcm_pointer = ctypes.cast(pcm, c_float_pointer)
+
+    # Converting from a boolean to int
+    decode_fec = int(bool(decode_fec))
+
+    result = _decode_float(decoder, data, length, pcm_pointer, frame_size, decode_fec)
+    if result < 0:
+        raise OpusError(result)
+
+    return array.array('f', pcm[ : result * channels ]).tostring()
+
+
+_ctl = libopus.opus_decoder_ctl
+_ctl.restype = ctypes.c_int
+
+
+def ctl(decoder, request, value=None):
+    if value is not None:
+        return request(_ctl, decoder, value)
+
+    return request(_ctl, decoder)
+
+
+destroy = libopus.opus_decoder_destroy
+destroy.argtypes = (DecoderPointer,)
+destroy.restype = None
+destroy.__doc__ = 'Frees an OpusDecoder allocated by opus_decoder_create()'

opus/api/encoder.py

@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+
+import ctypes
+import array
+
+from opus.api import constants, libopus, c_int_pointer, c_int16_pointer, c_float_pointer
+from opus.exceptions import OpusError
+
+
+class Encoder(ctypes.Structure):
+    """Opus encoder state.
+
+    This contains the complete state of an Opus encoder.
+    """
+
+    pass
+
+EncoderPointer = ctypes.POINTER(Encoder)
+
+
+_get_size = libopus.opus_encoder_get_size
+_get_size.argtypes = (ctypes.c_int,)
+_get_size.restype = ctypes.c_int
+
+
+def get_size(channels):
+    """Gets the size of an OpusEncoder structure."""
+
+    if not channels in (1, 2):
+        raise ValueError('Wrong channels value. Must be equal to 1 or 2')
+
+    return _get_size(channels)
+
+
+_create = libopus.opus_encoder_create
+_create.argtypes = (ctypes.c_int, ctypes.c_int, ctypes.c_int, c_int_pointer)
+_create.restype = EncoderPointer
+
+
+def create(fs, channels, application):
+    """Allocates and initializes an encoder state."""
+
+    result_code = ctypes.c_int()
+
+    result = _create(fs, channels, application, ctypes.byref(result_code))
+    if result_code.value is not constants.OK:
+        raise OpusError(result_code.value)
+
+    return result
+
+
+_ctl = libopus.opus_encoder_ctl
+_ctl.restype = ctypes.c_int
+
+
+def ctl(encoder, request, value=None):
+    if value is not None:
+        return request(_ctl, encoder, value)
+
+    return request(_ctl, encoder)
+
+
+_encode = libopus.opus_encode
+_encode.argtypes = (EncoderPointer, c_int16_pointer, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32)
+_encode.restype = ctypes.c_int32
+
+
+def encode(encoder, pcm, frame_size, max_data_bytes):
+    """Encodes an Opus frame
+
+    Returns string output payload
+    """
+
+    pcm = ctypes.cast(pcm, c_int16_pointer)
+    data = (ctypes.c_char * max_data_bytes)()
+
+    result = _encode(encoder, pcm, frame_size, data, max_data_bytes)
+    if result < 0:
+        raise OpusError(result)
+
+    return array.array('c', data[:result]).tostring()
+
+
+_encode_float = libopus.opus_encode_float
+_encode_float.argtypes = (EncoderPointer, c_float_pointer, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32)
+_encode_float.restype = ctypes.c_int32
+
+
+def encode_float(encoder, pcm, frame_size, max_data_bytes):
+    """Encodes an Opus frame from floating point input"""
+
+    pcm = ctypes.cast(pcm, c_float_pointer)
+    data = (ctypes.c_char * max_data_bytes)()
+
+    result = _encode_float(encoder, pcm, frame_size, data, max_data_bytes)
+    if result < 0:
+        raise OpusError(result)
+
+    return array.array('c', data[:result]).tostring()
+
+
+destroy = libopus.opus_encoder_destroy
+destroy.argtypes = (EncoderPointer,)
+destroy.restype = None
+destroy.__doc__ = "Frees an OpusEncoder allocated by opus_encoder_create()"

opus/api/info.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+
+import ctypes
+
+from opus.api import libopus
+
+
+strerror = libopus.opus_strerror
+strerror.argtypes = (ctypes.c_int,)
+strerror.restype = ctypes.c_char_p
+strerror.__doc__ = '''Converts an opus error code into a human readable string'''
+
+
+get_version_string = libopus.opus_get_version_string
+get_version_string.argtypes = None
+get_version_string.restype = ctypes.c_char_p
+get_version_string.__doc__ = 'Gets the libopus version string'