Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
kws = list(it.starmap('{}={}'.format, kws.items()))
if fields:
if is_str_native(fields): fields = fields.split()
kws.extend('{}={!r}'.format(k, getattr(self, k)) for k in fields)
kws = sorted(kws)
if ext: kws.append(str(ext))
return ', '.join(kws)
def __str__(self):
return self._as_str(fields=self.c_struct_fields)
def __repr__(self):
return '<{} at {:x} - {}>'.format(self.__class__.__name__, id(self), str(self))
class PulsePortInfo(PulseObject):
c_struct_fields = 'name description available priority'
def _init_from_struct(self, struct):
self.available = PulsePortAvailableEnum._c_val(struct.available)
self.available_state = self.available # for compatibility with <=17.6.0
def __eq__(self, o):
if not isinstance(o, PulsePortInfo): raise TypeError(o)
return self.name == o.name
def __hash__(self): return hash(self.name)
class PulseClientInfo(PulseObject):
c_struct_fields = 'name index driver owner_module'
class PulseServerInfo(PulseObject):
"channel_list" can be a pulse channel map string (comma-separated) or list
of channel names. Defaults to stereo map, should probably match volume channels.
"device" - name of sink/source or None (default).'''
if is_str(struct_or_name):
struct_or_name = self.struct_from_value(
struct_or_name, volume, channel_list, mute, device )
super(PulseExtStreamRestoreInfo, self).__init__(struct_or_name)
def to_struct(self):
return self.struct_from_value(**dict(
(k, getattr(self, k)) for k in 'name volume channel_list mute device'.split() ))
def __str__(self):
return self._as_str(self.volume, fields='name mute device')
class PulseEventInfo(PulseObject):
def __init__(self, ev_t, facility, index):
self.t, self.facility, self.index = ev_t, facility, index
def __str__(self):
return self._as_str(fields='t facility index'.split())
class Pulse(object):
_ctx = None
def __init__(self, client_name=None, server=None, connect=True, threading_lock=False):
'''Connects to specified pulse server by default.
Specifying "connect=False" here prevents that, but be sure to call connect() later.
"connect=False" can also be used here to
class PulsePortInfo(PulseObject):
c_struct_fields = 'name description available priority'
def _init_from_struct(self, struct):
self.available = PulsePortAvailableEnum._c_val(struct.available)
self.available_state = self.available # for compatibility with <=17.6.0
def __eq__(self, o):
if not isinstance(o, PulsePortInfo): raise TypeError(o)
return self.name == o.name
def __hash__(self): return hash(self.name)
class PulseClientInfo(PulseObject):
c_struct_fields = 'name index driver owner_module'
class PulseServerInfo(PulseObject):
c_struct_fields = ( 'user_name host_name'
' server_version server_name default_sink_name default_source_name cookie' )
class PulseModuleInfo(PulseObject):
c_struct_fields = 'index name argument n_used auto_unload'
class PulseSinkInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver'
' monitor_source monitor_source_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
c_struct_fields = ( 'index name mute corked client'
' owner_module sink sample_spec'
' buffer_usec sink_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseSourceInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver monitor_of_sink'
' monitor_of_sink_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSourceOutputInfo(PulseObject):
c_struct_fields = ( 'index name mute corked client'
' owner_module source sample_spec'
' buffer_usec source_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseCardProfileInfo(PulseObject):
c_struct_fields = 'name description n_sinks n_sources priority available'
class PulseCardPortInfo(PulsePortInfo):
c_struct_fields = 'name description available priority direction latency_offset'
def _init_from_struct(self, struct):
super(PulseCardPortInfo, self)._init_from_struct(struct)
self.direction = PulseDirectionEnum._c_val(struct.direction)
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver'
' monitor_source monitor_source_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSinkInputInfo(PulseObject):
c_struct_fields = ( 'index name mute corked client'
' owner_module sink sample_spec'
' buffer_usec sink_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseSourceInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver monitor_of_sink'
' monitor_of_sink_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSourceOutputInfo(PulseObject):
c_struct_fields = ( 'index name mute corked client'
' owner_module source sample_spec'
' buffer_usec source_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseCardProfileInfo(PulseObject):
class PulseCardInfo(PulseObject):
c_struct_fields = 'name index driver owner_module n_profiles'
c_struct_wrappers = {PulsePortInfo: PulseCardPortInfo}
def __init__(self, struct):
super(PulseCardInfo, self).__init__(struct)
self.profile_list = list(
PulseCardProfileInfo(struct.profiles2[n][0]) for n in range(self.n_profiles) )
self.profile_active = PulseCardProfileInfo(struct.active_profile2.contents)
def __str__(self):
return self._as_str(
fields='name index driver n_profiles',
profile_active='[{}]'.format(self.profile_active.name) )
class PulseVolumeInfo(PulseObject):
def __init__(self, struct_or_values=None, channels=None):
if is_num(struct_or_values):
assert channels is not None, 'Channel count specified if volume value is not a list.'
self.values = [struct_or_values] * channels
elif is_list(struct_or_values): self.values = struct_or_values
else:
self.values = list( (x / c.PA_VOLUME_NORM)
for x in map(float, struct_or_values.values[:struct_or_values.channels]) )
@property
def value_flat(self): return (sum(self.values) / float(len(self.values))) if self.values else 0
@value_flat.setter
def value_flat(self, v): self.values = [v] * len(self.values)
def to_struct(self):
' buffer_usec source_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseCardProfileInfo(PulseObject):
c_struct_fields = 'name description n_sinks n_sources priority available'
class PulseCardPortInfo(PulsePortInfo):
c_struct_fields = 'name description available priority direction latency_offset'
def _init_from_struct(self, struct):
super(PulseCardPortInfo, self)._init_from_struct(struct)
self.direction = PulseDirectionEnum._c_val(struct.direction)
class PulseCardInfo(PulseObject):
c_struct_fields = 'name index driver owner_module n_profiles'
c_struct_wrappers = {PulsePortInfo: PulseCardPortInfo}
def __init__(self, struct):
super(PulseCardInfo, self).__init__(struct)
self.profile_list = list(
PulseCardProfileInfo(struct.profiles2[n][0]) for n in range(self.n_profiles) )
self.profile_active = PulseCardProfileInfo(struct.active_profile2.contents)
def __str__(self):
return self._as_str(
fields='name index driver n_profiles',
profile_active='[{}]'.format(self.profile_active.name) )
class PulseVolumeInfo(PulseObject):
class PulseServerInfo(PulseObject):
c_struct_fields = ( 'user_name host_name'
' server_version server_name default_sink_name default_source_name cookie' )
class PulseModuleInfo(PulseObject):
c_struct_fields = 'index name argument n_used auto_unload'
class PulseSinkInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver'
' monitor_source monitor_source_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSinkInputInfo(PulseObject):
c_struct_fields = ( 'index name mute corked client'
' owner_module sink sample_spec'
' buffer_usec sink_usec resample_method driver' )
def __str__(self):
return self._as_str(fields='index name mute')
class PulseSourceInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver monitor_of_sink'
' monitor_of_sink_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSourceOutputInfo(PulseObject):
self.available_state = self.available # for compatibility with <=17.6.0
def __eq__(self, o):
if not isinstance(o, PulsePortInfo): raise TypeError(o)
return self.name == o.name
def __hash__(self): return hash(self.name)
class PulseClientInfo(PulseObject):
c_struct_fields = 'name index driver owner_module'
class PulseServerInfo(PulseObject):
c_struct_fields = ( 'user_name host_name'
' server_version server_name default_sink_name default_source_name cookie' )
class PulseModuleInfo(PulseObject):
c_struct_fields = 'index name argument n_used auto_unload'
class PulseSinkInfo(PulseObject):
c_struct_fields = ( 'index name mute'
' description sample_spec owner_module latency driver'
' monitor_source monitor_source_name flags configured_latency card' )
def __str__(self):
return self._as_str(self.volume, fields='index name description mute')
class PulseSinkInputInfo(PulseObject):
c_struct_fields = ( 'index name mute corked client'
' owner_module sink sample_spec'
' buffer_usec sink_usec resample_method driver' )
def __str__(self):
@property
def value_flat(self): return (sum(self.values) / float(len(self.values))) if self.values else 0
@value_flat.setter
def value_flat(self, v): self.values = [v] * len(self.values)
def to_struct(self):
return c.PA_CVOLUME(
len(self.values), tuple(min( c.PA_VOLUME_UI_MAX,
int(round(v * c.PA_VOLUME_NORM)) ) for v in self.values) )
def __str__(self):
return self._as_str(
channels=len(self.values), volumes='[{}]'.format(
' '.join('{}%'.format(int(round(v*100))) for v in self.values) ) )
class PulseExtStreamRestoreInfo(PulseObject):
c_struct_fields = 'name channel_map volume mute device'
@classmethod
def struct_from_value( cls, name, volume,
channel_list=None, mute=False, device=None ):
'Same arguments as with class instance init.'
chan_map = c.PA_CHANNEL_MAP()
if not channel_list: c.pa.channel_map_init_mono(chan_map)
else:
if not is_str(channel_list):
channel_list = b','.join(map(c.force_bytes, channel_list))
c.pa.channel_map_parse(chan_map, channel_list)
if not isinstance(volume, PulseVolumeInfo):
volume = PulseVolumeInfo(volume, chan_map.channels)
struct = c.PA_EXT_STREAM_RESTORE_INFO(
name=c.force_bytes(name),