Merge remote-tracking branch 'refs/remotes/origin/network' into dev

# Conflicts:
#	poetry.lock

pyproject.toml

@@ -19,6 +19,7 @@ kivy = {version = "2.3.0"}
 docutils = "^0.21.2"
 keke = "^0.1.4"
 numpy = "^1.26.4"
+sdbus-networkmanager = "^2.0.0"
 
 [build-system]
 requires = ["poetry-core"]

rotary_controller_python/components/forms/dropdown_item.kv

@@ -0,0 +1,23 @@
+<DropDownItem>:
+  main_button: main_button
+  height: 36
+  size_hint_y: None
+  BoxLayout:
+    orientation: "horizontal"
+    Button:
+      size_hint_x: None
+      width: self.height
+      font_name: "fonts/Font Awesome 6 Free-Solid-900.otf"
+      font_size: 21
+      text: "\uf129"
+    Label:
+      size_hint_x: 0.7
+      text: root.name
+    Button:
+      id: main_button
+      size_hint_x: None
+      width: 250
+#      multiline: False
+#      text: 'ON' if root.value else 'OFF'
+#      state: 'down' if root.value else 'normal'
+      on_release: root.dropdown.open(self)

rotary_controller_python/components/forms/dropdown_item.py

@@ -0,0 +1,46 @@
+import os
+
+from kivy.lang import Builder
+from kivy.logger import Logger
+from kivy.properties import StringProperty, ListProperty, ObjectProperty
+from kivy.uix.boxlayout import BoxLayout
+from kivy.uix.dropdown import DropDown
+from kivy.uix.button import Button
+
+
+log = Logger.getChild(__name__)
+kv_file = os.path.join(os.path.dirname(__file__), __file__.replace(".py", ".kv"))
+if os.path.exists(kv_file):
+    log.info(f"Loading KV file: {kv_file}")
+    Builder.load_file(kv_file)
+
+
+class DropDownItem(BoxLayout):
+    name = StringProperty("")
+    value = StringProperty(False)
+    options = ListProperty([])
+    dropdown = ObjectProperty()
+    main_button: Button = ObjectProperty()
+
+    def __init__(self, **kv):
+        super().__init__(**kv)
+        self.dropdown = DropDown()
+        self._options = []
+        self.dropdown.bind(on_select=lambda instance, x: setattr(self, 'value', x))
+
+    def delete_all_dropdown_options(self):
+        for item in self._options:
+            self.dropdown.remove_widget(item)
+
+    def on_value(self, instance, value):
+        self.main_button.text = value
+
+    def on_options(self, instance, value):
+        # Clean any existing
+        self.delete_all_dropdown_options()
+
+        for item in self.options:
+            btn = Button(text=item, size_hint_y=None, height=44)
+            btn.bind(on_release=lambda btn: self.dropdown.select(btn.text))
+            self.dropdown.add_widget(btn)
+            self._options.append(btn)

rotary_controller_python/components/setup/network_panel.kv

@@ -1,5 +1,6 @@
 #: import StringItem components.forms.string_item
 #: import BooleanItem components.forms.boolean_item
+#: import DropDownItem components.forms.dropdown_item
 
 <NetworkPanel>:
   orientation: "vertical"
@@ -11,10 +12,16 @@
       cols: 1
       spacing: 1
       size_hint_y: None
-      StringItem:
+      DropDownItem:
         name: "Network Interface"
-        value: root.device
-        on_value: root.device = self.value
+        options: root.get_all_network_interfaces()
+        value: root.get_all_network_interfaces()[0] if len(root.get_all_network_interfaces()) > 0 else ""
+
+      DropDownItem:
+        name: "Connection Method"
+        options: ["auto", "manual"]
+        value: root.connection_method
+        on_value: root.connection_method = self.value
 
       StringItem:
         name: "Wifi Network"
@@ -26,33 +33,25 @@
         value: root.wpa_psk
         on_value: root.wpa_psk = self.value
 
-      BooleanItem:
-        name: "Use DHCP"
-        value: root.dhcp
-        on_value: root.dhcp = self.value
-
       StringItem:
         name: "IP Address"
         value: root.address
-        disabled: root.dhcp
+        disabled: root.connection_method == "auto"
         on_value: root.address = self.value
 
       StringItem:
         name: "Netmask Bits"
         value: root.netmask
-        disabled: root.dhcp
+        disabled: root.connection_method == "auto"
         on_value: root.netmask = self.value
 
       StringItem:
         name: "Default Gateway"
         value: root.gateway
-        disabled: root.dhcp
+        disabled: root.connection_method == "auto"
         on_value: root.gateway = self.value
 
 
-      BoxLayout:
-        orientation: "horizontal"
-
       BoxLayout:
         size_hint_y: None
         height: 64
@@ -72,5 +71,10 @@
 
         Button:
           text: "Apply"
+          background_color: "#3acf3a"
+          on_release: root.apply()
+
+        Button:
+          text: "Disable"
           background_color: "#cf3a3a"
-          on_release: root.test_configuration()
+          on_release: root.disable()

rotary_controller_python/components/setup/network_panel.py

@@ -1,12 +1,13 @@
 import os
 
-from kivy.properties import StringProperty, BooleanProperty
+from kivy.clock import Clock
+from kivy.properties import StringProperty, ObjectProperty
 from kivy.logger import Logger
 from kivy.uix.boxlayout import BoxLayout
 from kivy.lang import Builder
 
