Ptosiek/move-network-utilsからのマージプルリクエスト#38

Fix rfkill commands

modules/config.py

@@ -1,10 +1,9 @@
-import os
-import datetime
 import argparse
+import asyncio
+import datetime
 import json
 import logging
-import socket
-import asyncio
+import os
 import shutil
 import traceback
 import math
@@ -26,10 +25,17 @@
 from logger import CustomRotatingFileHandler, app_logger
 from modules.helper.setting import Setting
 from modules.button_config import Button_Config
-from modules.utils.cmd import exec_cmd, exec_cmd_return_value, is_running_as_service
+from modules.utils.cmd import (
+    exec_cmd,
+    exec_cmd_return_value,
+    is_running_as_service,
+)
 from modules.utils.timer import Timer
 
 
+BOOT_FILE = "/boot/config.txt"
+
+
 class Config:
     #######################
     # configurable values #
@@ -331,9 +337,6 @@ class Config:
     # for read load average in sensor_core
     G_PID = os.getpid()
 
-    # IP ADDRESS
-    G_IP_ADDRESS = ""
-
     # stopwatch state
     G_MANUAL_STATUS = "INIT"
     G_STOPWATCH_STATUS = "INIT"  # with Auto Pause
@@ -631,7 +634,7 @@ class Config:
     G_IMU_MAG_DECLINATION = 0.0
 
     # Bluetooth tethering
-    G_BT_ADDRESS = {}
+    G_BT_ADDRESSES = {}
     G_BT_USE_ADDRESS = ""
 
     # for track
@@ -818,16 +821,16 @@ async def delay_init(self):
             elif HAS_DBUS:
                 self.bt_pan = BTPanDbus()
             if HAS_DBUS_NEXT or HAS_DBUS:
-                await self.bt_pan.check_dbus()
-                if self.bt_pan.is_available:
-                    self.G_BT_ADDRESS = await self.bt_pan.find_bt_pan_devices()
+                is_available = await self.bt_pan.check_dbus()
+                if is_available:
+                    self.G_BT_ADDRESSES = await self.bt_pan.find_bt_pan_devices()
 
             try:
                 from modules.helper.ble_gatt_server import GadgetbridgeService
 
                 self.ble_uart = GadgetbridgeService(self)
-            except:
-                pass
+            except Exception as e:  # noqa
+                app_logger.info(f"Gadgetbridge service not initialized: {e}")
 
         # logger, sensor
         await self.gui.set_boot_status("initialize sensor...")
@@ -853,10 +856,9 @@ async def delay_init(self):
             ] = self.setting.get_config_pickle(
                 "AUTO_UPLOAD_VIA_BT", self.G_THINGSBOARD_API["AUTO_UPLOAD_VIA_BT"]
             )
-
             # resume BT tethering
             if (
-                self.G_BT_USE_ADDRESS != ""
+                self.G_BT_USE_ADDRESS
                 and not self.G_THINGSBOARD_API["AUTO_UPLOAD_VIA_BT"]
             ):
                 await self.bluetooth_tethering()
@@ -1013,14 +1015,6 @@ def reboot(self):
         if self.G_IS_RASPI:
             exec_cmd(["sudo", "reboot"])
 
-    def hardware_wifi_bt_on(self):
-        if self.G_IS_RASPI:
-            exec_cmd(["scripts/comment_out.sh"])
-
-    def hardware_wifi_bt_off(self):
-        if self.G_IS_RASPI:
-            exec_cmd(["scripts/uncomment.sh"])
-
     def update_application(self):
         if self.G_IS_RASPI:
             exec_cmd(["git", "pull", "origin", "master"])
@@ -1030,16 +1024,47 @@ def restart_application(self):
         if self.G_IS_RASPI:
             exec_cmd(["sudo", "systemctl", "restart", "pizero_bikecomputer"])
 
-    def detect_network(self):
-        try:
-            socket.setdefaulttimeout(3)
-            connect_interface = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            connect_interface.connect(("8.8.8.8", 53))
-            self.G_IP_ADDRESS = connect_interface.getsockname()[0]
-            return True
-        except socket.error:
-            self.G_IP_ADDRESS = "No address"
-            return False
+    def hardware_wifi_bt(self, status):
+        app_logger.info(f"Hardware Wifi/BT: {status}")
+        if self.G_IS_RASPI:
+            with open(BOOT_FILE, "r") as f:
+                data = f.read()
+            for dev in ["wifi", "bt"]:
+                disable = f"dtoverlay=disable-{dev}"
+                if status:
+                    if disable in data and f"#{disable}" not in data:
+                        # comment it
+                        exec_cmd(
+                            [
+                                "sudo",
+                                "sed",
+                                "-i",
+                                f"s/^dtoverlay\=disable\-{dev}/\#dtoverlay\=disable\-{dev}/",
+                                BOOT_FILE,
+                            ],
+                            False,
+                        )
+                    # else nothing to do it's not disabled then (not present or commented)
+                else:
+                    if f"#{disable}" in data:
+                        # uncomment it, so it's disabled
+                        exec_cmd(
+                            [
+                                "sudo",
+                                "sed",
+                                "-i",
+                                f"s/^\#dtoverlay\=disable\-{dev}/dtoverlay\=disable\-{dev}/",
+                                BOOT_FILE,
+                            ],
+                            False,
+                        )
+                    elif disable in data:
+                        # do nothing it's already the proper state...
+                        pass
+                    else:
+                        exec_cmd(
+                            ["sudo", "sed", "-i", f"$a{disable}", BOOT_FILE], False
+                        )
 
     def get_wifi_bt_status(self):
         if not self.G_IS_RASPI:
