diff --git a/linux.old/README.md b/linux.old/README.md deleted file mode 100644 index 3e119a8..0000000 --- a/linux.old/README.md +++ /dev/null @@ -1,99 +0,0 @@ -# ALN - AirPods like Normal (Linux Only) - -### Check out the packet definitions at [AAP Definitions](/AAP%20Definitions.md) - -## Currently supported device(s) -- AirPods Pro 2 - -### 1. Install the required packages - -```bash -sudo apt install python3 python3-pip -pip3 install pybluez -``` - -### 2. Clone the repository - -```bash -git clone https://github.com/kavishdevar/aln.git -cd aln -``` - -### 3. Preprare -Pair your AirPods with your machine before running this script! -> **Note:** DO NOT FORGET TO EDIT THE `AIRPODS_MAC` VARIABLE IN EXAMPLE SCRIPTS WITH YOUR AIRPODS MAC ADDRESS BEFORE RUNNING THEM! - -# Versions - -## Non-Daemon based -This version is the most polished version of the script. It can do the following: -- fetch the battery percentage, -- fetch in-ear status (but not actually controlling the media with that information). -- control ANC modes -```bash -python3 examples/logger-and-anc.py -``` - -## As a daemon (using a UNIX socket) -![Daemon Log Screenshot](imgs/daemon-log.png) -If you want to run a deamon for multiple programs to read/write airpods data, you can use the `airpods_daemon.py` script. -- This creates a standard UNIX socket at `/tmp/airpods_daemon.sock` and listens for commands -- and sends battery/in-ear info -You can run it as follows: - -```bash -python3 airpods_daemon.py -``` - -## Interacting with the daemon -![Set ANC Screenshot](imgs/set-anc.png) -- Sending data to the daemon -You can send data to the daemon using the `set-anc.py` script. Since it's a standard UNIX socket, you can send data to it using any programming language that supports UNIX sockets. - -This package includes a demo script that sends a command to turn off the ANC. You can run it as follows: - -```bash -python3 examples/daemon/set-anc.py -``` - -- Reading data from the daemon -![Read Data Screenshot](imgs/read-data.png) -You can listen to the daemon's output by running the `read-data.py` script. This script listens to the UNIX socket and prints the data it receives. Currenty, it recognizes the battery percentage and the in-ear status and dumps the rest of the data to the terminal. - -```bash -python3 examples/daemon/read-data.py -``` - -- Controlling the media with the in-ear status (and get battery status) - -![Ear Detection Screenshot](imgs/ear-detection.png) -This script is basically the standalone script, but interacts with the UNIX socket created by the daemon instead. It can control the media with the in-ear status and remove the device as an audio sink when the AirPods are not in your ears. - -```bash -python3 examples/daemon/ear-detection.py -``` - -- App Indicator/Tray Icon - -![Tray Icon Hover Screenshot](imgs/tray-icon-hover.png) -![Tray Icon Menu Screenshot](imgs/tray-icon-menu.png) -This script is a simple tray icon that shows the battery percentage and set ANC modes. It can also control the media with the in-ear status. -> Note: This script uses QT. - -```bash -python3 examples/daemon/tray.py -``` - -## Standalone version (without module dependency, mainly for testing, and reverse engineering purposes) -- Controlling the media with the in-ear status. -- Remove the device as an audio sink when the AirPods are not in your ears. -- Try to connect with the AirPods if media is playing and the AirPods are not connected. -- Control ANC modes. - -```bash -python3 examples/standalone.py -``` - -# License - -See [README](/README.md). diff --git a/linux.old/aln/AirPods/Pro2.py b/linux.old/aln/AirPods/Pro2.py deleted file mode 100644 index aa6833c..0000000 --- a/linux.old/aln/AirPods/Pro2.py +++ /dev/null @@ -1,30 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from ..Capabilites import Capabilites -class Pro2: - def __init__(self): - self.name = 'AirPods Pro 2' - self.capabilites = { - Capabilites.NOISE_CANCELLATION: [ - Capabilites.NoiseCancellation.OFF, - Capabilites.NoiseCancellation.ON, - Capabilites.NoiseCancellation.TRANSPARENCY, - Capabilites.NoiseCancellation.ADAPTIVE, - ], - Capabilites.CONVERSATION_AWARENESS: True, - Capabilites.CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY: True - } \ No newline at end of file diff --git a/linux.old/aln/AirPods/__init__.py b/linux.old/aln/AirPods/__init__.py deleted file mode 100644 index d0a8855..0000000 --- a/linux.old/aln/AirPods/__init__.py +++ /dev/null @@ -1,20 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - - -from . import Pro2 - -Pro2 = Pro2.Pro2 \ No newline at end of file diff --git a/linux.old/aln/Capabilites/__init__.py b/linux.old/aln/Capabilites/__init__.py deleted file mode 100644 index f913904..0000000 --- a/linux.old/aln/Capabilites/__init__.py +++ /dev/null @@ -1,34 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -class NoiseCancellation: - OFF = b"\x01" - ON = b"\x02" - TRANSPARENCY = b"\x03" - ADAPTIVE = b"\x04" - -class ConversationAwareness: - 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 \ No newline at end of file diff --git a/linux.old/aln/Notifications/ANC.py b/linux.old/aln/Notifications/ANC.py deleted file mode 100644 index 72896be..0000000 --- a/linux.old/aln/Notifications/ANC.py +++ /dev/null @@ -1,41 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from ..enums import enums -from ..Capabilites import Capabilites -class ANCNotification: - NOTIFICATION_PREFIX = enums.NOISE_CANCELLATION_PREFIX - OFF = Capabilites.NoiseCancellation.OFF - ON = Capabilites.NoiseCancellation.ON - TRANSPARENCY = Capabilites.NoiseCancellation.TRANSPARENCY - ADAPTIVE = Capabilites.NoiseCancellation.ADAPTIVE - - def __init__(self): - pass - - def isANCData(self, data: bytes): - # 04 00 04 00 09 00 0D 01 00 00 00 - if len(data) != 11: - return False - - if data.hex().startswith(self.NOTIFICATION_PREFIX.hex()): - return True - else: - return False - - def setData(self, data: bytes): - self.status = data[7] - pass \ No newline at end of file diff --git a/linux.old/aln/Notifications/Battery.py b/linux.old/aln/Notifications/Battery.py deleted file mode 100644 index c9cffda..0000000 --- a/linux.old/aln/Notifications/Battery.py +++ /dev/null @@ -1,80 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -class BatteryComponent: - LEFT = 4 - RIGHT = 2 - CASE = 8 - pass -class BatteryStatus: - CHARGING = 1 - NOT_CHARGING = 2 - DISCONNECTED = 4 - pass - -class Battery: - - def get_name(self, cls, value): - for key, val in cls.__dict__.items(): - if val == value: - return key - return None - - def get_component(self): - return self.get_name(BatteryComponent, self.component) - - def get_level(self): - return self.level - - def get_status(self): - return self.get_name(BatteryStatus, self.status) - - def __init__(self, component: int, level: int, status: int): - self.component = component - self.level = level - self.status = status - pass - -class BatteryNotification: - def __init__(self): - self.first = Battery(BatteryComponent.LEFT, 0, BatteryStatus.DISCONNECTED) - self.second = Battery(BatteryComponent.RIGHT, 0, BatteryStatus.DISCONNECTED) - self.case = Battery(BatteryComponent.CASE, 0, BatteryStatus.DISCONNECTED) - pass - - def isBatteryData(self, data): - if len(data) != 22: - return False - if data[0] == 0x04 and data[1] == 0x00 and data[2] == 0x04 and data[3] == 0x00 and data[4] == 0x04 and data[5] == 0x00: - return True - else: - return False - def setBattery(self, data): - self.count = data[6] - self.first = Battery(data[7], data[9], data[10]) - self.second = Battery(data[12], data[14], data[15]) - self.case = Battery(data[17], data[19], data[20]) - pass - - def getBattery(self): - if self.first.component == BatteryComponent.LEFT: - self.left = self.first - self.right = self.second - else: - self.left = self.second - self.right = self.first - self.case = self.case - return [self.left, self.right, self.case] \ No newline at end of file diff --git a/linux.old/aln/Notifications/ConversationalAwareness.py b/linux.old/aln/Notifications/ConversationalAwareness.py deleted file mode 100644 index 55609f2..0000000 --- a/linux.old/aln/Notifications/ConversationalAwareness.py +++ /dev/null @@ -1,33 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -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 \ No newline at end of file diff --git a/linux.old/aln/Notifications/EarDetection.py b/linux.old/aln/Notifications/EarDetection.py deleted file mode 100644 index d72bb90..0000000 --- a/linux.old/aln/Notifications/EarDetection.py +++ /dev/null @@ -1,42 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from ..Capabilites import Capabilites -from ..enums import enums -from typing import Literal - -class EarDetectionNotification: - NOTIFICATION_BIT = Capabilites.EAR_DETECTION - NOTIFICATION_PREFIX = enums.PREFIX + NOTIFICATION_BIT - IN_EAR = 0x00 - OUT_OF_EAR = 0x01 - def __init__(self): - pass - - def isEarDetectionData(self, data: bytes): - if len(data) != 8: - return False - if data.hex().startswith(self.NOTIFICATION_PREFIX.hex()): - return True - - def setEarDetection(self, data: bytes): - self.first = data[6] - self.second = data[7] - - def getEarDetection(self): - return [self.first, self.second] - pass - \ No newline at end of file diff --git a/linux.old/aln/Notifications/Listener.py b/linux.old/aln/Notifications/Listener.py deleted file mode 100644 index 902ffc9..0000000 --- a/linux.old/aln/Notifications/Listener.py +++ /dev/null @@ -1,73 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . -from bluetooth import BluetoothSocket -import threading -from .Battery import BatteryNotification -from .EarDetection import EarDetectionNotification -from .ConversationalAwareness import ConversationalAwarenessNotification -from .ANC import ANCNotification -import logging - -logging = logging.getLogger(__name__) - -class NotificationListener: - BATTERY_UPDATED = 0x01 - ANC_UPDATED = 0x02 - EAR_DETECTION_UPDATED = 0x03 - CA_UPDATED = 0x04 - UNKNOWN = 0x00 - - def __init__(self, socket: BluetoothSocket, callback: callable): - self.socket = socket - self.BatteryNotification = BatteryNotification() - self.EarDetectionNotification = EarDetectionNotification() - self.ANCNotification = ANCNotification() - self.ConversationalAwarenessNotification = ConversationalAwarenessNotification() - self.callback = callback - pass - - def __start(self): - while True: - data = self.socket.recv(1024) - if len(data) == 0: - break - if self.BatteryNotification.isBatteryData(data): - self.BatteryNotification.setBattery(data) - self.callback(self.BATTERY_UPDATED, data) - pass - if self.EarDetectionNotification.isEarDetectionData(data): - self.EarDetectionNotification.setEarDetection(data) - self.callback(self.EAR_DETECTION_UPDATED, data) - if self.ANCNotification.isANCData(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 - pass - pass - - def start(self): - threading.Thread(target=self.__start).start() - pass - - def stop(self): - self.socket.close() - pass - \ No newline at end of file diff --git a/linux.old/aln/Notifications/__init__.py b/linux.old/aln/Notifications/__init__.py deleted file mode 100644 index 2de182b..0000000 --- a/linux.old/aln/Notifications/__init__.py +++ /dev/null @@ -1,56 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from .Listener import NotificationListener -from ..enums import enums -import bluetooth -import logging - -logging = logging.getLogger(__name__) - -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): - self.socket = socket - self.notificationListener = NotificationListener(self.socket, callback) - 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.SET_SPECIFIC_FEATURES) - self.socket.send(enums.REQUEST_NOTIFICATIONS) - self.notificationListener.start() - - except bluetooth.btcommon.BluetoothError as e: - logging.error(f'Failed to send data to {self.mac_address}: {e}') - return False - return True - - def __del__(self): - self.notificationListener.stop() - self.socket.close() - pass - pass \ No newline at end of file diff --git a/linux.old/aln/__init__.py b/linux.old/aln/__init__.py deleted file mode 100644 index f5a319d..0000000 --- a/linux.old/aln/__init__.py +++ /dev/null @@ -1,87 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from .Notifications import Notifications -import bluetooth -import logging - -logging = logging.getLogger("Connection Handler") - -class Connection: - def __init__(self, mac_address: str): - self.mac_address = mac_address - self.socket = bluetooth.BluetoothSocket(bluetooth.L2CAP) - def connect(self): - try: - self.socket.connect((self.mac_address, 0x1001)) - except bluetooth.btcommon.BluetoothError as e: - logging.error(f'Failed to connect to {self.mac_address}: {e}') - return False - return True - - def initialize_notifications(self, callback: callable = None): - if callback is None: - callback = self.notification_callback - self.notifications = Notifications(self.socket, callback) - 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): - try: - logging.info(f"Sending data to {self.mac_address}: {data}") - self.socket.send(data) - except bluetooth.btcommon.BluetoothError as e: - logging.error(f'Failed to send data to {self.mac_address}: {e}') - return False - return True - - def notification_callback(self, notification_type: int, data: bytes): - import logging - if notification_type == Notifications.BATTERY_UPDATED: - logging = logging.getLogger("Battery Status") - for i in self.notificationListener.BatteryNotification.getBattery(): - logging.debug(f'{i.get_component()} - {i.get_status()}: {i.get_level()}') - pass - elif notification_type == Notifications.EAR_DETECTION_UPDATED: - logging = logging.getLogger("In-Ear Status") - logging.debug(f'{self.notificationListener.EarDetectionNotification.getEarDetection()}') - pass - elif notification_type == Notifications.ANC_UPDATED: - logging = logging.getLogger("ANC Status") - 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") - hex_data = ' '.join(f'{byte:02x}' for byte in data) - logging.debug(f'{hex_data}') - pass - pass - - def disconnect(self): - self.socket.close() - pass - - def __del__(self): - self.socket.close() - pass diff --git a/linux.old/aln/__main__.py b/linux.old/aln/__main__.py deleted file mode 100644 index bf15225..0000000 --- a/linux.old/aln/__main__.py +++ /dev/null @@ -1,265 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import logging.handlers -import socket -import threading -import signal -import sys -import logging -from . import Connection -import enums -from aln.Notifications import Notifications -from aln.Notifications.Battery import Battery -import os -import bluetooth -from aln.enums import enums -connection = None - -SOCKET_PATH = '/tmp/airpods_daemon.sock' -LOG_FOLDER = '.' -LOG_FILE = os.path.join(LOG_FOLDER, 'airpods_daemon.log') - -# Global flag to control the server loop -running = True - -# Configure logging to write to a file -# logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG, format='%(asctime)s %(levelname)s : %(message)s') - -# RotatingFileHandler - -formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') -handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes=2**20) -handler.setFormatter(formatter) -handler.setLevel(logging.DEBUG) - -l = logging.getLogger() -l.setLevel(logging.DEBUG) -l.addHandler(handler) - -from json import JSONEncoder - -def handle_client(connection, client_socket): - """Handle client requests by forwarding all received data to aln.Connection, send data back to the client.""" - - def send_status(): - while running: - try: - for notif_key in list(globals().keys()): - if notif_key.startswith("notif_"): - data = globals().get(notif_key) - if data: - if notif_key == "notif_battery": - data: list[Battery] = data - batteryJSON = {"type": "battery"} - for i in data: - batteryJSON[i.get_component()] = { - "status": i.get_status(), - "level": i.get_level() - } - data: str = JSONEncoder().encode(batteryJSON) - - elif notif_key == "notif_ear_detection": - # noinspection PyTypeChecker - data: list[int] = data - earDetectionJSON = { - "type": "ear_detection", - "primary": data[0], - "secondary": data[1] - } - data: str = JSONEncoder().encode(earDetectionJSON) - elif notif_key == "notif_anc": - data: int = data - ancJSON = { - "type": "anc", - "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 - logging.info(f'Sending {notif_key} status: {data}') - client_socket.sendall(data.encode('utf-8')) - logging.info(f'Sent {notif_key} status: {data}') - globals()[notif_key] = None - except socket.error as e: - logging.error(f"Socket error sending status: {e}") - break - except Exception as e: - logging.error(f"Error sending status: {e}") - break - - def receive_commands(): - while running: - try: - data = client_socket.recv(1024) - if not data: - break - logging.info(f'Received command: {data}') - connection.send(data) - except Exception as e: - logging.error(f"Error receiving command: {e}") - break - - # Start two threads to handle sending and receiving data - send_thread = threading.Thread(target=send_status) - send_thread.start() - receive_thread = threading.Thread(target=receive_commands) - receive_thread.start() - - send_thread.join() - receive_thread.join() - - client_socket.close() - logging.info("Client socket closed") - -def start_socket_server(connection): - """Start a UNIX domain socket server.""" - global running - - # Set up the socket - server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - try: - server_socket.bind(SOCKET_PATH) - except OSError: - logging.error(f"Socket already in use or unavailable: {SOCKET_PATH}") - sys.exit(1) - - server_socket.listen(1) - logging.info(f"Socket server listening on {SOCKET_PATH}") - - while running: - try: - client_socket, _ = server_socket.accept() - logging.info("Client connected") - - # Handle the client connection in a separate thread - client_thread = threading.Thread(target=handle_client, args=(connection, client_socket)) - client_thread.start() - except Exception as e: - logging.error(f"Error accepting connection: {e}") - - # Close the server socket when stopped - server_socket.close() - logging.info("Socket server stopped") - -def stop_daemon(_, __): - """Signal handler to stop the daemon.""" - global running - logging.info("Received termination signal. Stopping daemon...") - running = False # Set running flag to False to stop the loop - - # Close the socket gracefully by removing the file path - try: - socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).connect(SOCKET_PATH) - except socket.error: - pass - finally: - # Remove the socket file - if os.path.exists(SOCKET_PATH): - os.remove(SOCKET_PATH) - - sys.exit(0) - -def notification_handler(notification_type: int, data: bytes): - global connection - - logging.debug(f"Received notification: {notification_type}") - if notification_type == Notifications.BATTERY_UPDATED: - logger = logging.getLogger("Battery Status") - battery = connection.notificationListener.BatteryNotification.getBattery() - globals()["notif_battery"] = battery - for i in battery: - logger.debug(f'{i.get_component()} - {i.get_status()}: {i.get_level()}') - elif notification_type == Notifications.EAR_DETECTION_UPDATED: - logger = logging.getLogger("In-Ear Status") - earDetection = connection.notificationListener.EarDetectionNotification.getEarDetection() - globals()["notif_ear_detection"] = earDetection - logger.debug(earDetection) - elif notification_type == Notifications.ANC_UPDATED: - logger = logging.getLogger("ANC Status") - 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) - globals()["notif_unknown"] = hex_data - logger.debug(hex_data) - -def main(): - global running - logging.info("Starting AirPods daemon") - - connection = Connection(mac) - globals()['connection'] = connection - - # Connect to the AirPods and send the handshake - try: - connection.connect() - except bluetooth.btcommon.BluetoothError as e: - logging.error(f"Failed to connect to {mac}: {e}") - sys.exit(1) - - connection.send(enums.HANDSHAKE) - logging.info("Handshake sent") - - connection.initialize_notifications(notification_handler) - - # Start the socket server to listen for client connections - start_socket_server(connection) - - # Set up signal handlers to handle termination signals - signal.signal(signal.SIGINT, stop_daemon) # Handle Ctrl+C - signal.signal(signal.SIGTERM, stop_daemon) # Handle kill signal - -if __name__ == "__main__": - # Daemonize the process - if os.fork(): - sys.exit() - - os.setsid() - - if os.fork(): - sys.exit() - - sys.stdout.flush() - sys.stderr.flush() - - with open('/dev/null', 'r') as devnull: - os.dup2(devnull.fileno(), sys.stdin.fileno()) - - with open(LOG_FILE, 'a+') as logfile: - os.dup2(logfile.fileno(), sys.stdout.fileno()) - os.dup2(logfile.fileno(), sys.stderr.fileno()) - - main() diff --git a/linux.old/aln/enums.py b/linux.old/aln/enums.py deleted file mode 100644 index 8fc49b6..0000000 --- a/linux.old/aln/enums.py +++ /dev/null @@ -1,43 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from .Capabilites import Capabilites - -class enums: - NOISE_CANCELLATION = Capabilites.NOISE_CANCELLATION - CONVERSATION_AWARENESS = Capabilites.CONVERSATION_AWARENESS - CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY = Capabilites.CUSTOMIZABLE_ADAPTIVE_TRANSPARENCY - 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 = 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 = 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 = 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" - - class longPress: - PREFIX = b'\x04\x00\x04\x00\x09\x00\x1A' \ No newline at end of file diff --git a/linux.old/aln/listener.py b/linux.old/aln/listener.py deleted file mode 100644 index f526b3c..0000000 --- a/linux.old/aln/listener.py +++ /dev/null @@ -1,47 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import bluetooth -import time - -class BluetoothListener: - def __init__(self): - self.connected_devices = set() - - def scan_devices(self): - nearby_devices = bluetooth.discover_devices(lookup_names=True, lookup_class=True, device_id=-1, duration=8, flush_cache=True) - return nearby_devices - - def start_listening(self): - print("Listening for Bluetooth devices") - while True: - nearby_devices = self.scan_devices() - current_devices = set() - - for addr, name, device_class in nearby_devices: - current_devices.add(addr) - if addr not in self.connected_devices: - print(f"Device connected: {name} [{addr}]") - - for addr in self.connected_devices - current_devices: - print(f"Device disconnected: [{addr}]") - - self.connected_devices = current_devices - time.sleep(5) - -if __name__ == "__main__": - listener = BluetoothListener() - listener.start_listening() \ No newline at end of file diff --git a/linux.old/examples/daemon/ear-detection.py b/linux.old/examples/daemon/ear-detection.py deleted file mode 100644 index 4fb2ac3..0000000 --- a/linux.old/examples/daemon/ear-detection.py +++ /dev/null @@ -1,180 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import socket -import json -import subprocess -from aln.Notifications import Battery -import threading -import time -import os -import logging - -class CustomFormatter(logging.Formatter): - # Define color codes for different log levels - COLORS = { - logging.DEBUG: "\033[48;5;240;38;5;15m%s\033[1;0m", # Grey background, white bold text - logging.INFO: "\033[48;5;34;38;5;15m%s\033[1;0m", # Green background, white bold text - logging.WARNING: "\033[1;48;5;214;38;5;0m%s\033[1;0m", # Orange background, black bold text - logging.ERROR: "\033[1;48;5;202;38;5;15m%s\033[1;0m", # Orange-red background, white bold text - logging.CRITICAL: "\033[1;48;5;196;38;5;15m%s\033[1;0m", # Pure red background, white bold text - } - - def format(self, record): - # Apply color to the level name - levelname = self.COLORS.get(record.levelno, "%s") % record.levelname.ljust(8) - record.levelname = levelname - - # Format the message - formatted_message = super().format(record) - - return formatted_message - -# Custom formatter with fixed width for level name -formatter = CustomFormatter('\033[2;37m%(asctime)s\033[1;0m - %(levelname)s - %(message)s') - -logging.basicConfig(level=logging.DEBUG) - -# Set the custom formatter for the root logger -logging.getLogger().handlers[0].setFormatter(formatter) - -SOCKET_PATH = "/tmp/airpods_daemon.sock" - -class MediaController: - def __init__(self): - self.earStatus = "Both out" - self.wasMusicPlayingInSingle = False - self.wasMusicPlayingInBoth = False - self.firstEarOutTime = 0 - self.stop_thread_event = threading.Event() - - def playMusic(self): - logging.info("Playing music") - subprocess.call(("playerctl", "play")) - - def pauseMusic(self): - logging.info("Pausing music") - subprocess.call(("playerctl", "--all-players", "pause")) - - 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] - - logging.debug(f"Handling play/pause with data: {data}, previousStatus: {self.earStatus}, wasMusicPlaying: {self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth}") - - def delayed_action(s): - if not self.stop_thread_event.is_set(): - if self.wasMusicPlayingInSingle: - self.playMusic() - self.wasMusicPlayingInBoth = False - elif self.wasMusicPlayingInBoth or s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - - if primary_status and secondary_status: - if self.earStatus != "Both out": - s = self.isPlaying() - if s: - self.pauseMusic() - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B off") - logging.info("Setting profile to off") - if self.earStatus == "Only one in": - if self.firstEarOutTime != 0 and time.time() - self.firstEarOutTime < 0.3: - self.wasMusicPlayingInSingle = True - self.wasMusicPlayingInBoth = True - self.stop_thread_event.set() - else: - if s: - self.wasMusicPlayingInSingle = True - else: - self.wasMusicPlayingInSingle = False - elif self.earStatus == "Both in": - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - else: - self.wasMusicPlayingInSingle = False - self.earStatus = "Both out" - return "Both out" - elif not primary_status and not secondary_status: - if self.earStatus != "Both in": - if self.earStatus == "Both out": - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp-sink") - logging.info("Setting profile to a2dp-sink") - elif self.earStatus == "Only one in": - self.stop_thread_event.set() - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - if self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth: - self.playMusic() - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - self.earStatus = "Both in" - return "Both in" - elif (primary_status and not secondary_status) or (not primary_status and secondary_status): - if self.earStatus != "Only one in": - self.stop_thread_event.clear() - s = self.isPlaying() - if s: - self.pauseMusic() - delayed_thread = threading.Timer(0.3, delayed_action, args=[s]) - delayed_thread.start() - self.firstEarOutTime = time.time() - if self.earStatus == "Both out": - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp-sink") - logging.info("Setting profile to a2dp-sink") - self.earStatus = "Only one in" - return "Only one in" - -def read(): - """Send a command to the daemon via UNIX domain socket.""" - client_socket = None - try: - # Create a socket connection to the daemon - client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - logging.info("Connecting to daemon...") - client_socket.connect(SOCKET_PATH) - - media_controller = MediaController() - - # Receive data - while True: - d = client_socket.recv(1024) - if d: - try: - data: dict = json.loads(d.decode('utf-8')) - if data["type"] == "ear_detection": - media_controller.handlePlayPause([data['primary'], data['secondary']]) - except json.JSONDecodeError as e: - # logging.error(f"Error deserializing data: {e}") - pass - else: - break - - except Exception as e: - logging.error(f"Error communicating with daemon: {e}") - finally: - if client_socket: - client_socket.close() - logging.warning("Socket closed") - -if __name__ == "__main__": - read() diff --git a/linux.old/examples/daemon/read-data.py b/linux.old/examples/daemon/read-data.py deleted file mode 100644 index d0660f7..0000000 --- a/linux.old/examples/daemon/read-data.py +++ /dev/null @@ -1,103 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . -import socket -import json -import logging -from aln.Notifications.ANC import ANCNotification -SOCKET_PATH = "/tmp/airpods_daemon.sock" - -import logging - -class CustomFormatter(logging.Formatter): - # Define color codes for different log levels - COLORS = { - logging.DEBUG: "\033[48;5;240;38;5;15m%s\033[1;0m", # Grey background, white bold text - logging.INFO: "\033[48;5;34;38;5;15m%s\033[1;0m", # Green background, white bold text - logging.WARNING: "\033[1;48;5;214;38;5;0m%s\033[1;0m", # Orange background, black bold text - logging.ERROR: "\033[1;48;5;202;38;5;15m%s\033[1;0m", # Orange-red background, white bold text - logging.CRITICAL: "\033[1;48;5;196;38;5;15m%s\033[1;0m", # Pure red background, white bold text - } - - def format(self, record): - # Apply color to the level name - levelname = self.COLORS.get(record.levelno, "%s") % record.levelname.ljust(8) - record.levelname = levelname - - # Format the message - formatted_message = super().format(record) - - return formatted_message - -# Custom formatter with fixed width for level name -formatter = CustomFormatter('\033[2;37m%(asctime)s\033[1;0m - %(levelname)s - %(message)s') - -logging.basicConfig(level=logging.DEBUG) - -# Set the custom formatter for the root logger -logging.getLogger().handlers[0].setFormatter(formatter) - -def read(): - """Send a command to the daemon via UNIX domain socket.""" - client_socket = None - try: - # Create a socket connection to the daemon - client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - logging.info("Connecting to daemon...") - client_socket.connect(SOCKET_PATH) - - # Receive data - while True: - d = client_socket.recv(1024) - if d: - try: - data = json.loads(d.decode('utf-8')) - if isinstance(data, dict): - if data["type"] == "battery": - for b, battery_data in data.items(): - if b != "type": # Skip the "type" key - logging.info(f"\033[1;33mReceived battery status: {b} - {battery_data['status']} - {battery_data['level']}\033[1;0m") - 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['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: - logging.info(f"Received data: {data}") - else: - logging.error("Received data is not a dictionary") - except json.JSONDecodeError as e: - # logging.warning(f"Error deserializing data: {e}") - # logging.warning(f"raw data: {d}") - pass - except KeyError as e: - logging.error(f"KeyError: {e} in data: {data}") - except TypeError as e: - logging.error(f"TypeError: {e} in data: {data}") - else: - break - - except Exception as e: - logging.critical(f"Error communicating with daemon: {e}") - finally: - if client_socket: - client_socket.close() - logging.warning("Socket closed") - -if __name__ == "__main__": - read() \ No newline at end of file diff --git a/linux.old/examples/daemon/set-anc.py b/linux.old/examples/daemon/set-anc.py deleted file mode 100644 index b04b0e8..0000000 --- a/linux.old/examples/daemon/set-anc.py +++ /dev/null @@ -1,89 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import socket -import argparse -from aln import enums -import logging - -class CustomFormatter(logging.Formatter): - # Define color codes for different log levels - COLORS = { - logging.DEBUG: "\033[48;5;240;38;5;15m%s\033[1;0m", # Grey background, white bold text - logging.INFO: "\033[48;5;34;38;5;15m%s\033[1;0m", # Green background, white bold text - logging.WARNING: "\033[1;48;5;214;38;5;0m%s\033[1;0m", # Orange background, black bold text - logging.ERROR: "\033[1;48;5;202;38;5;15m%s\033[1;0m", # Orange-red background, white bold text - logging.CRITICAL: "\033[1;48;5;196;38;5;15m%s\033[1;0m", # Pure red background, white bold text - } - - def format(self, record): - # Apply color to the level name - levelname = self.COLORS.get(record.levelno, "%s") % record.levelname.ljust(8) - record.levelname = levelname - - # Format the message - formatted_message = super().format(record) - - return formatted_message - -# Custom formatter with fixed width for level name -formatter = CustomFormatter('\033[2;37m%(asctime)s\033[1;0m - %(levelname)s - %(message)s') - -logging.basicConfig(level=logging.DEBUG) - -# Set the custom formatter for the root logger -logging.getLogger().handlers[0].setFormatter(formatter) - - -SOCKET_PATH = "/tmp/airpods_daemon.sock" - -def send_command(command): - """Send a command to the daemon via UNIX domain socket.""" - try: - # Create a socket connection to the daemon - logging.info("Connecting to daemon...") - client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - client_socket.connect(SOCKET_PATH) - logging.info("Connected to daemon") - - # Send the command - client_socket.sendall(command) - logging.info(f"Sent command: {command}") - - # Close the connection - client_socket.close() - logging.info("Socket closed") - except Exception as e: - logging.error(f"Error communicating with daemon: {e}") - -def parse_arguments(): - parser = argparse.ArgumentParser(description="Set AirPods ANC mode.") - parser.add_argument("mode", choices=["off", "on", "transparency", "adaptive", "1", "2", "3", "4"], help="ANC mode to set") - return parser.parse_args() - -if __name__ == "__main__": - args = parse_arguments() - - if args.mode == "off" or args.mode == "1": - command = enums.NOISE_CANCELLATION_OFF - elif args.mode == "on" or args.mode == "2": - command = enums.NOISE_CANCELLATION_ON - elif args.mode == "transparency" or args.mode == "3": - command = enums.NOISE_CANCELLATION_TRANSPARENCY - elif args.mode == "adaptive" or args.mode == "4": - command = enums.NOISE_CANCELLATION_ADAPTIVE - - send_command(command) diff --git a/linux.old/examples/daemon/tray.py b/linux.old/examples/daemon/tray.py deleted file mode 100644 index 74ccd99..0000000 --- a/linux.old/examples/daemon/tray.py +++ /dev/null @@ -1,333 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import sys -import socket -import json -import signal -import threading -from PyQt5.QtWidgets import QApplication, QSystemTrayIcon, QMenu, QAction, QMessageBox -from PyQt5.QtGui import QIcon -from PyQt5.QtCore import QObject, pyqtSignal -import logging -import subprocess -import time -import os - - -class CustomFormatter(logging.Formatter): - # Define color codes for different log levels - COLORS = { - logging.DEBUG: "\033[48;5;240;38;5;15m%s\033[1;0m", # Grey background, white bold text - logging.INFO: "\033[48;5;34;38;5;15m%s\033[1;0m", # Green background, white bold text - logging.WARNING: "\033[1;48;5;214;38;5;0m%s\033[1;0m", # Orange background, black bold text - logging.ERROR: "\033[1;48;5;202;38;5;15m%s\033[1;0m", # Orange-red background, white bold text - logging.CRITICAL: "\033[1;48;5;196;38;5;15m%s\033[1;0m", # Pure red background, white bold text - } - - def format(self, record): - # Apply color to the level name - levelname = self.COLORS.get(record.levelno, "%s") % record.levelname.ljust(8) - record.levelname = levelname - - # Format the message - formatted_message = super().format(record) - - return formatted_message - -# Custom formatter with fixed width for level name -formatter = CustomFormatter('\033[2;37m%(asctime)s\033[1;0m - %(levelname)s - %(message)s') - -logging.basicConfig(level=logging.DEBUG) - -# Set the custom formatter for the root logger -logging.getLogger().handlers[0].setFormatter(formatter) - -SOCKET_PATH = "/tmp/airpods_daemon.sock" - -# Initialize battery_status at the module level -battery_status = { - "LEFT": {"status": "Unknown", "level": 0}, - "RIGHT": {"status": "Unknown", "level": 0}, - "CASE": {"status": "Unknown", "level": 0} -} -anc_mode = 0 -# Define a lock -battery_status_lock = threading.Lock() - -class MediaController: - def __init__(self): - self.earStatus = "Both out" - self.wasMusicPlayingInSingle = False - self.wasMusicPlayingInBoth = False - self.firstEarOutTime = 0 - self.stop_thread_event = threading.Event() - - def playMusic(self): - logging.info("Playing music") - subprocess.call(("playerctl", "play")) - - def pauseMusic(self): - logging.info("Pausing music") - subprocess.call(("playerctl", "--all-players", "pause")) - - 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] - - logging.debug(f"Handling play/pause with data: {data}, previousStatus: {self.earStatus}, wasMusicPlaying: {self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth}") - - def delayed_action(s): - if not self.stop_thread_event.is_set(): - if self.wasMusicPlayingInSingle: - self.playMusic() - self.wasMusicPlayingInBoth = False - elif self.wasMusicPlayingInBoth or s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - - if primary_status and secondary_status: - if self.earStatus != "Both out": - s = self.isPlaying() - if s: - self.pauseMusic() - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B off") - logging.info("Setting profile to off") - if self.earStatus == "Only one in": - if self.firstEarOutTime != 0 and time.time() - self.firstEarOutTime < 0.3: - self.wasMusicPlayingInSingle = True - self.wasMusicPlayingInBoth = True - self.stop_thread_event.set() - else: - if s: - self.wasMusicPlayingInSingle = True - else: - self.wasMusicPlayingInSingle = False - elif self.earStatus == "Both in": - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - else: - self.wasMusicPlayingInSingle = False - self.earStatus = "Both out" - return "Both out" - elif not primary_status and not secondary_status: - if self.earStatus != "Both in": - if self.earStatus == "Both out": - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp-sink") - logging.info("Setting profile to a2dp-sink") - elif self.earStatus == "Only one in": - self.stop_thread_event.set() - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - if self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth: - self.playMusic() - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - self.earStatus = "Both in" - return "Both in" - elif (primary_status and not secondary_status) or (not primary_status and secondary_status): - if self.earStatus != "Only one in": - self.stop_thread_event.clear() - s = self.isPlaying() - if s: - self.pauseMusic() - delayed_thread = threading.Timer(0.3, delayed_action, args=[s]) - delayed_thread.start() - self.firstEarOutTime = time.time() - if self.earStatus == "Both out": - os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp-sink") - logging.info("Setting profile to a2dp-sink") - 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 - try: - set_volume(new_volume) - except Exception as e: - logging.error(f"Error setting volume: {e}") - 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: - data = client.recv(1024) - if data: - try: - response = json.loads(data.decode('utf-8')) - if response["type"] == "battery": - with battery_status_lock: - battery_status = response - logging.debug(f"Received battery status: {response}") - self.battery_status_updated.emit() - 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: - logging.error(f"KeyError: {e} in data: {response}") - -def get_battery_status(): - global battery_status - with battery_status_lock: - logging.info(f"Getting battery status: {battery_status}") - left = battery_status["LEFT"] - right = battery_status["RIGHT"] - case = battery_status["CASE"] - left_status = (left['status'] or 'Unknown').title().replace('_', ' ') - right_status = (right['status'] or 'Unknown').title().replace('_', ' ') - case_status = (case['status'] or 'Unknown').title().replace('_', ' ') - return f"Left: {left['level']}% - {left_status} | Right: {right['level']}% - {right_status} | Case: {case['level']}% - {case_status}" - -from aln import enums - -enums=enums.enums - -def set_anc_mode(mode): - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: - client.connect(SOCKET_PATH) - command = enums.NOISE_CANCELLATION_OFF - if mode == "on": - command = enums.NOISE_CANCELLATION_ON - elif mode == "off": - command = enums.NOISE_CANCELLATION_OFF - elif mode == "transparency": - command = enums.NOISE_CANCELLATION_TRANSPARENCY - elif mode == "adaptive": - command = enums.NOISE_CANCELLATION_ADAPTIVE - client.sendall(command) - response = client.recv(1024) - return json.loads(response.decode()) - -def control_anc(action): - response = set_anc_mode(action) - logging.info(f"ANC action: {action}, Response: {response}") - -def signal_handler(sig, frame): - logging.info("Exiting...") - QApplication.quit() - sys.exit(0) - -# Register the signal handler for SIGINT -signal.signal(signal.SIGINT, signal_handler) - -app = QApplication(sys.argv) - -# Create the system tray icon -tray_icon = QSystemTrayIcon(QIcon("icon.png"), app) -tray_icon.setToolTip(get_battery_status()) - -# Create the menu -menu = QMenu() - -# Add ANC control actions -anc_on_action = QAction("ANC On") -anc_on_action.triggered.connect(lambda: control_anc("on")) -menu.addAction(anc_on_action) - -anc_off_action = QAction("ANC Off") -anc_off_action.triggered.connect(lambda: control_anc("off")) -menu.addAction(anc_off_action) - -anc_transparency_action = QAction("Transparency Mode") -anc_transparency_action.triggered.connect(lambda: control_anc("transparency")) -menu.addAction(anc_transparency_action) - -anc_adaptive_action = QAction("Adaptive Mode") -anc_adaptive_action.triggered.connect(lambda: control_anc("adaptive")) -menu.addAction(anc_adaptive_action) - -quit = QAction("Quit") -quit.triggered.connect(app.quit) -menu.addAction(quit) - -# Add the menu to the tray icon -tray_icon.setContextMenu(menu) - -# Show the tray icon -tray_icon.show() - -# Create an instance of BatteryStatusUpdater -battery_status_updater = BatteryStatusUpdater() - -# Connect the signal to the slot -battery_status_updater.battery_status_updated.connect(lambda: tray_icon.setToolTip(get_battery_status())) - -# Start the battery status listener thread -listener_thread = threading.Thread(target=battery_status_updater.listen_to_socket, daemon=True) -listener_thread.start() - -# Run the application -sys.exit(app.exec_()) diff --git a/linux.old/examples/daemon/write-data.py b/linux.old/examples/daemon/write-data.py deleted file mode 100644 index 5490692..0000000 --- a/linux.old/examples/daemon/write-data.py +++ /dev/null @@ -1,81 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import socket -import argparse -from aln import enums -import logging - -class CustomFormatter(logging.Formatter): - # Define color codes for different log levels - COLORS = { - logging.DEBUG: "\033[48;5;240;38;5;15m%s\033[1;0m", # Grey background, white bold text - logging.INFO: "\033[48;5;34;38;5;15m%s\033[1;0m", # Green background, white bold text - logging.WARNING: "\033[1;48;5;214;38;5;0m%s\033[1;0m", # Orange background, black bold text - logging.ERROR: "\033[1;48;5;202;38;5;15m%s\033[1;0m", # Orange-red background, white bold text - logging.CRITICAL: "\033[1;48;5;196;38;5;15m%s\033[1;0m", # Pure red background, white bold text - } - - def format(self, record): - # Apply color to the level name - levelname = self.COLORS.get(record.levelno, "%s") % record.levelname.ljust(8) - record.levelname = levelname - - # Format the message - formatted_message = super().format(record) - - return formatted_message - -# Custom formatter with fixed width for level name -formatter = CustomFormatter('\033[2;37m%(asctime)s\033[1;0m - %(levelname)s - %(message)s') - -logging.basicConfig(level=logging.DEBUG) - -# Set the custom formatter for the root logger -logging.getLogger().handlers[0].setFormatter(formatter) - - -SOCKET_PATH = "/tmp/airpods_daemon.sock" - -def send_command(command): - """Send a command to the daemon via UNIX domain socket.""" - try: - # Create a socket connection to the daemon - logging.info("Connecting to daemon...") - client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - client_socket.connect(SOCKET_PATH) - logging.info("Connected to daemon") - - # Send the command - client_socket.sendall(command) - logging.info(f"Sent command: {command}") - - # Close the connection - client_socket.close() - logging.info("Socket closed") - except Exception as e: - logging.error(f"Error communicating with daemon: {e}") - -def parse_arguments(): - parser = argparse.ArgumentParser(description="Send data.") - parser.add_argument("data", help="Bytes to send, in hex") - return parser.parse_args() - -if __name__ == "__main__": - args = parse_arguments() - command = bytes.fromhex(args.data) - - send_command(command) diff --git a/linux.old/examples/logger-and-anc.py b/linux.old/examples/logger-and-anc.py deleted file mode 100644 index 0273a2c..0000000 --- a/linux.old/examples/logger-and-anc.py +++ /dev/null @@ -1,121 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -from aln import Connection -from aln import enums -import logging -import threading -import time -import sys -import shutil - -AIRPODS_MAC = '28:2D:7F:C2:05:5B' - -class CustomFormatter(logging.Formatter): - def format(self, record): - # Format the log message with spaces around colons without altering the original message - formatted_message = record.getMessage().replace(':', ': ') - record.message = formatted_message - return super().format(record) - -class ConsoleHandler(logging.StreamHandler): - def __init__(self, stream=None): - super().__init__(stream) - self.terminator = '\n' - self.log_lines = [] - - def emit(self, record): - try: - msg = self.format(record) - self.log_lines.append(msg) - self.display_logs() - except Exception: - self.handleError(record) - - def display_logs(self): - sys.stdout.write('\033[H\033[J') # Clear the screen - terminal_height, _ = shutil.get_terminal_size() - log_display_lines = self.log_lines[-(terminal_height - 5):] # Display the last terminal_height - 5 log lines - empty_lines = terminal_height - 5 - len(log_display_lines) - - for _ in range(empty_lines): - sys.stdout.write('\n') # Fill empty space with new lines - - for line in log_display_lines: - sys.stdout.write(line + self.terminator) - - sys.stdout.write('1: ANC Off\n') - sys.stdout.write('2: Transparency\n') - sys.stdout.write('3: Adaptive Transparency\n') - sys.stdout.write('4: ANC On\n') - sys.stdout.write('Select ANC Mode: ') - sys.stdout.flush() - -def input_thread(connection: Connection): - while True: - anc_mode = input() - if anc_mode == '1': - connection.send(enums.NOISE_CANCELLATION_OFF) - logging.info('ANC Off') - elif anc_mode == '2': - connection.send(enums.NOISE_CANCELLATION_TRANSPARENCY) - logging.info('Transparency On') - elif anc_mode == '3': - connection.send(enums.NOISE_CANCELLATION_ADAPTIVE) - logging.info('Adaptive Transparency On') - elif anc_mode == '4': - connection.send(enums.NOISE_CANCELLATION_ON) - logging.info('ANC On') - else: - logging.error('Invalid ANC Mode') - -def main(): - # Set up logging - handler = ConsoleHandler() - - logging.addLevelName(logging.DEBUG, "\033[1;34m%s\033[1;0m" % logging.getLevelName(logging.DEBUG)) - logging.addLevelName(logging.INFO, "\033[1;32m%s\033[1;0m" % logging.getLevelName(logging.INFO)) - logging.addLevelName(logging.WARNING, "\033[1;33m%s\033[1;0m" % logging.getLevelName(logging.WARNING)) - logging.addLevelName(logging.ERROR, "\033[1;31m%s\033[1;0m" % logging.getLevelName(logging.ERROR)) - logging.addLevelName(logging.CRITICAL, "\033[1;41m%s\033[1;0m" % logging.getLevelName(logging.CRITICAL)) - - formatter = CustomFormatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s') - handler.setFormatter(formatter) - logging.basicConfig(level=logging.DEBUG, handlers=[handler]) - - connection = Connection(AIRPODS_MAC) - connection.connect() - logging.info('Sending Handshake') - connection.send(enums.HANDSHAKE) - logging.info('Initializing Notifications') - connection.initialize_notifications() - logging.info('Initialized Notifications') - - # Start the input thread - thread = threading.Thread(target=input_thread, args=(connection,)) - thread.daemon = True - thread.start() - - try: - # Keep the main thread alive to handle logging - while True: - time.sleep(1) - except KeyboardInterrupt: - logging.info('Program interrupted. Exiting...') - connection.disconnect() # Ensure the connection is properly closed - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/linux.old/examples/standalone-old.py b/linux.old/examples/standalone-old.py deleted file mode 100644 index 69e38eb..0000000 --- a/linux.old/examples/standalone-old.py +++ /dev/null @@ -1,339 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import threading -import bluetooth -import subprocess -import time -import threading -import os - -# Bluetooth MAC address of AirPods -AIRPODS_MAC = "28:2D:7F:C2:05:5B" - -class initL2CAP(): - lastEarStatus = "" - earStatus = "" - wasMusicPlayingInBoth = False - wasMusicPlayingInSingle = False - - def pauseMusic(self): - subprocess.call(("playerctl", "pause", "--ignore-player", "OnePlus_7")) - - def playMusic(self): - subprocess.call(("playerctl", "play", "--ignore-player", "OnePlus_7")) - - def getMusicStatus(self): - return subprocess.getoutput("playerctl status --ignore-player OnePlus_7").strip() - - # Change to MAC address of your AirPods - - connected = False - - cmd_off = b"\x04\x00\x04\x00\x09\x00\x0d\x01\x00\x00\x00" - cmd_on = b"\x04\x00\x04\x00\x09\x00\x0d\x02\x00\x00\x00" - cmd_transparency = b"\x04\x00\x04\x00\x09\x00\x0d\x03\x00\x00\x00" - cmd_adaptive = b"\x04\x00\x04\x00\x09\x00\x0d\x04\x00\x00\x00" - cmd_ca_off = b"\x04\x00\x04\x00\x09\x00\x28\x02\x00\x00\x00" - cmd_ca_on = b"\x04\x00\x04\x00\x09\x00\x28\x01\x00\x00\x00" - - def start(self): - cmd_handshake = b"\x00\x00\x04\x00\x01\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00" - # cmd_smth0 = b"\x04\x00\x04\x00\x0f\x00\xff\xff\xfe\xff" - cmd_smth1 = b"\x04\x00\x04\x00\x0f\x00\xff\xff\xff\xff" - - address = "28:2D:7F:C2:05:5B" - - aap_service = "74EC2172-0BAD-4D01-8F77-997B2BE0722A" - aap_port = 0x1001 - - services = bluetooth.find_service(address=address) - - service = [s for s in services if s["service-classes"] == [aap_service]] - - if not service: - print("Device does not have AAP service") - exit() - - self.sock = bluetooth.BluetoothSocket(bluetooth.L2CAP) - sock = self.sock - sock.connect((address, aap_port)) - - print("Connected to AirPods") - self.connected = True - print("Sending handshake...") - print(sock.type) - - sock.send(cmd_handshake) - # sock.send(cmd_smth0) - sock.send(cmd_smth1) - - threading.Thread(target=self.listen).start() - - # battery info: 04 00 04 00 04 00 03 02 01 64 01 01 04 01 64 01 01 08 01 34 02 01 - - def parse_battery_status(self, data): - if len(data) != 22: - return - self.left_bud_level = data[9] - self.left_bud_status = data[10] - - self.right_bud_level = data[14] - self.right_bud_status = data[15] - - self.case_level = data[19] - self.case_status = data[20] - - # Interpret the status - def interpret_status(status): - if status == 1: - return "Charging" - elif status == 2: - return "Not charging" - elif status == 4: - return "Disconnected" - else: - return "Unknown" - - # Print the results - print(f"Left Bud: {self.left_bud_level}% - {interpret_status(self.left_bud_status)}") - print(f"Right Bud: {self.right_bud_level}% - {interpret_status(self.right_bud_status)}") - print(f"Case: {self.case_level}% - {interpret_status(self.case_status)}") - - - def parse_anc_status(self, data): - # 04 00 04 00 09 00 0d 03 00 00 00 - if len(data) != 11 and data.hex().startswith("040004000600"): - return - if data[7] == 1: - return "Off" - elif data[7] == 2: - return "On" - elif data[7] == 3: - return "Transparency" - elif data[7] == 4: - return "Adaptive" - - firstEarOutTime = 0 - stop_thread_event = threading.Event() - - def parse_inear_status(self, data): - if len(data) != 8: - return - - second_status = data[6] - first_status = data[7] - - def delayed_action(self, s): - print(s) - if not self.stop_thread_event.is_set(): - print("Delayed action") - if self.wasMusicPlayingInSingle: - self.playMusic() - self.wasMusicPlayingInBoth = False - elif self.wasMusicPlayingInBoth or s == "Playing": - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - - if first_status and second_status: - if self.earStatus != "Both out": - s = self.getMusicStatus() - self.pauseMusic() - os.system("pacmd set-card-profile bluez_card.28_2D_7F_C2_05_5B off") - if self.earStatus == "Only one in": - if self.firstEarOutTime != 0 and time.time() - self.firstEarOutTime < 0.3: - print("Only one in called with both out") - self.wasMusicPlayingInSingle = True - self.wasMusicPlayingInBoth = True - self.stop_thread_event.set() - else: - if s == "Playing": - self.wasMusicPlayingInSingle = True - else: - self.wasMusicPlayingInSingle = False - # wasMusicPlayingInSingle = True - elif self.earStatus == "Both in": - # should be unreachable - s = self.getMusicStatus() - if s == "Playing": - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - else: - self.wasMusicPlayingInSingle = False - self.earStatus = "Both out" - return "Both out" - elif not first_status and not second_status: - if self.earStatus != "Both in": - if self.earStatus == "Both out": - os.system("pacmd set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp_sink") - elif self.earStatus == "Only one in": - self.stop_thread_event.set() - s = self.getMusicStatus() - if s == "Playing": - self.wasMusicPlayingInBoth = True - if self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth: - self.playMusic() - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - self.earStatus = "Both in" - return "Both in" - elif (first_status and not second_status) or (not first_status and second_status): - if self.earStatus != "Only one in": - self.stop_thread_event.clear() - s = self.getMusicStatus() - self.pauseMusic() - delayed_thread = threading.Timer(0.3, delayed_action, args=[self, s]) - delayed_thread.start() - self.firstEarOutTime = time.time() - if self.earStatus == "Both out": - os.system("pacmd set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp_sink") - self.earStatus = "Only one in" - - return "Only one in" - - def listen(self): - while True: - res = self.sock.recv(1024) - print(f"Response: {res.hex()}") - self.battery_status = self.parse_battery_status(res) - self.inear_status = self.parse_inear_status(res) - # anc_status = parse_anc_status(res) - # if anc_status: - # print("ANC: ", anc_status) - if self.battery_status: - print(self.battery_status) - if self.inear_status: - print(self.inear_status) - - - # while True: - # print("Select command:") - # print("1. Turn off") - # print("2. Turn on") - # print("3. Toggle transparency") - # print("4. Toggle Adaptive") - # print("5. Conversational Awareness On") - # print("6. Conversational Awareness Off") - # print("0. Exit") - - # cmd = input("Enter command: ") - - # if cmd == "0": - # break - # elif cmd == "1": - # self.sock.send(cmd_off) - # elif cmd == "2": - # self.sock.send(cmd_on) - # elif cmd == "3": - # self.sock.send(cmd_transparency) - # elif cmd == "4": - # self.sock.send(cmd_adaptive) - # elif cmd == "5": - # self.sock.send(cmd_ca_on) - # elif cmd == "6": - # self.sock.send(cmd_ca_off) - - def stop(self): - self.connected = False - self.sock.close() - - - -def is_bluetooth_connected(): - try: - result = subprocess.run(["bluetoothctl", "info", AIRPODS_MAC], capture_output=True, text=True) - return "Connected: yes" in result.stdout - except Exception as e: - print(f"Error checking Bluetooth connection status: {e}") - return False - -# Connect to Bluetooth device using bluetoothctl if not already connected -def connect_bluetooth_device(): - if is_bluetooth_connected(): - print("AirPods are already connected.") - return - - print("Checking if AirPods are available...") - result = subprocess.run(["bluetoothctl", "devices"], capture_output=True, text=True) - if AIRPODS_MAC in result.stdout: - print("AirPods are available. Connecting...") - subprocess.run(["bluetoothctl", "connect", AIRPODS_MAC]) - else: - print("AirPods are not available.") - - time.sleep(2) # Wait for the connection to establish - - # Switch audio output to AirPods (PulseAudio) - try: - result = subprocess.run(["pactl", "list", "short", "sinks"], capture_output=True, text=True) - sink_name = next((line.split()[1] for line in result.stdout.splitlines() if "bluez_sink" in line), None) - if sink_name: - subprocess.run(["pactl", "set-default-sink", sink_name]) - print(f"Switched audio to AirPods: {sink_name}") - - else: - print("Failed to switch audio to AirPods.") - except Exception as e: - print(f"Error switching audio: {e}") - -# Disconnect from Bluetooth device if connected -def disconnect_bluetooth_device(): - if not is_bluetooth_connected(): - print("AirPods are already disconnected.") - return - - print("Disconnecting from AirPods...") - subprocess.run(["bluetoothctl", "disconnect", AIRPODS_MAC]) - -l2cap = initL2CAP() - -# Function to listen to `playerctl --follow` and react to status changes -def mediaListener(): - try: - # Run playerctl --follow in a subprocess - process = subprocess.Popen( - ["playerctl", "--follow", "status"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True - ) - - # Continuously read from the subprocess stdout - for line in process.stdout: - if line: # Make sure the line is not empty - line = line.strip() # Remove any extraneous whitespace - print(f"Received event from playerctl: {line}") - - if "Playing" in line: - print("Media started playing") - connect_bluetooth_device() - if not l2cap.connected: - l2cap.start() - elif "Paused" in line or "Stopped" in line: - print("Media paused or stopped") - # disconnect_bluetooth_device() - - # Check for any errors in stderr - stderr = process.stderr.read() - if stderr: - print(f"Error: {stderr}") - - except Exception as e: - print(f"An error occurred in mediaListener: {e}") - -mediaListener() - -# thread = threading.Thread(target=mediaListener) -# thread.start() - -# thread.stop() diff --git a/linux.old/examples/standalone.py b/linux.old/examples/standalone.py deleted file mode 100644 index 8a75da1..0000000 --- a/linux.old/examples/standalone.py +++ /dev/null @@ -1,474 +0,0 @@ -import sys -import threading -from PyQt5.QtWidgets import QApplication, QSystemTrayIcon, QMenu, QAction, QMessageBox -from PyQt5.QtGui import QIcon, QPixmap, QPainter -from PyQt5.QtCore import QObject, pyqtSignal, Qt -from PyQt5.QtGui import QFont, QPalette -import logging -import signal -import subprocess -import time -import os -from aln import Connection, enums -from aln.Notifications import Notifications -import argparse -import dbus -import dbus.mainloop.glib - -enums = enums.enums -logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') -logging.getLogger().addHandler(logging.StreamHandler()) - -tray_icon = None -anc_actions = None - -battery_status = { - "LEFT": {"status": "Unknown", "level": 0}, - "RIGHT": {"status": "Unknown", "level": 0}, - "CASE": {"status": "Unknown", "level": 0} -} -anc_mode = 0 -battery_status_lock = threading.Lock() - -CONVERSATIONAL_AWARENESS_FILE = os.path.expanduser("~/.airpods_conversational_awareness") -CONFIG_FILE = os.path.expanduser("~/.config/aln") - -def load_conversational_awareness_state(): - if os.path.exists(CONVERSATIONAL_AWARENESS_FILE): - with open(CONVERSATIONAL_AWARENESS_FILE, "r") as file: - return file.read().strip() == "enabled" - return False - -def save_conversational_awareness_state(enabled): - with open(CONVERSATIONAL_AWARENESS_FILE, "w") as file: - file.write("enabled" if enabled else "disabled") - -def toggle_conversational_awareness(): - current_state = load_conversational_awareness_state() - new_state = not current_state - save_conversational_awareness_state(new_state) - connection.send(enums.SET_CONVERSATION_AWARENESS_ON if new_state else enums.SET_CONVERSATION_AWARENESS_OFF) - logging.info(f"Conversational Awareness {'enabled' if new_state else 'disabled'}") - -def load_mac_address(): - if os.path.exists(CONFIG_FILE): - with open(CONFIG_FILE, "r") as file: - return file.read().strip() - return None - -def save_mac_address(mac_address): - with open(CONFIG_FILE, "w") as file: - file.write(mac_address) - -def parse_arguments(): - parser = argparse.ArgumentParser(description="Standalone tray application for AirPods.") - parser.add_argument("--mac", help="MAC address of the AirPods") - return parser.parse_args() - -def get_connected_airpods(): - logging.info("Checking for connected AirPods...") - result = subprocess.run("bluetoothctl devices | cut -f2 -d' ' | while read uuid; do bluetoothctl info $uuid; done | grep -e 'Device\\|Connected\\|Name'", shell=True, capture_output=True, text=True) - lines = result.stdout.splitlines() - for i in range(0, len(lines), 3): - if "Connected: yes" in lines[i + 2]: - addr = lines[i].split()[1] - name = lines[i + 1].split(": ")[1] - logging.debug(f"Checking services for connected device: {name} ({addr})") - services_result = run_sdptool_browse(addr) - if services_result and "UUID 128: 74ec2172-0bad-4d01-8f77-997b2be0722a" in services_result: - logging.info(f"Found connected AirPods: {name} ({addr})") - return addr - logging.error("No connected AirPods found.") - return None - -def run_sdptool_browse(addr, retries=5): - for attempt in range(retries): - services_result = subprocess.run(f"sdptool browse {addr}", shell=True, capture_output=True, text=True) - if "Failed to connect to SDP server" not in services_result.stderr: - return services_result.stdout - logging.warning(f"Failed to connect to SDP server on {addr}, attempt {attempt + 1} of {retries}") - time.sleep(1) - logging.error(f"Failed to connect to SDP server on {addr} after {retries} attempts") - return None - -def set_card_profile(mac_address, profile): - os.system(f"pactl set-card-profile bluez_card.{mac_address.replace(':', '_')} {profile}") - -class MediaController: - def __init__(self, mac_address): - self.mac_address = mac_address - self.earStatus = "Both out" - self.wasMusicPlayingInSingle = False - self.wasMusicPlayingInBoth = False - self.firstEarOutTime = 0 - self.stop_thread_event = threading.Event() - - def playMusic(self): - logging.info("Playing music") - subprocess.call(("playerctl", "play")) - - def pauseMusic(self): - logging.info("Pausing music") - subprocess.call(("playerctl", "--all-players", "pause")) - - 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] - - logging.debug(f"Handling play/pause with data: {data}, previousStatus: {self.earStatus}, wasMusicPlaying: {self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth}") - - def delayed_action(s): - if not self.stop_thread_event.is_set(): - if self.wasMusicPlayingInSingle: - self.playMusic() - self.wasMusicPlayingInBoth = False - elif self.wasMusicPlayingInBoth or s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - - if primary_status and secondary_status: - if self.earStatus != "Both out": - s = self.isPlaying() - if s: - self.pauseMusic() - set_card_profile(self.mac_address, "off") - logging.info("Setting profile to off") - if self.earStatus == "Only one in": - if self.firstEarOutTime != 0 and time.time() - self.firstEarOutTime < 0.3: - self.wasMusicPlayingInSingle = True - self.wasMusicPlayingInBoth = True - self.stop_thread_event.set() - else: - if s: - self.wasMusicPlayingInSingle = True - else: - self.wasMusicPlayingInSingle = False - elif self.earStatus == "Both in": - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - else: - self.wasMusicPlayingInSingle = False - self.earStatus = "Both out" - return "Both out" - elif not primary_status and not secondary_status: - if self.earStatus != "Both in": - if self.earStatus == "Both out": - set_card_profile(self.mac_address, "a2dp-sink") - logging.info("Setting profile to a2dp-sink") - elif self.earStatus == "Only one in": - self.stop_thread_event.set() - s = self.isPlaying() - if s: - self.wasMusicPlayingInBoth = True - if self.wasMusicPlayingInSingle or self.wasMusicPlayingInBoth: - self.playMusic() - self.wasMusicPlayingInBoth = True - self.wasMusicPlayingInSingle = False - self.earStatus = "Both in" - return "Both in" - elif (primary_status and not secondary_status) or (not primary_status and secondary_status): - if self.earStatus != "Only one in": - self.stop_thread_event.clear() - s = self.isPlaying() - if s: - self.pauseMusic() - delayed_thread = threading.Timer(0.3, delayed_action, args=[s]) - delayed_thread.start() - self.firstEarOutTime = time.time() - if self.earStatus == "Both out": - set_card_profile(self.mac_address, "a2dp-sink") - logging.info("Setting profile to a2dp-sink") - self.earStatus = "Only one in" - return "Only one in" - -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 - -def set_volume(percent): - subprocess.run(["pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{percent}%"]) - -initial_volume = get_current_volume() - -def handle_conversational_awareness(status): - if status < 1 or status > 9: - logging.error(f"Invalid status: {status}") - pass - - global initial_volume - - if status == 1 or status == 2: - globals()["initial_volume"] = get_current_volume() - new_volume = max(0, min(int(initial_volume * 0.1), 100)) - elif status == 3: - new_volume = max(0, min(int(initial_volume * 0.4), 100)) - elif status == 6: - new_volume = max(0, min(int(initial_volume * 0.5), 100)) - elif status >= 8: - new_volume = initial_volume - try: - set_volume(new_volume) - except Exception as e: - logging.error(f"Error setting volume: {e}") - logging.getLogger("Conversational Awareness").info(f"Volume set to {new_volume}% based on conversational awareness status: {status}") - - if status == 9: - logging.getLogger("Conversational Awareness").info("Conversation ended. Restored volume to original level.") - -class BatteryStatusUpdater(QObject): - battery_status_updated = pyqtSignal(str) - anc_mode_updated = pyqtSignal(int) - - def __init__(self, connection, mac_address): - super().__init__() - self.connection = connection - self.media_controller = MediaController(mac_address) - - def notification_handler(self, notification_type: int, data: bytes): - global battery_status, anc_mode - logging.debug(f"Received data: {' '.join(f'{byte:02X}' for byte in data)}") - if notification_type == Notifications.BATTERY_UPDATED: - battery = self.connection.notificationListener.BatteryNotification.getBattery() - with battery_status_lock: - battery_status = { - "LEFT": {"status": battery[0].get_status(), "level": battery[0].get_level()}, - "RIGHT": {"status": battery[1].get_status(), "level": battery[1].get_level()}, - "CASE": {"status": battery[2].get_status(), "level": battery[2].get_level()} - } - status_str = get_battery_status() - self.battery_status_updated.emit(status_str) - logging.debug(f"Updated battery status: {battery_status}") - elif notification_type == Notifications.EAR_DETECTION_UPDATED: - earDetection = self.connection.notificationListener.EarDetectionNotification.getEarDetection() - self.media_controller.handlePlayPause(earDetection) - logging.debug(f"Received ear detection status: {earDetection}") - elif notification_type == Notifications.ANC_UPDATED: - anc_mode = self.connection.notificationListener.ANCNotification.status - self.anc_mode_updated.emit(anc_mode) - logging.debug(f"Received ANC status: {anc_mode}") - elif notification_type == Notifications.CA_UPDATED: - ca_status = self.connection.notificationListener.ConversationalAwarenessNotification.status - handle_conversational_awareness(ca_status) - logging.debug(f"Received CA status: {ca_status}") - -def get_battery_status(): - global battery_status - with battery_status_lock: - left = battery_status["LEFT"] - right = battery_status["RIGHT"] - case = battery_status["CASE"] - left_status = (left['status'] or 'Unknown').title().replace('_', ' ') - right_status = (right['status'] or 'Unknown').title().replace('_', ' ') - case_status = (case['status'] or 'Unknown').title().replace('_', ' ') - - status_emoji = { - "Not Charging": "", - "Charging": "⚡", - } - - left_status_emoji = status_emoji.get(left_status, "") - right_status_emoji = status_emoji.get(right_status, "") - case_status_emoji = status_emoji.get(case_status, "") - - return f"Left: {left['level']}% {left_status_emoji} | Right: {right['level']}% {right_status_emoji} | Case: {case['level']}% {case_status_emoji}" - -def create_battery_icon(): - global battery_status - with battery_status_lock: - left_level = battery_status["LEFT"]["level"] - right_level = battery_status["RIGHT"]["level"] - lowest_level = min(left_level, right_level) - icon_size = 64 - - pixmap = QPixmap(icon_size, icon_size) - pixmap.fill(Qt.transparent) - - painter = QPainter(pixmap) - - is_dark_mode = QApplication.palette().color(QPalette.Window).value() < 128 - text_color = Qt.white if is_dark_mode else Qt.black - - painter.setPen(text_color) - painter.setFont(QFont('Arial', 20, QFont.Bold)) - painter.drawText(pixmap.rect(), Qt.AlignCenter, f"{lowest_level}%") - painter.end() - - return QIcon(pixmap) - -def signal_handler(sig, frame): - logging.info("Exiting...") - QApplication.quit() - sys.exit(0) - -connection=None -battery_status_updater = None - -def update_anc_menu(anc_mode, actions): - for action in actions: - action.setChecked(False) - if anc_mode == 1: - actions[0].setChecked(True) - elif anc_mode == 2: - actions[1].setChecked(True) - elif anc_mode == 3: - actions[2].setChecked(True) - elif anc_mode == 4: - actions[3].setChecked(True) - -def toggle_conversational_awareness_action(action): - toggle_conversational_awareness() - action.setChecked(load_conversational_awareness_state()) - -def listen_for_device_connections(): - dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) - bus = dbus.SystemBus() - logging.info("Listening for device connections...") - - def device_connected(interface, changed, invalidated, path): - # /org/bluez/hci0/dev_mac_address/* - if 'Connected' in changed and changed['Connected']: - if path.split("/")[-1] == "": - return - addr = path.split("/")[-1].replace("_", ":").replace("dev:", "") - name = changed.get('Name', 'Unknown') - logging.info(f"Device connected: {name} ({addr})") - logging.debug(f"Running command: sdptool browse {addr}") - services_result = run_sdptool_browse(addr) - logging.debug(f"Services result: {services_result}") - if services_result and "UUID 128: 74ec2172-0bad-4d01-8f77-997b2be0722a" in services_result: - logging.info(f"Found connected AirPods: {name} ({addr})") - connect_to_airpods(addr) - - bus.add_signal_receiver(device_connected, dbus_interface="org.freedesktop.DBus.Properties", signal_name="PropertiesChanged", path_keyword="path") - - def interfaces_added(path, interfaces): - logging.debug(f"Interfaces added: {path}") - if 'org.bluez.Device1' in interfaces and interfaces['org.bluez.Device1'].get('Connected', False): - addr = interfaces['org.bluez.Device1']['Address'] - name = interfaces['org.bluez.Device1']['Name'] - logging.info(f"Device connected: {name} ({addr})") - if path.endswith("/sep1"): - services_result = run_sdptool_browse(addr) - if services_result and "UUID 128: 74ec2172-0bad-4d01-8f77-997b2be0722a" in services_result: - logging.info(f"Found connected AirPods: {name} ({addr})") - connect_to_airpods(addr) - - bus.add_signal_receiver(interfaces_added, dbus_interface="org.freedesktop.DBus.ObjectManager", signal_name="InterfacesAdded") - - bus.add_signal_receiver(audio_device_changed, dbus_interface="org.PulseAudio.Core1.Device", signal_name="NewPlaybackStream") - - from gi.repository import GLib - loop = GLib.MainLoop() - loop.run() - -def audio_device_changed(*args, **kwargs): - logging.info("Audio output device changed, checking for connected AirPods...") - mac_address = get_connected_airpods() - if mac_address: - connect_to_airpods(mac_address) - -def connect_to_airpods(mac_address): - logging.info(f"Attempting to connect to AirPods at {mac_address}...") - globals()["connection"] = Connection(mac_address) - try: - connection.connect() - connection.send(enums.HANDSHAKE) - globals()["battery_status_updater"] = BatteryStatusUpdater(connection, mac_address) - connection.initialize_notifications(battery_status_updater.notification_handler) - - battery_status_updater.battery_status_updated.connect(lambda status: tray_icon.setToolTip(status)) - battery_status_updater.battery_status_updated.connect(lambda: tray_icon.setIcon(create_battery_icon())) - battery_status_updater.anc_mode_updated.connect(lambda mode: update_anc_menu(mode, anc_actions)) - - save_mac_address(mac_address) - logging.info("Connected to AirPods successfully.") - except Exception as e: - logging.error(f"Failed to connect to AirPods: {e}") - -def main(): - args = parse_arguments() - mac_address = args.mac or load_mac_address() - - logging.debug("Starting standalone tray application...") - app = QApplication(sys.argv) - - globals()["tray_icon"] = QSystemTrayIcon(create_battery_icon(), app) - tray_icon.setToolTip(get_battery_status()) - - menu = QMenu() - - ca_toggle_action = QAction("Toggle Conversational Awareness") - ca_toggle_action.setCheckable(True) - ca_toggle_action.setChecked(load_conversational_awareness_state()) - ca_toggle_action.triggered.connect(lambda: toggle_conversational_awareness_action(ca_toggle_action)) - menu.addAction(ca_toggle_action) - - anc_on_action = QAction("Noise Cancellation") - anc_on_action.setCheckable(True) - anc_on_action.triggered.connect(lambda: control_anc("on")) - menu.addAction(anc_on_action) - - anc_off_action = QAction("Off") - anc_off_action.setCheckable(True) - anc_off_action.triggered.connect(lambda: control_anc("off")) - menu.addAction(anc_off_action) - - anc_transparency_action = QAction("Transparency") - anc_transparency_action.setCheckable(True) - anc_transparency_action.triggered.connect(lambda: control_anc("transparency")) - menu.addAction(anc_transparency_action) - - anc_adaptive_action = QAction("Adaptive") - anc_adaptive_action.setCheckable(True) - anc_adaptive_action.triggered.connect(lambda: control_anc("adaptive")) - menu.addAction(anc_adaptive_action) - - globals()["anc_actions"] = [anc_off_action, anc_on_action, anc_transparency_action, anc_adaptive_action] - - quit_action = QAction("Quit") - quit_action.triggered.connect(lambda: signal_handler(signal.SIGINT, None)) - menu.addAction(quit_action) - - tray_icon.setContextMenu(menu) - tray_icon.show() - - logging.info("Standalone tray application started.") - - if mac_address: - connect_to_airpods(mac_address) - else: - mac_address = get_connected_airpods() - if mac_address: - connect_to_airpods(mac_address) - else: - listen_for_device_connections() - - signal.signal(signal.SIGINT, signal_handler) - - try: - sys.exit(app.exec_()) - except Exception as e: - logging.error(f"An error occurred: {e}") - sys.exit(1) - -def control_anc(action): - command = enums.NOISE_CANCELLATION_OFF - if action == "on": - command = enums.NOISE_CANCELLATION_ON - elif action == "off": - command = enums.NOISE_CANCELLATION_OFF - elif action == "transparency": - command = enums.NOISE_CANCELLATION_TRANSPARENCY - elif action == "adaptive": - command = enums.NOISE_CANCELLATION_ADAPTIVE - connection.send(command) - logging.info(f"ANC action: {action}") - -main() \ No newline at end of file diff --git a/linux.old/icon.png b/linux.old/icon.png deleted file mode 100644 index 861ddf7..0000000 Binary files a/linux.old/icon.png and /dev/null differ diff --git a/linux.old/imgs/daemon-log.png b/linux.old/imgs/daemon-log.png deleted file mode 100644 index 02eea7f..0000000 Binary files a/linux.old/imgs/daemon-log.png and /dev/null differ diff --git a/linux.old/imgs/ear-detection.png b/linux.old/imgs/ear-detection.png deleted file mode 100644 index 47630b2..0000000 Binary files a/linux.old/imgs/ear-detection.png and /dev/null differ diff --git a/linux.old/imgs/read-data.png b/linux.old/imgs/read-data.png deleted file mode 100644 index c1a9385..0000000 Binary files a/linux.old/imgs/read-data.png and /dev/null differ diff --git a/linux.old/imgs/set-anc.png b/linux.old/imgs/set-anc.png deleted file mode 100644 index e7c80e1..0000000 Binary files a/linux.old/imgs/set-anc.png and /dev/null differ diff --git a/linux.old/imgs/tray-icon-hover.png b/linux.old/imgs/tray-icon-hover.png deleted file mode 100644 index ca51be7..0000000 Binary files a/linux.old/imgs/tray-icon-hover.png and /dev/null differ diff --git a/linux.old/imgs/tray-icon-menu.png b/linux.old/imgs/tray-icon-menu.png deleted file mode 100644 index 8f9e827..0000000 Binary files a/linux.old/imgs/tray-icon-menu.png and /dev/null differ diff --git a/linux.old/pyproject.toml b/linux.old/pyproject.toml deleted file mode 100644 index b1bb6ef..0000000 --- a/linux.old/pyproject.toml +++ /dev/null @@ -1,25 +0,0 @@ -[build-system] -requires = ["hatchling"] -build-backend = "hatchling.build" - -[tool.hatch.build.targets.wheel] -packages = ["aln"] - -[project] -name = "aln" -version = "0.0.1" -authors = [ - { name="Kavish Devar", email="mail@kavishdevar.me" }, -] -description = "Use your AirPods the way they were intended." -readme = "README.md" -requires-python = ">=3.8" -classifiers = [ - "Programming Language :: Python :: 3", - "License :: OSI Approved :: MIT License", - "Operating System :: OS Independent", -] - -[project.urls] -Homepage = "https://github.com/kavishdevar/aln" -Issues = "https://github.com/kavishdevar/aln/issues" \ No newline at end of file diff --git a/linux.old/start-daemon.py b/linux.old/start-daemon.py deleted file mode 100644 index b91d9f9..0000000 --- a/linux.old/start-daemon.py +++ /dev/null @@ -1,263 +0,0 @@ -# AirPods like Normal (ALN) - Bringing Apple-only features to Linux and Android for seamless AirPods functionality! -# -# Copyright (C) 2024 Kavish Devar -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import logging.handlers -import socket -import threading -import signal -import sys -import logging -from aln import Connection, enums -from aln.Notifications import Notifications -from aln.Notifications.Battery import Battery -import os -import bluetooth -from aln.enums import enums -connection = None - -AIRPODS_MAC = '28:2D:7F:C2:05:5B' -SOCKET_PATH = '/tmp/airpods_daemon.sock' -LOG_FOLDER = '.' -LOG_FILE = os.path.join(LOG_FOLDER, 'airpods_daemon.log') - -# Global flag to control the server loop -running = True - -# Configure logging to write to a file -# logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG, format='%(asctime)s %(levelname)s : %(message)s') - -# RotatingFileHandler - -formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') -handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes=2**20) -handler.setFormatter(formatter) -handler.setLevel(logging.DEBUG) - -l = logging.getLogger() -l.setLevel(logging.DEBUG) -l.addHandler(handler) - -from json import JSONEncoder - -def handle_client(connection, client_socket): - """Handle client requests by forwarding all received data to aln.Connection, send data back to the client.""" - - def send_status(): - while running: - try: - for notif_key in list(globals().keys()): - if notif_key.startswith("notif_"): - data = globals().get(notif_key) - if data: - if notif_key == "notif_battery": - data: list[Battery] = data - batteryJSON = {"type": "battery"} - for i in data: - batteryJSON[i.get_component()] = { - "status": i.get_status(), - "level": i.get_level() - } - data: str = JSONEncoder().encode(batteryJSON) - - elif notif_key == "notif_ear_detection": - # noinspection PyTypeChecker - data: list[int] = data - earDetectionJSON = { - "type": "ear_detection", - "primary": data[0], - "secondary": data[1] - } - data: str = JSONEncoder().encode(earDetectionJSON) - elif notif_key == "notif_anc": - data: int = data - ancJSON = { - "type": "anc", - "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 - logging.info(f'Sending {notif_key} status: {data}') - client_socket.sendall(data.encode('utf-8')) - logging.info(f'Sent {notif_key} status: {data}') - globals()[notif_key] = None - except socket.error as e: - logging.error(f"Socket error sending status: {e}") - break - except Exception as e: - logging.error(f"Error sending status: {e}") - break - - def receive_commands(): - while running: - try: - data = client_socket.recv(1024) - if not data: - break - logging.info(f'Received command: {data}') - connection.send(data) - except Exception as e: - logging.error(f"Error receiving command: {e}") - break - - # Start two threads to handle sending and receiving data - send_thread = threading.Thread(target=send_status) - send_thread.start() - receive_thread = threading.Thread(target=receive_commands) - receive_thread.start() - - send_thread.join() - receive_thread.join() - - client_socket.close() - logging.info("Client socket closed") - -def start_socket_server(connection): - """Start a UNIX domain socket server.""" - global running - - # Set up the socket - server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - try: - server_socket.bind(SOCKET_PATH) - except OSError: - logging.error(f"Socket already in use or unavailable: {SOCKET_PATH}") - sys.exit(1) - - server_socket.listen(1) - logging.info(f"Socket server listening on {SOCKET_PATH}") - - while running: - try: - client_socket, _ = server_socket.accept() - logging.info("Client connected") - - # Handle the client connection in a separate thread - client_thread = threading.Thread(target=handle_client, args=(connection, client_socket)) - client_thread.start() - except Exception as e: - logging.error(f"Error accepting connection: {e}") - - # Close the server socket when stopped - server_socket.close() - logging.info("Socket server stopped") - -def stop_daemon(_, __): - """Signal handler to stop the daemon.""" - global running - logging.info("Received termination signal. Stopping daemon...") - running = False # Set running flag to False to stop the loop - - # Close the socket gracefully by removing the file path - try: - socket.socket(socket.AF_UNIX, socket.SOCK_STREAM).connect(SOCKET_PATH) - except socket.error: - pass - finally: - # Remove the socket file - if os.path.exists(SOCKET_PATH): - os.remove(SOCKET_PATH) - - sys.exit(0) - -def notification_handler(notification_type: int, data: bytes): - global connection - - logging.debug(f"Received notification: {notification_type}") - if notification_type == Notifications.BATTERY_UPDATED: - logger = logging.getLogger("Battery Status") - battery = connection.notificationListener.BatteryNotification.getBattery() - globals()["notif_battery"] = battery - for i in battery: - logger.debug(f'{i.get_component()} - {i.get_status()}: {i.get_level()}') - elif notification_type == Notifications.EAR_DETECTION_UPDATED: - logger = logging.getLogger("In-Ear Status") - earDetection = connection.notificationListener.EarDetectionNotification.getEarDetection() - globals()["notif_ear_detection"] = earDetection - logger.debug(earDetection) - elif notification_type == Notifications.ANC_UPDATED: - logger = logging.getLogger("ANC Status") - 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) - globals()["notif_unknown"] = hex_data - logger.debug(hex_data) -def main(): - global running - logging.info("Starting AirPods daemon") - connection = Connection(AIRPODS_MAC) - globals()['connection'] = connection - - # Connect to the AirPods and send the handshake - try: - connection.connect() - except bluetooth.btcommon.BluetoothError as e: - logging.error(f"Failed to connect to {AIRPODS_MAC}: {e}") - sys.exit(1) - - connection.send(enums.HANDSHAKE) - logging.info("Handshake sent") - - connection.initialize_notifications(notification_handler) - - # Start the socket server to listen for client connections - start_socket_server(connection) - - # Set up signal handlers to handle termination signals - signal.signal(signal.SIGINT, stop_daemon) # Handle Ctrl+C - signal.signal(signal.SIGTERM, stop_daemon) # Handle kill signal - -if __name__ == "__main__": - # Daemonize the process - if os.fork(): - sys.exit() - - os.setsid() - - if os.fork(): - sys.exit() - - sys.stdout.flush() - sys.stderr.flush() - - with open('/dev/null', 'r') as devnull: - os.dup2(devnull.fileno(), sys.stdin.fileno()) - - with open(LOG_FILE, 'a+') as logfile: - os.dup2(logfile.fileno(), sys.stdout.fileno()) - os.dup2(logfile.fileno(), sys.stderr.fileno()) - - main() \ No newline at end of file