calvin.erfmann/librepods
1
0
Fork 0
mirror of https://github.com/kavishdevar/librepods.git synced 2026-01-31 23:29:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

implement conversational awareness in tray app

Browse Source
This commit is contained in:
Kavish Devar
2024-10-07 10:27:46 +05:30
parent 2aeb2b02a7
commit 36f10bbe84
11 changed files with 117 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
aln/Capabilites/__init__.py
View File

@@ -5,13 +5,14 @@ class NoiseCancellation:
ADAPTIVE = b"\x04"
class ConversationAwareness:
Off = b"\x02"
On = b"\x01"
OFF = b"\x02"
ON = b"\x01"
class Capabilites:
NOISE_CANCELLATION = b"\x0d"
CONVERSATION_AWARENESS = b"\x28"
CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY = b"\x01\x02"
EAR_DETECTION = b"\x06"
NoiseCancellation = NoiseCancellation
ConversationAwareness = ConversationAwareness

29
aln/Notifications/ANC.py
View File

@@ -20,31 +20,6 @@ class ANCNotification:
else:
return False
def setANC(self, data: bytes):
def setData(self, data: bytes):
self.status = data[7]
pass
def getANC(self, returnString: bool = False, fromInt: int = None):
if fromInt is not None:
fromInt = bytes([fromInt])
if fromInt == self.OFF:
return "Off"
elif fromInt == self.ON:
return "On"
elif fromInt == self.TRANSPARENCY:
return "Transparency"
elif fromInt == self.ADAPTIVE:
return "Adaptive"
pass
if returnString:
return self.status
else:
if self.status == self.OFF:
return "Off"
elif self.status == self.ON:
return "On"
elif self.status == self.TRANSPARENCY:
return "Transparency"
elif self.status == self.ADAPTIVE:
return "Adaptive"
pass
pass

19
aln/Notifications/ConversationalAwareness.py Normal file
View File

@@ -0,0 +1,19 @@
# 04 00 04 00 4b 00 02 00 01 [01/02/03/0b/09]
from ..enums import enums
class ConversationalAwarenessNotification:
NOTIFICATION_PREFIX = enums.CONVERSATION_AWARENESS_RECEIVE_PREFIX
def __init__(self):
pass
def isConversationalAwarenessData(self, data: bytes):
if len(data) != 10:
return False
if data.hex().startswith(self.NOTIFICATION_PREFIX.hex()):
return True
def setData(self, data: bytes):
self.status = data[9]
pass

2
aln/Notifications/EarDetection.py
View File

@@ -4,7 +4,7 @@ from typing import Literal
class EarDetectionNotification:
NOTIFICATION_BIT = Capabilites.EAR_DETECTION
NOTIFICATION_PREFIX = enums.SEND_PREFIX + NOTIFICATION_BIT
NOTIFICATION_PREFIX = enums.PREFIX + NOTIFICATION_BIT
IN_EAR = 0x00
OUT_OF_EAR = 0x01
def __init__(self):

8
aln/Notifications/Listener.py
View File

@@ -2,6 +2,7 @@ from bluetooth import BluetoothSocket
import threading
from .Battery import BatteryNotification
from .EarDetection import EarDetectionNotification
from .ConversationalAwareness import ConversationalAwarenessNotification
from .ANC import ANCNotification
import logging
@@ -11,6 +12,7 @@ class NotificationListener:
BATTERY_UPDATED = 0x01
ANC_UPDATED = 0x02
EAR_DETECTION_UPDATED = 0x03
CA_UPDATED = 0x04
UNKNOWN = 0x00
def __init__(self, socket: BluetoothSocket, callback: callable):
@@ -18,6 +20,7 @@ class NotificationListener:
self.BatteryNotification = BatteryNotification()
self.EarDetectionNotification = EarDetectionNotification()
self.ANCNotification = ANCNotification()
self.ConversationalAwarenessNotification = ConversationalAwarenessNotification()
self.callback = callback
pass
@@ -34,8 +37,11 @@ class NotificationListener:
self.EarDetectionNotification.setEarDetection(data)
self.callback(self.EAR_DETECTION_UPDATED, data)
if self.ANCNotification.isANCData(data):
self.ANCNotification.setANC(data)
self.ANCNotification.setData(data)
self.callback(self.ANC_UPDATED, data)
if self.ConversationalAwarenessNotification.isConversationalAwarenessData(data):
self.ConversationalAwarenessNotification.setData(data)
self.callback(self.CA_UPDATED, data)
else:
self.callback(self.UNKNOWN, data)
pass

4
aln/Notifications/__init__.py
View File