@@ -1048,15 +1073,15 @@ def get_wifi_bt_status(self):
         status = {"wlan": False, "bluetooth": False}
         try:
             # json option requires raspbian buster
-            raw_status = exec_cmd_return_value(["rfkill", "--json"], cmd_print=False)
+            raw_status = exec_cmd_return_value(["sudo", "rfkill", "--json"], cmd_print=False)
             json_status = json.loads(raw_status)
-            for l in json_status[""]:
-                if "type" not in l or l["type"] not in ["wlan", "bluetooth"]:
+            for device in json_status["rfkilldevices"]:
+                if "type" not in device or device["type"] not in ["wlan", "bluetooth"]:
                     continue
-                if l["soft"] == "unblocked":
-                    status[l["type"]] = True
-        except:
-            pass
+                if device["soft"] == "unblocked" and device["hard"] == "unblocked":
+                    status[device["type"]] = True
+        except Exception as e:
+            app_logger.warning(f"Exception occurred trying to get wifi/bt status: {e}")
         return status["wlan"], status["bluetooth"]
 
     def onoff_wifi_bt(self, key=None):
@@ -1066,33 +1091,29 @@ def onoff_wifi_bt(self, key=None):
 
         onoff_cmd = {
             "Wifi": {
-                True: ["rfkill", "block", "wifi"],
-                False: ["rfkill", "unblock", "wifi"],
+                True: ["sudo", "rfkill", "block", "wifi"],
+                False: ["sudo"", rfkill", "unblock", "wifi"],
             },
             "Bluetooth": {
-                True: ["rfkill", "block", "bluetooth"],
-                False: ["rfkill", "unblock", "bluetooth"],
+                True: ["sudo", "rfkill", "block", "bluetooth"],
+                False: ["sudo", "rfkill", "unblock", "bluetooth"],
             },
         }
         status = {}
         status["Wifi"], status["Bluetooth"] = self.get_wifi_bt_status()
         exec_cmd(onoff_cmd[key][status[key]])
 
     async def bluetooth_tethering(self, disconnect=False):
-        if not self.G_IS_RASPI:
-            return
-        if self.G_BT_USE_ADDRESS == "":
-            return
-        if self.bt_pan is None:
+        if not self.G_IS_RASPI or not self.G_BT_USE_ADDRESS or not self.bt_pan:
             return
 
         if not disconnect:
             res = await self.bt_pan.connect_tethering(
-                self.G_BT_ADDRESS[self.G_BT_USE_ADDRESS]
+                self.G_BT_ADDRESSES[self.G_BT_USE_ADDRESS]
             )
         else:
             res = await self.bt_pan.disconnect_tethering(
-                self.G_BT_ADDRESS[self.G_BT_USE_ADDRESS]
+                self.G_BT_ADDRESSES[self.G_BT_USE_ADDRESS]
             )
         return bool(res)
 

modules/helper/api.py

@@ -10,6 +10,7 @@
 
 import numpy as np
 
+from modules.utils.network import detect_network
 from logger import app_logger
 
 _IMPORT_GARMINCONNECT = False
@@ -59,11 +60,8 @@ def __init__(self, config):
             self.course_send_status = "RESET"
             self.bt_cmd_lock = False
 
-    async def get_google_route(self, x1, y1, x2, y2):
-        if (
-            not self.config.detect_network()
-            or self.config.G_GOOGLE_DIRECTION_API["TOKEN"] == ""
-        ):
+    async def get_google_routes(self, x1, y1, x2, y2):
+        if not detect_network() or self.config.G_GOOGLE_DIRECTION_API["TOKEN"] == "":
             return None
         if np.any(np.isnan([x1, y1, x2, y2])):
             return None
@@ -101,10 +99,7 @@ async def get_google_route_from_mapstogpx(self, url):
         return response
 
     async def get_openweathermap_data(self, x, y):
-        if (
-            not self.config.detect_network()
-            or self.config.G_OPENWEATHERMAP_API["TOKEN"] == ""
-        ):
+        if detect_network() or self.config.G_OPENWEATHERMAP_API["TOKEN"] == "":
             return None
         if np.any(np.isnan([x, y])):
             return None
