diff --git a/aln/Notifications/EarDetection.py b/aln/Notifications/EarDetection.py index 028bf46..e246d4d 100644 --- a/aln/Notifications/EarDetection.py +++ b/aln/Notifications/EarDetection.py @@ -1,6 +1,5 @@ from ..Capabilites import Capabilites from ..enums import enums -import logging from typing import Literal class EarDetectionNotification: diff --git a/examples/daemon/ear-detection.py b/examples/daemon/ear-detection.py index 5d0f953..1942433 100644 --- a/examples/daemon/ear-detection.py +++ b/examples/daemon/ear-detection.py @@ -56,7 +56,7 @@ class MediaController: if self.earStatus != "Both out": s = self.isPlaying() self.pauseMusic() - os.system("pacmd set-card-profile bluez_card.28_2D_7F_C2_05_5B off") + os.system("pactl 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: self.wasMusicPlayingInSingle = True @@ -79,7 +79,7 @@ class MediaController: elif not primary_status and not secondary_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") + os.system("pactl 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.isPlaying() @@ -100,7 +100,7 @@ class MediaController: 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") + os.system("pactl set-card-profile bluez_card.28_2D_7F_C2_05_5B a2dp-sink") self.earStatus = "Only one in" return "Only one in" @@ -122,7 +122,6 @@ def read(): try: data: dict = json.loads(d.decode('utf-8')) if data["type"] == "ear_detection": - logging.debug(f"Ear detection: {data['primary']} - {data['secondary']}") media_controller.handlePlayPause([data['primary'], data['secondary']]) except json.JSONDecodeError as e: logging.error(f"Error deserializing data: {e}") diff --git a/examples/daemon/read-data.py b/examples/daemon/read-data.py index 63519c2..ccbf51c 100644 --- a/examples/daemon/read-data.py +++ b/examples/daemon/read-data.py @@ -1,16 +1,49 @@ import socket import json -from aln.Notifications import Battery +import logging SOCKET_PATH = "/tmp/airpods_daemon.sock" +import logging +import textwrap +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;90m%(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) + +# Example usage +logging.info("This is an info message. This is a continuation of the info message. This is a continuation of the info message. This is a continuation of the info message.") + 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) - print("Connecting to daemon...") + logging.info("Connecting to daemon...") client_socket.connect(SOCKET_PATH) # Receive data @@ -18,25 +51,36 @@ def read(): d = client_socket.recv(1024) if d: try: - data: dict = json.loads(d.decode('utf-8')) - if data["type"] == "battery": - for b in data.keys(): - print(f"Received battery status: {b} - {data[b]}") - elif data["type"] == "ear_detection": - print(f"Ear detection: {data['primary']} - {data['secondary']}") + 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"] == "unknown": + logging.info(f"Received data: {data['data']}") + else: + logging.info(f"Received data: {data}") else: - print(f"Received data: {data}") + logging.error("Received data is not a dictionary") except json.JSONDecodeError as e: - print(f"Error deserializing data: {e}") + logging.warning(f"Error deserializing data: {e}") + logging.warning(f"raw data: {d}") + 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: - print(f"Error communicating with daemon: {e}") + logging.critical(f"Error communicating with daemon: {e}") finally: if client_socket: client_socket.close() - print("Socket closed") + logging.warning("Socket closed") if __name__ == "__main__": read() \ No newline at end of file diff --git a/examples/daemon/set-anc.py b/examples/daemon/set-anc.py index 00dc6a7..8b27948 100644 --- a/examples/daemon/set-anc.py +++ b/examples/daemon/set-anc.py @@ -3,15 +3,33 @@ import argparse from aln import enums import logging -# Configure logging -logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') +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 + } -# Colorful logging -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)) + 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;90m%(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" diff --git a/start-daemon.py b/start-daemon.py index efb751d..916c881 100644 --- a/start-daemon.py +++ b/start-daemon.py @@ -20,8 +20,12 @@ LOG_FILE = os.path.join(LOG_FOLDER, 'airpods_daemon.log') running = True # Configure logging to write to a file -logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG, format='%(asctime)s %(levelname)s : %(message)s') -# logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)s : %(message)s') +# 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.FileHandler(LOG_FILE, maxBytes=2**20, backupCount=5, delay=0) from json import JSONEncoder @@ -52,11 +56,10 @@ def handle_client(connection, client_socket): "secondary": data[1] } data: str = JSONEncoder().encode(earDetectionJSON) - else: + 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}) - continue if not client_socket or not isinstance(client_socket, socket.socket): logging.error("Invalid client socket")