Wifi connection menu with textual

.pre-commit-config.yaml

@@ -41,7 +41,9 @@ repos:
       additional_dependencies:
         - pydantic
         - pytest
+        - pytest-mock
         - cryptography
+        - textual
   - repo: local
     hooks:
       - id: pylint

PKGBUILD

@@ -28,6 +28,7 @@ depends=(
   'python-cryptography'
   'python-pydantic'
   'python-pyparted'
+  'python-textual'
   'systemd'
   'util-linux'
   'xfsprogs'

archinstall/__init__.py

@@ -8,7 +8,10 @@
 
 from archinstall.lib.args import arch_config_handler
 from archinstall.lib.disk.utils import disk_layouts
+from archinstall.lib.network.wifi_handler import wifi_handler
+from archinstall.lib.networking import ping
 from archinstall.lib.packages.packages import check_package_upgrade
+from archinstall.tui.ui.components import tui as ttui
 
 from .lib.hardware import SysInfo
 from .lib.output import FormattedOutput, debug, error, info, log, warn
@@ -36,6 +39,17 @@ def _log_sys_info() -> None:
 	debug(f'Disk states before installing:\n{disk_layouts()}')
 
 
+def _check_online() -> None:
+	try:
+		ping('1.1.1.1')
+	except OSError as ex:
+		if 'Network is unreachable' in str(ex):
+			if not arch_config_handler.args.skip_wifi_check:
+				success = not wifi_handler.setup()
+				if not success:
+					exit(0)
+
+
 def _fetch_arch_db() -> None:
 	info('Fetching Arch Linux package database...')
 	try:
@@ -44,13 +58,14 @@ def _fetch_arch_db() -> None:
 		error('Failed to sync Arch Linux package database.')
 		if 'could not resolve host' in str(e).lower():
 			error('Most likely due to a missing network connection or DNS issue.')
+
 		error('Run archinstall --debug and check /var/log/archinstall/install.log for details.')
 
 		debug(f'Failed to sync Arch Linux package database: {e}')
 		exit(1)
 
 
-def _check_new_version() -> None:
+def check_version_upgrade() -> str | None:
 	info('Checking version...')
 	upgrade = None
 
@@ -62,7 +77,7 @@ def _check_new_version() -> None:
 
 	text = tr('New version available') + f': {upgrade}'
 	info(text)
-	time.sleep(3)
+	return text
 
 
 def main() -> int:
@@ -81,11 +96,19 @@ def main() -> int:
 
 	_log_sys_info()
 
+	ttui.global_header = 'Archinstall'
+
 	if not arch_config_handler.args.offline:
+		_check_online()
 		_fetch_arch_db()
 
 		if not arch_config_handler.args.skip_version_check:
-			_check_new_version()
+			new_version = check_version_upgrade()
+
+			if new_version:
+				ttui.global_header = f'{ttui.global_header} {new_version}'
+				info(new_version)
+				time.sleep(3)
 
 	script = arch_config_handler.get_script()
 

archinstall/lib/args.py

@@ -49,6 +49,7 @@ class Arguments:
 	no_pkg_lookups: bool = False
 	plugin: str | None = None
 	skip_version_check: bool = False
+	skip_wifi_check: bool = False
 	advanced: bool = False
 	verbose: bool = False
 
@@ -407,6 +408,12 @@ def _define_arguments(self) -> ArgumentParser:
 			default=False,
 			help='Skip the version check when running archinstall',
 		)
+		parser.add_argument(
+			'--skip-wifi-check',
+			action='store_true',
+			default=False,
+			help='Skip wifi check when running archinstall',
+		)
 		parser.add_argument(
 			'--advanced',
 			action='store_true',

archinstall/lib/models/network.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import TYPE_CHECKING, NotRequired, TypedDict
+from typing import TYPE_CHECKING, NotRequired, TypedDict, override
 
+from archinstall.lib.output import debug
 from archinstall.lib.translationhandler import tr
 
 from ..models.profile import ProfileConfiguration
@@ -157,3 +159,101 @@ def install_network_config(
 
 				installation.enable_service('systemd-networkd')
 				installation.enable_service('systemd-resolved')
+
+
+@dataclass
+class WifiNetwork:
+	bssid: str
+	frequency: str
+	signal_level: str
+	flags: str
+	ssid: str
+
+	@override
+	def __hash__(self) -> int:
+		return hash((self.bssid, self.frequency, self.signal_level, self.flags, self.ssid))
+
+	def table_data(self) -> dict[str, str | int]:
+		"""Format WiFi data for table display"""
+		return {
+			'SSID': self.ssid,
+			'Signal': f'{self.signal_level} dBm',
+			'Frequency': f'{self.frequency} MHz',
+			'Security': self.flags,
+			'BSSID': self.bssid,
+		}
+
+	@staticmethod
+	def from_wpa(results: str) -> list[WifiNetwork]:
+		entries: list[WifiNetwork] = []
+
+		for line in results.splitlines():
+			line = line.strip()
+			if not line:
+				continue
+
+			parts = line.split()
+			if len(parts) != 5:
+				continue
+
+			wifi = WifiNetwork(bssid=parts[0], frequency=parts[1], signal_level=parts[2], flags=parts[3], ssid=parts[4])
+			entries.append(wifi)
+
+		return entries
+
+
+@dataclass
+class WifiConfiguredNetwork:
+	network_id: int
+	ssid: str
+	bssid: str
+	flags: list[str]
+
+	@classmethod
+	def from_wpa_cli_output(cls, list_networks: str) -> list[WifiConfiguredNetwork]:
+		"""
+		Example output from 'wpa_cli list_networks'
+
+		Selected interface 'wlan0'
+		network id / ssid / bssid / flags
+		0	WifiGuest any	[CURRENT]
+		1		any [DISABLED]
+		2		any [DISABLED]
+		"""
+
+		lines = list_networks.strip().splitlines()
+		lines = lines[1:]  # remove the header row from the wpa_cli output
+
+		networks = []
+
+		for line in lines:
+			line = line.strip()
+			parts = line.split('\t')
+
+			if len(parts) < 3:
+				continue
+
+			try:
+				# flags = cls._extract_flags(parts[3])
+				flags: list[str] = []
+
+				networks.append(
+					WifiConfiguredNetwork(
+						network_id=int(parts[0]),
+						ssid=parts[1],
+						bssid=parts[2],
+						flags=flags,
+					)
+				)
+			except (ValueError, IndexError):
+				debug('Parsing error for network output')
+
+		return networks
+
+	@classmethod
+	def _extract_flags(cls, flag_string: str) -> list[str]:
+		pattern = r'\[([^\]]+)\]'
+
+		extracted_values = re.findall(pattern, flag_string)
+
+		return extracted_values