columba/python/ble_modules/android_ble_interface.py
torlando-tech 88f5966fbe fix: address PR #609 code review — volatile fields, dead code, UI dedup, overlap warning
- Add @Volatile to cross-thread power settings fields (KotlinBLEBridge,
  BleScanner, BleAdvertiser) for JMM visibility between Python/Kotlin threads
- Remove dead **power_kwargs from AndroidBLEDriver.start()
- Add ordering-safety comment in AndroidBLEInterface.__init__()
- Upgrade closeImmediate() off-thread log from Log.w to Log.e
- Deduplicate preset values in InterfaceConfigDialog using BlePowerPreset.getSettings()
- Add scan overlap warning when scan duration >= active scan interval

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-06 20:10:26 -05:00

113 lines
4.6 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AndroidBLEInterface - Reticulum Interface for Android BLE
=========================================================
This module provides a Reticulum interface for Bluetooth Low Energy (BLE) on
Android devices. It leverages the `ble-reticulum` driver-based architecture
to reuse the core `BLEInterface` logic while plugging in an Android-specific
driver (`AndroidBLEDriver`).
The `AndroidBLEDriver` acts as a bridge to the `KotlinBLEBridge` via Chaquopy,
which in turn manages all native Android BLE operations.
This interface is automatically discovered and loaded by Reticulum if placed
in the `~/.reticulum/interfaces/` directory and configured in `config`.
This file is a thin wrapper that configures and initializes the generic
`BLEInterface` with the `AndroidBLEDriver`.
Author: Columba Project
License: MIT
"""
import RNS
import sys
import os
# When Reticulum loads this interface with exec(), we need to ensure the interfaces
# directory is in sys.path so imports work. The interfaces are in the app storage.
_storage_base = os.environ.get("HOME", "/data/user/0/com.lxmf.messenger/files")
_interfaces_dir = os.path.join(_storage_base, "reticulum", "interfaces")
if os.path.exists(_interfaces_dir) and _interfaces_dir not in sys.path:
sys.path.insert(0, _interfaces_dir)
# Import the generic BLEInterface and the Android-specific driver
# Note: BLEInterface is deployed to the same interfaces directory as this file
from BLEInterface import BLEInterface
from drivers.android_ble_driver import AndroidBLEDriver
class AndroidBLEInterface(BLEInterface):
"""
Reticulum interface for Android BLE.
This class inherits from the generic `BLEInterface` and uses the
`AndroidBLEDriver` to provide Android-specific BLE functionality.
All the complex logic for peer management, connection handling, and
fragmentation is handled by the parent `BLEInterface`.
"""
# Override driver class to use Android implementation
driver_class = AndroidBLEDriver
def __init__(self, owner, config=None):
"""
Initialize the Android BLE interface.
Args:
owner: The Reticulum Transport instance that owns this interface.
config: A dictionary containing configuration options.
"""
# Call parent constructor - it will use our driver_class
super().__init__(owner, config)
# Configure BLE power settings from config.
# Safe to call after super().__init__(): the bridge (and its scanner/advertiser)
# is created in KotlinBLEBridge's constructor, so the objects already exist
# even though start() hasn't been called yet.
if config and hasattr(self, 'driver') and self.driver is not None:
power_preset = config.get("ble_power_preset", "balanced")
self.driver.configure_power(
preset=power_preset,
discovery_interval_ms=int(config.get("ble_discovery_interval_ms", 5000)),
discovery_interval_idle_ms=int(config.get("ble_discovery_interval_idle_ms", 30000)),
scan_duration_ms=int(config.get("ble_scan_duration_ms", 10000)),
advertising_refresh_interval_ms=int(config.get("ble_advertising_refresh_interval_ms", 60000)),
)
RNS.log(f"Android BLE Interface '{self.name}' initialized", RNS.LOG_INFO)
# Log configuration details if attributes are available
if hasattr(self, 'mode_str'):
RNS.log(f" Mode: {self.mode_str}", RNS.LOG_INFO)
if hasattr(self, 'enable_central'):
RNS.log(f" Central: {'Enabled' if self.enable_central else 'Disabled'}", RNS.LOG_INFO)
if hasattr(self, 'enable_peripheral'):
RNS.log(f" Peripheral: {'Enabled' if self.enable_peripheral else 'Disabled'}", RNS.LOG_INFO)
RNS.log(f" Max Peers: {self.max_peers}", RNS.LOG_INFO)
def get_rssi(self):
"""Get the RSSI of the most recently received message.
This method is called by signal_quality.py at message delivery time
to extract signal strength metrics.
Returns:
RSSI in dBm (negative integer), or None if unavailable
Note:
Android BLE does not provide per-packet RSSI like RNode hardware.
This returns the last known connection RSSI from the Kotlin bridge,
which is updated from the scanner cache during the connection.
"""
if hasattr(self, 'driver') and self.driver is not None:
return self.driver.get_last_receive_rssi()
return None
# Register this class as the interface entry point for Reticulum
interface_class = AndroidBLEInterface