Skip to content
Open
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
78 changes: 77 additions & 1 deletion core.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,14 @@
from enum import Enum
import re
import shlex
import subprocess
from enum import Enum
from system import run_and_check, CommandValidationException


PREFFERED_PROFILE = 'headset-head-unit'


class BluezAddressType(Enum):
BR_EDR = 0
LE_PUBLIC = 1
Expand Down Expand Up @@ -113,11 +117,83 @@ def to_source_name(target: BluezTarget) -> str:
return "bluez_input." + normalize_address(target=target) + ".0"


def get_bluetooth_profile(card_name: str, verbose: bool = False) -> str:
result = subprocess.run(
['pactl', 'list', 'cards'],
capture_output=True,
text=True,
check=False
)

if result.returncode != 0:
raise CommandValidationException(
"pactl list cards",
result.stderr or "pactl command failed"
)

output = result.stdout

card_section = []
in_our_card = False

for line in output.split('\n'):
if f"Name: {card_name}" in line:
in_our_card = True
card_section = []
elif in_our_card:
if line.startswith('Card #') or (line.startswith('\t\tName:') and line.strip() != f"Name: {card_name}"):
break
card_section.append(line)

if not card_section:
raise CommandValidationException(
f"pactl list cards (search for {card_name})",
f"Card {card_name} not found in pactl output"
)

profiles = []
in_profiles_section = False

for line in card_section:
if 'Profiles:' in line:
in_profiles_section = True
continue

if in_profiles_section:
if line.startswith('\t\t') and ':' in line:
match = re.match(r'\t\t([^:]+):', line)
if match:
profile_name = match.group(1).strip()
line_lower = line.lower()
is_available = 'available: yes' in line_lower or 'available: unknown' in line_lower
has_sources = 'sources: 0' not in line_lower
if is_available and has_sources:
profiles.append(profile_name)
if profile_name == PREFFERED_PROFILE:
if verbose:
print(f"Using preferred profile: {profile_name}")
return profile_name
elif not line.startswith('\t\t'):
break

if not profiles:
raise CommandValidationException(
f"pactl list cards (get profiles for {card_name})",
f"No available profiles found for card {card_name}"
)

if verbose:
print(f"Using fallback profile: {profiles[0]} (available: {', '.join(profiles)})")

return profiles[0]


def record(target: BluezTarget, outfile: str, verbose: bool = True):
source_name = to_source_name(target)
card_name = to_card_name(target)

run_and_check(
shlex.split(f"pactl set-card-profile {card_name} headset-head-unit-msbc"),
shlex.split(f"pactl set-card-profile {card_name} {get_bluetooth_profile(card_name, verbose)}"),
verbose=verbose,
)
try:
Expand Down