Skip to content
Merged
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
17 changes: 16 additions & 1 deletion micro/app_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,14 @@ def IRQ_button_press(pin):
status_led=None,
timeout_seconds=config["wifi"]["timeout_seconds"],
)
network.connect_with_fallback(watchdog_feed=watchdog.feed)
printer_api_ready = config_loader.is_runtime_config_ready(config)
if printer_api_ready:
network.connect_with_fallback(watchdog_feed=watchdog.feed)
else:
# Keep the setup UI reachable until a complete, explicit printer
# configuration has been saved. In particular, do not poll or clear the
# baked-in/default printer ID while settings are incomplete.
network.start_access_point()
if network.is_ap_mode():
print("Wi-Fi unavailable; connect to the setup access point to update settings")
flasher.on()
Expand Down Expand Up @@ -117,6 +124,10 @@ def handle_pending_button_press():
global PENDING_BUTTON_PRESS
global PRINTER_STATUS_UPDATE_REQUIRED

if not printer_api_ready:
PENDING_BUTTON_PRESS = False
return

try:
with_network_connection(
lambda: api.clear_plate(config["printer"]["id"])
Expand All @@ -133,6 +144,10 @@ def handle_printer_status_update():
global PRINTER_AWAITING_PLATE_CLEAR, PRINTER_STATUS_UPDATE_REQUIRED
global CHAMBER_LIGHT_IS_ON

if not printer_api_ready:
PRINTER_STATUS_UPDATE_REQUIRED = False
return

try:
response = with_network_connection(
lambda: api.get_printer_status(config["printer"]["id"])
Expand Down
170 changes: 164 additions & 6 deletions micro/config_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
except ImportError:
import json

import os
import random


Expand Down Expand Up @@ -36,7 +37,9 @@ def generate_hostname():
"request_timeout_seconds": 3,
},
"printer": {
"id": 3,
# A printer must be selected in setup. Never guess a device ID: a
# default ID can send a button press to an unrelated printer.
"id": None,
"poll_interval_seconds": 5,
},
"led": {
Expand All @@ -55,20 +58,175 @@ def generate_hostname():
}


_STRING_FIELDS = {
"wifi": ("ssid", "password", "hostname", "ap_password"),
"api": ("base_url", "key"),
"button": ("pull", "trigger"),
"web": ("password",),
}
_NUMBER_FIELDS = {
"wifi": ("timeout_seconds",),
"api": ("request_timeout_seconds",),
"printer": ("poll_interval_seconds",),
"led": ("pin", "flash_interval_ms"),
"button": ("pin", "debounce_ms"),
}
_KNOWN_SECTIONS = tuple(DEFAULT_CONFIG.keys())


def load_config(path="config.json"):
config = _copy_dict(DEFAULT_CONFIG)

try:
with open(path) as config_file:
loaded_config = json.load(config_file)
except OSError:
print("Config file not found, using defaults:", path)
path = str(path)
loaded_config = _load_config_file(path)
source_path = path
if loaded_config is None:
backup_path = path + ".bak"
loaded_config = _load_config_file(backup_path)
if loaded_config is not None:
source_path = backup_path

if loaded_config is None:
print("Config file missing or invalid, using defaults:", path)
return config

_deep_update(config, loaded_config)
if source_path != path:
print("Recovered configuration from backup:", source_path)
return config


def is_runtime_config_ready(config):
"""Return whether the settings can safely address a printer API."""
if not isinstance(config, dict):
return False

wifi = config.get("wifi")
api = config.get("api")
printer = config.get("printer")
if not isinstance(wifi, dict) or not isinstance(api, dict):
return False
if not isinstance(printer, dict):
return False

printer_id = printer.get("id")
return bool(
wifi.get("ssid")
and api.get("base_url")
and api.get("key")
and isinstance(printer_id, int)
and not isinstance(printer_id, bool)
and printer_id >= 0
)


def save_config(path, config):
"""Replace a config atomically, retaining the previous copy until verified."""
path = str(path)
if not isinstance(config, dict):
raise ValueError("Configuration must be an object")

temporary_path = path + ".tmp"
backup_path = path + ".bak"
serialized = json.dumps(config)
with open(temporary_path, "w") as config_file:
config_file.write(serialized)
config_file.write("\n")
sync = getattr(os, "sync", None)
if sync is not None:
sync()

staged_config = _load_config_file(temporary_path)
if staged_config != config:
_remove_file(temporary_path)
raise ValueError("Could not verify saved configuration")

if _load_config_file(path) is not None:
_remove_file(backup_path)
os.rename(path, backup_path)
else:
# Keep any valid prior backup until the new config is installed.
_remove_file(path)

try:
os.rename(temporary_path, path)
except OSError:
# Restore a known-good config when the replacement could not be made.
if (
_load_config_file(path) is None
and _load_config_file(backup_path) is not None
):
try:
os.rename(backup_path, path)
except OSError:
pass
_remove_file(temporary_path)
raise

if _load_config_file(path) != config:
_remove_file(path)
if _load_config_file(backup_path) is not None:
try:
os.rename(backup_path, path)
except OSError:
pass
raise ValueError("Could not verify active configuration")
_remove_file(backup_path)


def _load_config_file(path):
try:
with open(path) as config_file:
loaded_config = json.load(config_file)
except (OSError, ValueError, TypeError):
return None
if not _is_valid_config_shape(loaded_config):
return None
return loaded_config


def _is_valid_config_shape(config):
if not isinstance(config, dict):
return False

for section in _KNOWN_SECTIONS:
if section in config and not isinstance(config[section], dict):
return False

for section, fields in _STRING_FIELDS.items():
values = config.get(section, {})
for field in fields:
if field in values and not isinstance(values[field], str):
return False

for section, fields in _NUMBER_FIELDS.items():
values = config.get(section, {})
for field in fields:
value = values.get(field)
if field in values and (
not isinstance(value, (int, float)) or isinstance(value, bool)
):
return False

if "id" in config.get("printer", {}):
printer_id = config["printer"]["id"]
if printer_id is not None and (
not isinstance(printer_id, int)
or isinstance(printer_id, bool)
or printer_id < 0
):
return False

return True


def _remove_file(path):
try:
os.remove(path)
except OSError:
pass


def _copy_dict(source):
result = {}

Expand Down
12 changes: 4 additions & 8 deletions micro/web_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,11 @@

try:
import ota_manager
import config_loader
except ImportError:
from micro import ota_manager
from micro import ota_manager, config_loader

save_config = config_loader.save_config


CONFIG_PATH = "config.json"
Expand Down Expand Up @@ -292,13 +295,6 @@ def build_config(form, current_config):
return config


def save_config(path, config):
serialized = _json_dumps(config)
with open(path, "w") as config_file:
config_file.write(serialized)
config_file.write("\n")


def render_config_page(config, message="", update_status=None):
wifi = config.get("wifi", {})
api = config.get("api", {})
Expand Down
75 changes: 74 additions & 1 deletion tests/test_config_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,83 @@ def test_web_setup_default_hostname_has_four_character_suffix():
assert re.fullmatch(r"bambutton-[A-Z0-9]{4}", config_loader.DEFAULT_HOSTNAME)


def test_explicit_hostname_is_preserved_when_loading_config(tmp_path):
def test_partial_config_keeps_explicit_settings_and_fills_defaults(tmp_path):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"wifi": {"hostname": "shop-button"}}))