@@ -120,7 +115,7 @@ async def get_openweathermap_data(self, x, y):
 
     async def get_ridewithgps_route(self, add=False, reset=False):
         if (
-            not self.config.detect_network()
+            detect_network()
             or self.config.G_RIDEWITHGPS_API["APIKEY"] == ""
             or self.config.G_RIDEWITHGPS_API["TOKEN"] == ""
         ):
@@ -287,7 +282,7 @@ def check_ridewithgps_files(self, route_id, mode):
 
     def upload_check(self, blank_check, blank_msg, file_check=True):
         # network check
-        if not self.config.detect_network():
+        if not detect_network():
             app_logger.warning("No Internet connection")
             return False
 
@@ -448,7 +443,7 @@ async def rwgps_upload(self):
 
     async def get_scw_list(self, wind_config, mode=None):
         # network check
-        if not self.config.detect_network():
+        if not detect_network():
             app_logger.warning("No Internet connection")
             return None
 
@@ -482,7 +477,7 @@ def get_strava_cookie(self):
             app_logger.warning("Install stravacookies")
             return
 
-        if not self.config.detect_network():
+        if not detect_network():
             return None
 
         strava_cookie = StravaCookieFetcher()
@@ -520,7 +515,7 @@ def check_livetrack(self):
             return False
         # network check
         if (
-            not self.config.detect_network()
+            not detect_network()
             and not self.config.G_THINGSBOARD_API["AUTO_UPLOAD_VIA_BT"]
         ):
             # print("No Internet connection")
@@ -553,7 +548,7 @@ async def send_livetrack_data_internal(self, quick_send=False):
 
             while (
                 bt_pan_status
-                and not self.config.detect_network()
+                and not detect_network()
                 and count < self.config.G_THINGSBOARD_API["TIMEOUT_SEC"]
             ):
                 await asyncio.sleep(1)
@@ -568,7 +563,7 @@ async def send_livetrack_data_internal(self, quick_send=False):
                 await asyncio.sleep(5)
                 self.bt_cmd_lock = False
                 app_logger.error(
-                    f"[BT] {timestamp_str} connect error, network status: {self.config.detect_network()}"
+                    f"[BT] {timestamp_str} connect error, network status: {detect_network()}"
                 )
                 self.config.logger.sensor.values["integrated"]["send_time"] = (
                     datetime.datetime.now().strftime("%H:%M") + "CE"
@@ -637,7 +632,7 @@ async def send_livetrack_data_internal(self, quick_send=False):
         if self.config.G_THINGSBOARD_API["AUTO_UPLOAD_VIA_BT"]:
             bt_pan_status = await self.config.bluetooth_tethering(disconnect=True)
             self.bt_cmd_lock = False
-            network_status = self.config.detect_network()
+            network_status = detect_network()
             # print("[BT] {} disconnect, network status:{}".format(timestamp_str, network_status))
             if network_status:
                 self.config.logger.sensor.values["integrated"]["send_time"] = (

modules/helper/bt_pan.py

@@ -26,7 +26,6 @@
 
 class BTPan:
     bus = None
-    is_available = False
     devices = {}
 
     obj_bluez = "org.bluez"
@@ -45,10 +44,11 @@ class BTPanDbusNext(BTPan):
     async def check_dbus(self):
         try:
             self.bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
-            introspection = await self.bus.introspect(self.obj_bluez, self.path_bluez)
-            self.is_available = True
-        except:
-            pass
+            await self.bus.introspect(self.obj_bluez, self.path_bluez)
+            return True
+        except Exception as e:
+            app_logger.warning(f"DBus not available {e}")
+            return False
 
     async def find_bt_pan_devices(self):
         res = {}
@@ -123,10 +123,11 @@ class BTPanDbus(BTPan):
     async def check_dbus(self):
         try:
             self.bus = dbus.SystemBus()
-            proxy = self.bus.get_name_owner(self.obj_bluez)
-            self.is_available = True
-        except:
-            pass
+            self.bus.get_name_owner(self.obj_bluez)
+            return True
+        except Exception as e:
+            app_logger.warning(f"DBus not available {e}")
+            return False
 
     def get_managed_objects(self):
         manager = dbus.Interface(

modules/helper/network.py

@@ -9,6 +9,7 @@
 
 from logger import app_logger
 from modules.helper.api import api
+from modules.utils.network import detect_network
 
 
 class Network:
@@ -81,7 +82,7 @@ async def download_worker(self):
     async def download_maptile(
         self, map_config, map_name, z, tiles, additional_download=False
     ):
-        if not self.config.detect_network() or map_config[map_name]["url"] is None:
+        if not detect_network() or map_config[map_name]["url"] is None:
             return False
 
         urls = []
@@ -242,7 +243,7 @@ async def download_files(self, urls, save_paths, headers=None, params=None):
         return res
 
     async def download_demtile(self, z, x, y):
-        if not self.config.detect_network():
+        if not detect_network():
             return False
         header = {}
         try: