Add RTC Battery Charging

Previous impl. was setting a value for runtime config in hope
that offspot-runtime would support it and enable the charger.

While offspot-runtime may support it, given it is enabled in the firmware
it would require a restart which would be unwanted in our provisioning
scenario.

This changes it it so charger is enabled from next boot.

Other changes:
- commented OffspotYAMLStep as it's now doing nothing
- Moved boot order step last as this should only be done if all other
steps succeeded.

Fixes offspot/base-image#74

Author: rgaudin

src/provisioner/provisioning/charger.py

@@ -0,0 +1,48 @@
+import os
+
+from provisioner.constants import RTC_CHARGING_VOLTAGE
+from provisioner.context import Context
+from provisioner.provisioning.common import Step, StepResult
+from provisioner.utils.blk.misc import mount_to_temp, unmount
+from provisioner.utils.raspberry import BootConfig
+
+context = Context.get()
+logger = context.logger
+
+
+class EnableChargerStep(Step):
+
+    ident: str = "charger"
+    name: str = "Enable RTC Battery Charger"
+    reports_progress: bool = False
+    progress_interval_ms: int = 1000
+
+    def run(self, *, verbose: bool = False) -> StepResult:
+        root_dev = self.environment.target_disk.path  # /dev/nvme0n1
+        boot_part_dev = root_dev.with_name(f"{root_dev.name}p1")  # /dev/nvme0n1p1
+        mountpoint = mount_to_temp(boot_part_dev, rw=True)
+        try:
+            config_path = mountpoint.joinpath("config.txt")
+            config = BootConfig.parse(config_path.read_text())
+            config.add_key(
+                key="dtparam=rtc_bbat_vchg",
+                value=str(int(RTC_CHARGING_VOLTAGE * 1000000)),
+                cfilter="pi5",
+            )
+            config_path.write_text(config.serialize())
+            os.sync()
+            if verbose:
+                logger.info(f"{config_path=}:\n{config_path.read_text()}")
+        except Exception as exc:
+            return StepResult(
+                succeeded=False,
+                error_text="Failed to enable RTC Battery charging",
+                debug_text=str(exc),
+            )
+        finally:
+            unmount(mountpoint)
+            try:
+                mountpoint.rmdir()
+            except Exception:
+                ...
+        return StepResult(succeeded=True)

src/provisioner/provisioning/manager.py

@@ -3,11 +3,11 @@
 
 from provisioner.context import Context
 from provisioner.provisioning.bootorder import BootOrderStep
+from provisioner.provisioning.charger import EnableChargerStep
 from provisioner.provisioning.common import Environment, ImplementsProgress, Step
 from provisioner.provisioning.docker import DockerStep
 from provisioner.provisioning.hwclock import HardwareClockStep
 from provisioner.provisioning.imager import PiImagerStep
-from provisioner.provisioning.offspotyaml import OffspotYAMLStep
 from provisioner.provisioning.resizepart import ResizePartitionStep
 
 context = Context.get()
@@ -18,10 +18,11 @@ class ProvisionManager:
     STEPS: tuple[type[Step], ...] = (
         HardwareClockStep,
         PiImagerStep,
-        BootOrderStep,
+        EnableChargerStep,
         ResizePartitionStep,
-        OffspotYAMLStep,
+        # OffspotYAMLStep,  # no update ATM
         DockerStep,
+        BootOrderStep,
     )
 
     def __init__(

src/provisioner/provisioning/offspotyaml.py

@@ -1,6 +1,5 @@
 import os
 
-from provisioner.constants import RTC_CHARGING_VOLTAGE
 from provisioner.context import Context
 from provisioner.provisioning.common import Step, StepResult
 from provisioner.utils.blk.misc import mount_to_temp, unmount
@@ -25,15 +24,18 @@ def run(self, *, verbose: bool = False) -> StepResult:
             offspot_yaml = mountpoint.joinpath("offspot.yaml")
             config = yaml_load(offspot_yaml.read_text())
 
-            # update
-            # > set RTC charging voltage (to enable it)
-            config["rtc"] = {"charger": int(RTC_CHARGING_VOLTAGE * 10000)}
-            # country-update not required
+            # ADD runtime update HERE
 
             offspot_yaml.write_text(yaml_dump(config))
             os.sync()
             if verbose:
                 logger.info(f"{offspot_yaml=}:\n{offspot_yaml.read_text()}")
+        except Exception as exc:
+            return StepResult(
+                succeeded=False,
+                error_text="Failed to enable RTC Battery charging",
+                debug_text=str(exc),
+            )
         finally:
             unmount(mountpoint)
             try:

src/provisioner/utils/raspberry.py

@@ -3,6 +3,7 @@
 import tempfile
 from enum import StrEnum
 from pathlib import Path
+from typing import Self
 
 import attrs.validators
 from attrs import define, field
@@ -110,3 +111,65 @@ def update_eeprom(*, verbose: bool = False):
         check=True,
     )
     config_path.unlink(missing_ok=True)
+
+
+class BootConfig(dict):
+    """Manager for content of /boot/firmware/config.txt firmware config file"""
+
+    @classmethod
+    def parse(cls, text: str) -> Self:
+        board_type_re = re.compile(r"^\[board-type=(?P<name>\d+)\]$")
+        filter_re = re.compile(r"^\[(?P<name>[a-z0-9\+]+)\]$")
+        data = {}
+        currrent_filter = ""
+        for line in text.splitlines():
+            if m := board_type_re.match(line):
+                currrent_filter = m.groupdict()["name"]
+            elif m := filter_re.match(line):
+                currrent_filter = m.groupdict()["name"]
+            else:
+                if currrent_filter not in data:
+                    data[currrent_filter] = []
+                data[currrent_filter].append(line)
+        return cls(data)
+
+    def remove_key(self, key: str, *, only_in: list[str] | None = None):
+        """remove line for this key in every filter or selected ones"""
+        for fname, f_lines in self.items():
+            if only_in and key not in only_in:
+                continue
+            for index, line in enumerate(f_lines):
+                if line.startswith(f"{key}="):
+                    del self[fname][index]
+
+    def add_key(self, key: str, value: str, cfilter: str = "all"):
+        if cfilter not in self:
+            self[cfilter] = []
+        self[cfilter].append(f"{key}={value}")
+
+    def serialize(self) -> str:
+        # start with no filter ones (convewntion)
+        lines: list[str] = []
+        if "" in self:
+            lines += self[""]
+
+        for f_name, f_lines in self.items():
+            if f_name in ("", "all"):
+                continue
+
+            # no need to fill empty filters
+            if not f_lines:
+                continue
+            # add empty line before filter change is there was none
+            if lines and lines[-1]:
+                lines += [""]
+
+            lines += [f"[{f_name}]", *f_lines]
+
+        # end with the all ones (convention)
+        if self.get("all", []):
+            if lines and lines[-1]:
+                lines += [""]
+            lines += ["[all]", *self["all"]]
+
+        return "\n".join(lines)