calvin.erfmann/librepods
1
0
Fork 0
mirror of https://github.com/kavishdevar/librepods.git synced 2026-02-10 19:52:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Delete linux.old directory

Browse Source
This commit is contained in:
Tim Gromeyer
2025-04-24 08:56:07 +02:00
parent 893bcc97bc
commit 924efc4faa
31 changed files with 0 additions and 2958 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
linux.old/README.md
View File

@@ -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).

30
linux.old/aln/AirPods/Pro2.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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
}

20
linux.old/aln/AirPods/__init__.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
from . import Pro2
Pro2 = Pro2.Pro2

34
linux.old/aln/Capabilites/__init__.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

41
linux.old/aln/Notifications/ANC.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

80
linux.old/aln/Notifications/Battery.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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]

33
linux.old/aln/Notifications/ConversationalAwareness.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

42
linux.old/aln/Notifications/EarDetection.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

73
linux.old/aln/Notifications/Listener.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

56
linux.old/aln/Notifications/__init__.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

87
linux.old/aln/__init__.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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

265
linux.old/aln/__main__.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

43
linux.old/aln/enums.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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'

47
linux.old/aln/listener.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

180
linux.old/examples/daemon/ear-detection.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

103
linux.old/examples/daemon/read-data.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

89
linux.old/examples/daemon/set-anc.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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)

333
linux.old/examples/daemon/tray.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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_())

81
linux.old/examples/daemon/write-data.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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)

121
linux.old/examples/logger-and-anc.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

339
linux.old/examples/standalone-old.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()

474
linux.old/examples/standalone.py
View File

@@ -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()

BIN
linux.old/icon.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 605 KiB

BIN
linux.old/imgs/daemon-log.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 402 KiB

BIN
linux.old/imgs/ear-detection.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 190 KiB

BIN
linux.old/imgs/read-data.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 284 KiB

BIN
linux.old/imgs/set-anc.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 65 KiB

BIN
linux.old/imgs/tray-icon-hover.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 11 KiB

BIN
linux.old/imgs/tray-icon-menu.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 19 KiB

25
linux.old/pyproject.toml
View File

@@ -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"

263
linux.old/start-daemon.py
View File

@@ -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 <https://www.gnu.org/licenses/>.
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()