@@ -10,6 +10,7 @@ enums = enums()
class Notifications:
BATTERY_UPDATED = NotificationListener.BATTERY_UPDATED
ANC_UPDATED = NotificationListener.ANC_UPDATED
CA_UPDATED = NotificationListener.CA_UPDATED
EAR_DETECTION_UPDATED = NotificationListener.EAR_DETECTION_UPDATED
UNKNOWN = NotificationListener.UNKNOWN
def __init__(self, socket: bluetooth.BluetoothSocket, callback: callable):
@@ -18,12 +19,13 @@ class Notifications:
self.BatteryNotification = self.notificationListener.BatteryNotification
self.EarDetectionNotification = self.notificationListener.EarDetectionNotification
self.ANCNotification = self.notificationListener.ANCNotification
self.ConversationalAwarenessNotification = self.notificationListener.ConversationalAwarenessNotification
pass
def initialize(self):
try:
self.socket.send(enums.REQUEST_NOTIFICATIONS)
self.socket.send(enums.SET_SPECIFIC_FEATURES)
self.socket.send(enums.REQUEST_NOTIFICATIONS)
self.notificationListener.start()
except bluetooth.btcommon.BluetoothError as e:

8
aln/__init__.py
View File

@@ -23,6 +23,8 @@ class Connection:
self.notificationListener = self.notifications.notificationListener
self.BatteryNotification = self.notifications.BatteryNotification
self.ANCNotification = self.notifications.ANCNotification
self.EarDetectionNotification = self.notifications.EarDetectionNotification
self.ConversationalAwarenessNotification = self.notifications.ConversationalAwarenessNotification
self.notifications.initialize()
def send(self, data: bytes):
@@ -46,7 +48,11 @@ class Connection:
pass
elif notification_type == Notifications.ANC_UPDATED:
logging = logging.getLogger("ANC Status")
logging.debug(f'{self.notificationListener.ANCNotification.getANC()}')
logging.debug(f'{self.notificationListener.ANCNotification.status}')
pass
elif notification_type == Notifications.CA_UPDATED:
logging = logging.getLogger("Conversational Awareness Status")
logging.debug(f'{self.notificationListener.ConversationalAwarenessNotification.status}')
pass
elif notification_type == Notifications.UNKNOWN:
logging = logging.getLogger("Unknown Notification")

14
aln/enums.py
View File

@@ -4,19 +4,21 @@ class enums:
NOISE_CANCELLATION = Capabilites.NOISE_CANCELLATION
CONVERSATION_AWARENESS = Capabilites.CONVERSATION_AWARENESS
CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY = Capabilites.CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY
SEND_PREFIX = b'\x04\x00\x04\x00'
PREFIX = b'\x04\x00\x04\x00'
SETTINGS = b"\x09\x00"
SUFFIX = b'\x00\x00\x00'
NOTIFICATION_FILTER = b'\x0f'
SPECIFIC_FEATURES = b'\x4d'
SET_SPECIFIC_FEATURES = SEND_PREFIX + SPECIFIC_FEATURES + b"\x00\xff\x00\x00\x00\x00\x00\x00\x00"
REQUEST_NOTIFICATIONS = SEND_PREFIX + NOTIFICATION_FILTER + b"\x00\xff\xff\xff\xff"
SET_SPECIFIC_FEATURES = PREFIX + SPECIFIC_FEATURES + b"\x00\xff\x00\x00\x00\x00\x00\x00\x00"
REQUEST_NOTIFICATIONS = PREFIX + NOTIFICATION_FILTER + b"\x00\xff\xff\xff\xff"
HANDSHAKE = b"\x00\x00\x04\x00\x01\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00"
NOISE_CANCELLATION_PREFIX = SEND_PREFIX + SETTINGS + NOISE_CANCELLATION
NOISE_CANCELLATION_PREFIX = PREFIX + SETTINGS + NOISE_CANCELLATION
NOISE_CANCELLATION_OFF = NOISE_CANCELLATION_PREFIX + Capabilites.NoiseCancellation.OFF + SUFFIX
NOISE_CANCELLATION_ON = NOISE_CANCELLATION_PREFIX + Capabilites.NoiseCancellation.ON + SUFFIX
NOISE_CANCELLATION_TRANSPARENCY = NOISE_CANCELLATION_PREFIX + Capabilites.NoiseCancellation.TRANSPARENCY + SUFFIX
NOISE_CANCELLATION_ADAPTIVE = NOISE_CANCELLATION_PREFIX + Capabilites.NoiseCancellation.ADAPTIVE + SUFFIX
SET_CONVERSATION_AWARENESS_OFF = SEND_PREFIX + SETTINGS + CONVERSATION_AWARENESS + Capabilites.ConversationAwareness.Off + SUFFIX
SET_CONVERSATION_AWARENESS_ON = SEND_PREFIX + SETTINGS + CONVERSATION_AWARENESS + Capabilites.ConversationAwareness.On + SUFFIX
SET_CONVERSATION_AWARENESS_OFF = PREFIX + SETTINGS + CONVERSATION_AWARENESS + Capabilites.ConversationAwareness.OFF + SUFFIX
SET_CONVERSATION_AWARENESS_ON = PREFIX + SETTINGS + CONVERSATION_AWARENESS + Capabilites.ConversationAwareness.ON + SUFFIX
CONVERSATION_AWARENESS_RECEIVE_PREFIX = PREFIX + b"\x4b\x00\x02\00"

4
examples/daemon/read-data.py
View File

