Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions alsa_midi/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1122,11 +1122,58 @@ def query_next_port(self,
result = PortInfo._from_alsa(info)
return result

def list_clients(self, *,
include_system: bool = False,
include_kernel: bool = False,
include_midi_through: bool = True,
) -> list[ClientInfo]:
"""More friendly interface to list available clients..
:param include_system: include system clients
:param include_midi_through: include 'midi through' ports

:return: list of client information
"""

result = []
self._check_handle()

client_ainfo_p = ffi.new("snd_seq_client_info_t **")
err = alsa.snd_seq_client_info_malloc(client_ainfo_p)
_check_alsa_error(err)
client_ainfo = ffi.gc(client_ainfo_p[0], alsa.snd_seq_client_info_free)

alsa.snd_seq_client_info_set_client(client_ainfo, -1)
while True:
err = alsa.snd_seq_query_next_client(self.handle, client_ainfo)
if err == -errno.ENOENT:
break
_check_alsa_error(err)

client_id = alsa.snd_seq_client_info_get_client(client_ainfo)
if client_id == 0 and not include_system:
continue

client_name = alsa.snd_seq_client_info_get_name(client_ainfo)
client_name = ffi.string(client_name).decode()

if client_name == "Midi Through" and not include_midi_through:
continue

client_info = ClientInfo._from_alsa(client_ainfo)

if client_info.type == ClientType.KERNEL and not include_kernel:
continue

result.append(client_info)

return result

def list_ports(self, *,
input: bool = None,
output: bool = None,
type: PortType = PortType.MIDI_GENERIC,
include_system: bool = False,
include_kernel: bool = False,
include_midi_through: bool = True,
include_no_export: bool = True,
only_connectable: bool = True,
Expand All @@ -1147,6 +1194,7 @@ def list_ports(self, *,
:param output: return ports usable for event output (`PortCaps.WRITE`)
:param type: limit ouput to ports of this type
:param include_system: include system ports
:param include_kernel: include kernel ports
:param include_midi_through: include 'midi through' ports
:param include_no_export: include 'no export' ports
:param only_connectable: only list ports that can be connected to/from
Expand Down Expand Up @@ -1176,9 +1224,14 @@ def list_ports(self, *,
_check_alsa_error(err)

client_id = alsa.snd_seq_client_info_get_client(client_ainfo)

if client_id == 0 and not include_system:
continue

client_type = alsa.snd_seq_client_info_get_type(client_ainfo)
if client_type & alsa.SND_SEQ_KERNEL_CLIENT and not include_kernel:
continue

client_name = alsa.snd_seq_client_info_get_name(client_ainfo)
client_name = ffi.string(client_name).decode()

Expand Down
2 changes: 2 additions & 0 deletions alsa_midi/port.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
if TYPE_CHECKING:
from .client import SequencerClientBase, SubscriptionQuery, SubscriptionQueryType, _snd_seq_t

#


class PortCaps(IntFlag):
"""Port capability flags."""
Expand Down
18 changes: 10 additions & 8 deletions tests/test_port_query.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,7 @@ def test_list_ports(alsa_seq_state):
# test defaults
alsa_seq_state.load()
alsa_client_names = [client.name for client in alsa_seq_state.clients.values()]
ports = client.list_ports()
ports = client.list_ports(include_system=True, include_kernel=True)

for port in ports:
assert isinstance(port, PortInfo)
Expand Down Expand Up @@ -237,7 +237,7 @@ def test_list_ports(alsa_seq_state):
assert s_inout_port_a not in port_addrs

# test include_system=True
ports = client.list_ports(include_system=True, type=PortType.ANY)
ports = client.list_ports(include_system=True, include_kernel=True, type=PortType.ANY)
port_addrs = {(p.client_id, p.port_id) for p in ports}
client_names = {p.client_name for p in ports}
if "Midi Through" in alsa_client_names:
Expand All @@ -253,7 +253,7 @@ def test_list_ports(alsa_seq_state):
assert inout_port_ne_a in port_addrs

# test include_system=False
ports = client.list_ports(include_system=False, type=PortType.ANY)
ports = client.list_ports(include_system=False, include_kernel=True, type=PortType.ANY)
port_addrs = {(p.client_id, p.port_id) for p in ports}
client_names = {p.client_name for p in ports}
if "Midi Through" in alsa_client_names:
Expand All @@ -269,7 +269,7 @@ def test_list_ports(alsa_seq_state):
assert inout_port_ne_a in port_addrs

# test include_midi_through=True
ports = client.list_ports(include_midi_through=True)
ports = client.list_ports(include_midi_through=True, include_kernel=True)
port_addrs = {(p.client_id, p.port_id) for p in ports}
client_names = {p.client_name for p in ports}
if "Midi Through" in alsa_client_names:
Expand Down Expand Up @@ -378,7 +378,7 @@ def test_list_ports_sorting(alsa_seq_state):
# test defaults
alsa_seq_state.load()
alsa_client_names = [client.name for client in alsa_seq_state.clients.values()]
ports = client.list_ports(type=PortType.ANY)
ports = client.list_ports(type=PortType.ANY, include_kernel=True)

for port in ports:
assert isinstance(port, PortInfo)
Expand Down Expand Up @@ -444,7 +444,7 @@ def test_list_ports_sorting(alsa_seq_state):

# default sorting
ports = client.list_ports(only_connectable=False, include_system=True,
type=PortType.ANY)
type=PortType.ANY, include_kernel=True)
pa = [(p.client_id, p.port_id) for p in ports]

assert pa.index(specific_inout_a) > pa.index(midi_generic_input_a)
Expand All @@ -459,7 +459,8 @@ def test_list_ports_sorting(alsa_seq_state):
assert ports[-1].client_name == "Midi Through"

# default input port sorting
ports = client.list_ports(only_connectable=False, include_system=True, input=True,
ports = client.list_ports(only_connectable=False, include_system=True,
include_kernel=True, input=True,
type=PortType.ANY)
pa = [(p.client_id, p.port_id) for p in ports]

Expand All @@ -472,7 +473,8 @@ def test_list_ports_sorting(alsa_seq_state):
assert ports[-1].client_name == "Midi Through"

# default output port sorting
ports = client.list_ports(only_connectable=False, include_system=True, output=True,
ports = client.list_ports(only_connectable=False, include_system=True,
include_kernel=True, output=True,
type=PortType.ANY)
pa = [(p.client_id, p.port_id) for p in ports]

Expand Down
Loading
Loading