config = config_loader.load_config(config_path)

assert config["wifi"]["hostname"] == "shop-button"
assert config["wifi"]["ssid"] == ""
assert config["api"]["base_url"] == ""
assert config["printer"]["id"] is None
assert not config_loader.is_runtime_config_ready(config)


def test_malformed_config_recovers_from_valid_backup(tmp_path):
config_path = tmp_path / "config.json"
backup_path = tmp_path / "config.json.bak"
config_path.write_text('{"wifi":')
backup_path.write_text(json.dumps({
"wifi": {"ssid": "shop"},
"api": {"base_url": "http://bambuddy/api/v1"},
"printer": {"id": 7},
}))

config = config_loader.load_config(config_path)

assert config["wifi"]["ssid"] == "shop"
assert config["printer"]["id"] == 7


def test_malformed_config_without_backup_uses_bootable_defaults(tmp_path):
config_path = tmp_path / "config.json"
config_path.write_text('{"wifi":')

config = config_loader.load_config(config_path)

assert config["wifi"]["ssid"] == ""
assert config["api"]["base_url"] == ""
assert config["printer"]["id"] is None
assert config["web"]["password"]
assert not config_loader.is_runtime_config_ready(config)


def test_nested_invalid_active_config_recovers_from_valid_backup(tmp_path):
config_path = tmp_path / "config.json"
backup_path = tmp_path / "config.json.bak"
config_path.write_text(json.dumps({"wifi": None}))
backup_path.write_text(json.dumps({
"wifi": {"ssid": "shop"},
"api": {"base_url": "http://bambuddy/api/v1", "key": "secret"},
"printer": {"id": 7},
}))

config = config_loader.load_config(config_path)

assert config["wifi"]["ssid"] == "shop"
assert config["printer"]["id"] == 7
assert config_loader.is_runtime_config_ready(config)


def test_incomplete_config_never_inherits_a_printer_id(tmp_path):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({
"wifi": {"ssid": "shop"},
"api": {"base_url": "http://bambuddy/api/v1", "key": "secret"},
}))

config = config_loader.load_config(config_path)

assert config["printer"]["id"] is None
assert not config_loader.is_runtime_config_ready(config)


def test_non_numeric_printer_id_is_rejected_and_uses_safe_defaults(tmp_path):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"printer": {"id": "3"}}))

config = config_loader.load_config(config_path)

assert config["printer"]["id"] is None
assert not config_loader.is_runtime_config_ready(config)
Loading
Loading