@@ -57,7 +57,9 @@ def read():
elif data["type"] == "ear_detection":
logging.info(f"\033[1;33mReceived ear detection status: {data['primary']} - {data['secondary']}\033[1;0m")
elif data["type"] == "anc":
logging.info(f"\033[1;33mReceived ANC status: {data['status']}\033[1;0m")
logging.info(f"\033[1;33mReceived ANC status: {data['mode']}\033[1;0m")
elif data["type"] == "ca":
logging.info(f"\033[1;33mReceived Conversational Awareness status: {data['status']}\033[1;0m")
elif data["type"] == "unknown":
logging.info(f"Received data: {data['data']}")
else:

54
examples/daemon/tray.py
View File

@@ -48,7 +48,7 @@ battery_status = {
"RIGHT": {"status": "Unknown", "level": 0},
"CASE": {"status": "Unknown", "level": 0}
}
anc_mode = 0
# Define a lock
battery_status_lock = threading.Lock()
@@ -71,7 +71,6 @@ class MediaController:
def isPlaying(self):
return "Playing" in subprocess.getoutput("playerctl --all-players status").strip()
def handlePlayPause(self, data):
primary_status = data[0]
secondary_status = data[1]
@@ -144,15 +143,56 @@ class MediaController:
self.earStatus = "Only one in"
return "Only one in"
# Function to get current sink volume
def get_current_volume():
result = subprocess.run(["pactl", "get-sink-volume", "@DEFAULT_SINK@"], capture_output=True, text=True)
volume_line = result.stdout.splitlines()[0]
volume_percent = int(volume_line.split()[4].strip('%'))
return volume_percent
# Function to set sink volume
def set_volume(percent):
subprocess.run(["pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{percent}%"])
initial_volume = get_current_volume()
# Handle conversational awareness
def handle_conversational_awareness(status):
if status < 1 or status > 9:
logging.error(f"Invalid status: {status}")
pass
global initial_volume
# Volume adjustment logic
if status == 1 or status == 2:
globals()["initial_volume"] = get_current_volume()
new_volume = max(0, min(int(initial_volume * 0.1), 100)) # Reduce to 10% for initial speaking
elif status == 3:
new_volume = max(0, min(int(initial_volume * 0.4), 100)) # Slightly increase to 40%
elif status == 6:
new_volume = max(0, min(int(initial_volume * 0.5), 100)) # Set volume to 50%
elif status >= 8:
new_volume = initial_volume # Fully restore volume
set_volume(new_volume)
logging.getLogger("Conversational Awareness").info(f"Volume set to {new_volume}% based on conversational awareness status: {status}")
# If status is 9, print conversation end message
if status == 9:
logging.getLogger("Conversational Awareness").info("Conversation ended. Restored volume to original level.")
class BatteryStatusUpdater(QObject):
battery_status_updated = pyqtSignal()
anc_mode_updated = pyqtSignal()
def __init__(self):
super().__init__()
self.media_controller = MediaController()
def listen_to_socket(self):
global battery_status
global anc_mode
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
client.connect(SOCKET_PATH)
while True:
@@ -168,6 +208,14 @@ class BatteryStatusUpdater(QObject):
elif response["type"] == "ear_detection":
self.media_controller.handlePlayPause([response['primary'], response['secondary']])
logging.debug(f"Received ear detection status: {response}")
elif response["type"] == "anc":
anc_mode = response["mode"]
self.anc_mode_updated.emit()
logging.debug(f"Received ANC status: {anc_mode}")
elif response["type"] == "ca":
ca_status = response["status"]
handle_conversational_awareness(ca_status)
logging.debug(f"Received CA status: {ca_status}")
except json.JSONDecodeError as e:
logging.warning(f"Error deserializing data: {e}")
except KeyError as e:

15
start-daemon.py
View File

@@ -69,14 +69,20 @@ def handle_client(connection, client_socket):
data: int = data
ancJSON = {
"type": "anc",
"status": data,
"mode": data,
}
data: str = JSONEncoder().encode(ancJSON)
elif notif_key == "notif_ca":
data: int = data
caJSON = {
"type": "ca",
"status": data,
}
data: str = JSONEncoder().encode(caJSON)
elif notif_key == "notif_unknown":
logging.debug(f"Unhandled notification type: {notif_key}")
logging.debug(f"Data: {data}")
data: str = JSONEncoder().encode({"type": "unknown", "data": data})
if not client_socket or not isinstance(client_socket, socket.socket):
logging.error("Invalid client socket")
break
@@ -183,6 +189,11 @@ def notification_handler(notification_type: int, data: bytes):
anc = connection.notificationListener.ANCNotification.status
globals()["notif_anc"] = anc
logger.debug(anc)
elif notification_type == Notifications.CA_UPDATED:
logger = logging.getLogger("Conversational Awareness Status")
ca = connection.notificationListener.ConversationalAwarenessNotification.status
globals()["notif_ca"] = ca
logger.debug(ca)
elif notification_type == Notifications.UNKNOWN:
logger = logging.getLogger("Unknown Notification")
hex_data = ' '.join(f'{byte:02x}' for byte in data)