Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions BlocksScreen/configfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,40 @@ def add_option(
f'Unable to add "{option}" option to section "{section}": {e} '
)

def update_option(
self,
section: str,
option: str,
value: typing.Any,
) -> None:
"""Update an existing option's value in both raw tracking and configparser."""
try:
with self.file_lock:
if not self.config.has_section(section):
self.add_section(section)

if not self.config.has_option(section, option):
self.add_option(section, option, str(value))
return

line_idx = self._find_option_line_index(section, option)
self.raw_config[line_idx] = f"{option}: {value}"
self.config.set(section, option, str(value))
self.update_pending = True
except Exception as e:
logging.error(
f'Unable to update option "{option}" in section "{section}": {e}'
)

def _find_option_line_index(self, section: str, option: str) -> int:
"""Find the index of an option line within a specific section."""
start, end = self._find_section_limits(section)
opt_regex = re.compile(rf"^\s*{re.escape(option)}\s*[:=]")
for i in range(start + 1, end):
if opt_regex.match(self.raw_config[i]):
return i
raise configparser.Error(f'Option "{option}" not found in section "{section}"')

def save_configuration(self) -> None:
"""Save teh configuration to file"""
try:
Expand Down
53 changes: 47 additions & 6 deletions BlocksScreen/lib/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import typing
from uuid import uuid4

from configfile import BlocksScreenConfig, get_configparser

import sdbus
from PyQt6 import QtCore
from sdbus_async import networkmanager as dbusNm
Expand Down Expand Up @@ -59,8 +61,9 @@ def __init__(self) -> None:
logger.info(
f"Sdbus NetworkManager Monitor Thread {self.listener_thread.name} Running"
)
self.hotspot_ssid: str = "PrinterHotspot"
self.hotspot_password: str = "123456789"

self.config: BlocksScreenConfig = get_configparser()
self.load_hotspot_configuration()
self.check_connectivity()
self.available_wired_interfaces = self.get_wired_interfaces()
self.available_wireless_interfaces = self.get_wireless_interfaces()
Expand All @@ -78,10 +81,38 @@ def __init__(self) -> None:
wired_interfaces[0] if wired_interfaces else None
)

self.create_hotspot(self.hotspot_ssid, self.hotspot_password)
if self.primary_wifi_interface:
self.rescan_networks()

def load_hotspot_configuration(self):
self.config.load_config()
if not self.config.has_section("hotspot"):
self.config.add_section("hotspot")
self.config.save_configuration()

self.hotspot_config = self.config.get_section("hotspot", fallback=None)

if self.hotspot_config.has_option("ssid"):
ssid = self.hotspot_config.get("ssid", parser=str, default="PrinterHotspot")
self.hotspot_ssid = ssid if isinstance(ssid, str) else "PrinterHotspot"
else:
self.config.add_option("hotspot", "ssid", "PrinterHotspot")
self.hotspot_ssid = "PrinterHotspot"

if self.hotspot_config.has_option("password"):
password = self.hotspot_config.get(
"password", parser=str, default="123456789"
)
self.hotspot_password = (
password if isinstance(password, str) else "123456789"
)
else:
self.config.add_option("hotspot", "password", "123456789")
self.hotspot_password = "123456789"

self.config.save_configuration()
self.create_hotspot(self.hotspot_ssid, self.hotspot_password)

def _listener_run_loop(self) -> None:
try:
asyncio.set_event_loop(self.loop)
Expand Down Expand Up @@ -1380,11 +1411,18 @@ def create_hotspot(
password (str, optional): connection password. Defaults to "123456789".
"""
if self.is_known(ssid):
self.delete_network(ssid)
self.delete_network(self.hotspot_ssid)
logger.debug("old hotspot deleted")
try:
self.delete_network(ssid)
self.delete_network(self.hotspot_ssid)
# psk = hashlib.sha256(password.encode()).hexdigest()
self.config.update_option("hotspot", "ssid", ssid)
self.config.update_option("hotspot", "password", password)

self.hotspot_ssid = ssid
self.hotspot_password = password

self.config.save_configuration()
_properties: dbusNm.NetworkManagerConnectionProperties = {
"connection": {
"id": ("s", str(ssid)),
Expand All @@ -1409,7 +1447,6 @@ def create_hotspot(
},
"ipv6": {"method": ("s", "ignore")},
}

tasks = [
self.loop.create_task(
dbusNm.NetworkManagerSettings(bus=self.system_dbus).add_connection(
Expand Down Expand Up @@ -1503,7 +1540,11 @@ def update_connection_settings(

if ssid == self.hotspot_ssid and new_ssid:
self.hotspot_ssid = new_ssid
self.config.update_option("hotspot", "password", new_ssid)
if password != self.hotspot_password and password:
self.hotspot_password = password
self.config.update_option("hotspot", "password", password)

self.config.save_configuration()
except Exception as e:
logger.error("Caught Exception while updating network: %s", e)
8 changes: 0 additions & 8 deletions BlocksScreen/lib/panels/networkWindow.py
Original file line number Diff line number Diff line change
Expand Up @@ -2595,14 +2595,6 @@ def _evaluate_network_state(self, nm_state: str = "") -> None:

def _handle_first_run_state(self) -> None:
"""Handle initial state on first run."""
saved_networks = self._sdbus_network.get_saved_networks_with_for()

old_hotspot = next(
(n for n in saved_networks if "ap" in str(n.get("mode", ""))), None
)
if old_hotspot:
self.hotspot_name_input_field.setText(old_hotspot["ssid"])

connectivity = self._sdbus_network.check_connectivity()
wifi_btn = self.wifi_button.toggle_button
hotspot_btn = self.hotspot_button.toggle_button
Expand Down