pybatmesh/naxalnet/iwd.py

#!/usr/bin/env python3

"""Manage wifi adapter via iwd D-Bus api"""

from dasbus.connection import SystemMessageBus

IWD_BUS = "net.connman.iwd"
IWD_ROOT_PATH = "/"
IWD_DEVICE_INTERFACE = "net.connman.iwd.Device"
IWD_ADAPTER_INTERFACE = "net.connman.iwd.Adapter"

# If you are new to D-Bus, you might want to use a program
# such as D-Feet (https://wiki.gnome.org/Apps/DFeet) for reference.


class IWD:
    """Manage iwd via dbus"""

    def __init__(self, bus=SystemMessageBus()):
        # self._bus and self._proxy are meant for use only in this file
        self._bus = bus
        self.reload()

    def reload(self):
        """reload the proxy"""
        self._proxy = self._bus.get_proxy(IWD_BUS, IWD_ROOT_PATH)

    def get_name_from_path(self, path: str) -> str:
        """
        returns device or adapter name when d-bus path is given as arg
        """
        proxy = self._bus.get_proxy(IWD_BUS, path)
        return proxy.Name

    def get_device_path_from_name(self, name: str) -> str:
        """returns path of device as str"""
        device_paths = self.get_all_device_paths()
        for i in device_paths:
            proxy = self._bus.get_proxy(IWD_BUS, i)
            if proxy.Name == name:
                return i
            # If no devices were found, return None
        return None

    def get_adapter_path_from_name(self, name: str) -> str:
        """returns path of adapter as str"""
        adapter_paths = self.get_all_adapter_paths()
        for i in adapter_paths:
            proxy = self._bus.get_proxy(IWD_BUS, i)
            if proxy.Name == name:
                return i
            # If no adapters were found
        return None

    def get_all_device_paths(self) -> list:
        """returns list of paths of all devices"""
        objects = self._proxy.GetManagedObjects()
        paths = []
        for key, value in objects.items():
            # if value is a device, add its path to paths
            if IWD_DEVICE_INTERFACE in value:
                paths.append(key)
        return paths

    def get_all_adapter_paths(self) -> list:
        """returns list of paths of all adapters"""
        objects = self._proxy.GetManagedObjects()
        paths = []

        for key, value in objects.items():
            # if value is an adapter, add its path to paths
            if IWD_ADAPTER_INTERFACE in value:
                paths.append(key)
        return paths

    def get_devices(self) -> list:
        """
        returns list of device names as str
        example: ["wlan0", "wlan1"]
        """
        devices = []
        device_paths = self.get_all_device_paths()

        for i in device_paths:
            name = self.get_name_from_path(i)
            devices.append(name)
        return devices

    def get_adapters(self) -> list:
        """
        returns list of adapters
        example: ["phy0","phy1"]
        """
        adapters = []
        adapter_paths = self.get_all_adapter_paths()

        for i in adapter_paths:
            name = self.get_name_from_path(i)
            adapters.append(name)
        return adapters


class Device:
    """
    control devices with iwd
    name: name of device (str)
    adapter: name of adapter (str)
    """

    def __init__(self, name: str, bus=SystemMessageBus()):
        self._iwd = IWD(bus)
        self._bus = self._iwd._bus
        self._path = self._iwd.get_device_path_from_name(name)
        self.reload()
        self.name = self._proxy.Name
        adapter_path = self._proxy.Adapter
        self.adapter = self._iwd.get_name_from_path(adapter_path)

    def is_powered_on(self) -> bool:
        """returns True if devie is powered on"""
        return self._proxy.Powered

    def power_on(self):
        """Turn on the device and reload the proxy"""
        self._proxy.Powered = True
        self.reload()

    def power_off(self):
        """Turn off the device and reload the proxy"""
        self._proxy.Powered = False
        self.reload()

    def reload(self):
        """reload the proxy after changing mode"""
        self._proxy = self._bus.get_proxy(IWD_BUS, self._path)

    def is_adhoc_started(self):
        """
        Returns True if an adhoc network is started on this device.
        Returns None if device is not powered on or not in ad-hoc mode.
        """
        if self.is_powered_on() and self.get_mode() == "ad-hoc":
            return self._proxy.Started
        # If above condition is not true, return None
        return None

    def is_ap_started(self):
        """
        Same as is_adhoc_started(), but for ap
        """
        if self.is_powered_on() and self.get_mode == "ap":
            return self._proxy.Started
        return None

    def get_mode(self) -> str:
        """
        returns the mode in which the device is in
        example: "ap"
        """
        return self._proxy.Mode

    def set_mode(self, mode: str):
        """change the device mode to mode"""
        self._proxy.Mode = mode
        self.reload()

    def start_adhoc_open(self, name: str):
        """
        Create ad-hoc network with name, changing mode to ad-hoc
        if it isn't already on ad-hoc
        """
        if self.get_mode() != "ad-hoc":
            self.set_mode("ad-hoc")

        if not self.is_powered_on():
            self.power_on()

        self._proxy.StartOpen(name)