-from rotary_controller_python.network import read_interfaces, render_interfaces, reload_interfaces
-from rotary_controller_python.network.models import NetworkInterface, Wireless
+from rotary_controller_python.network.networkmanager import get_all_network_interface_names, get_profile_by_id, get_psk, \
+    get_ssid, get_connection_method, activate_connection, deactivate_connection, enable_wifi, disable_wifi, get_ipv4
 
 log = Logger.getChild(__name__)
 kv_file = os.path.join(os.path.dirname(__file__), __file__.replace(".py", ".kv"))
@@ -17,71 +18,50 @@
 
 class NetworkPanel(BoxLayout):
     device = StringProperty("")
-    dhcp = BooleanProperty(False)
     address = StringProperty("")
     netmask = StringProperty("")
     gateway = StringProperty("")
     wpa_ssid = StringProperty("")
     wpa_psk = StringProperty("")
     status_text = StringProperty("Ready")
+    profile = ObjectProperty()
+    connection_method = StringProperty("")
+    refresh_task = ObjectProperty()
 
     def __init__(self, **kv):
         super().__init__(**kv)
         self.ids['grid_layout'].bind(minimum_height=self.ids['grid_layout'].setter('height'))
 
-        configuration: NetworkInterface = read_interfaces()
-        self.device = configuration.name
-        self.dhcp = configuration.dhcp
-        if configuration.wireless is not None:
-            if configuration.wireless.ssid is not None:
-                self.wpa_ssid = configuration.wireless.ssid
-            if configuration.wireless.password is not None:
-                self.wpa_psk = configuration.wireless.password
+        self.profile = get_profile_by_id()
+        self.wpa_psk = get_psk(self.profile)
+        self.wpa_ssid = get_ssid(self.profile)
+        self.connection_method = get_connection_method(self.profile)
+        self.apply_thread = None
+        self.disable_thread = None
+        self.refresh_task = Clock.schedule_interval(self.refresh_thread, 1)
 
-        if configuration.address is not None and len(configuration.address) > 0:
-            self.address = configuration.address
+    def refresh_thread(self, *args):
+        address, netmask, gateway = get_ipv4(self.profile)
+        self.address = address
+        self.netmask = netmask
+        self.gateway = gateway
+        # self.status_text = self.status_text + "A"
 
-        if configuration.netmask is not None:
-            self.netmask = str(configuration.netmask)
+    def activate_connection(self, *args):
+        activate_connection(self.profile)
+        self.status_text = "Connection Ready"
 
-        if configuration.gateway is not None and len(configuration.gateway) > 0:
-            self.gateway = configuration.gateway
+    def enable_wifi(self, *args):
+        self.status_text = "Enabling Wifi Device"
+        enable_wifi()
+        self.status_text = "Enabling Connection"
+        Clock.schedule_once(self.activate_connection, timeout=2)
 
-    def confirm(self):
-        try:
-            configuration = NetworkInterface(
-                name=self.device,
-                dhcp=self.dhcp,
-                address=self.address if not self.dhcp else None,
-                gateway=self.gateway if not self.dhcp else None,
-                netmask=int(self.netmask) if not self.dhcp else None,
-                wireless=Wireless(
-                    password=self.wpa_psk,
-                    ssid=self.wpa_ssid
-                )
-            )
-            render_interfaces(configuration=configuration)
-            status = reload_interfaces()
-        except Exception as e:
-            log.exception(e.__str__())
-            status = e.__str__()
+    def apply(self):
+        Clock.schedule_once(self.enable_wifi)
 
-        self.status_text = status
+    def disable(self):
+        disable_wifi()
 
-    def test_configuration(self):
-        try:
-            configuration = NetworkInterface(
-                name=self.device,
-                dhcp=self.dhcp,
-                address=self.address if not self.dhcp else None,
-                gateway=self.gateway if not self.dhcp else None,
-                netmask=int(self.netmask) if not self.dhcp else None,
-                wireless=Wireless(
-                    password=self.wpa_psk,
-                    ssid=self.wpa_ssid
-                )
-            )
-            render_interfaces(configuration=configuration)
-            status = reload_interfaces()
-        except Exception as e:
-            status = e.__str__()
+    def get_all_network_interfaces(self):
+        return get_all_network_interface_names()

rotary_controller_python/network/__init__.py

@@ -14,7 +14,7 @@
 def read_interfaces(config_path: str = "/etc/network/interfaces"):
     interface = models.NetworkInterface(
         name="wlan0",
-        dhcp=False,
+        dhcp=True,
         wireless=models.Wireless(
             password="",
             ssid=""
@@ -93,13 +93,13 @@ def reload_interfaces():
         log.error(e.__str__())
 
 
-def read_wlan_status():
+def read_wlan_status() -> RfkillStatus:
     try:
         result = subprocess.run(["/usr/sbin/rfkill", "-J", "--output-all"], capture_output=True)
         result.check_returncode()
         json_data = result.stdout
         rfkill_data = json.loads(json_data)
-        wlan_data = [v for k,v in rfkill_data.items()][0]
+        wlan_data = [v for k, v in rfkill_data.items()][0]
         wlan_data = [item for item in wlan_data if item['type'] == 'wlan']
         if len(wlan_data) == 0:
             raise Exception("No wlan detected")