Gecko [ tsuru-no-tsurugi.tokyo ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2026-06-23 20:32
schemas	[ DIR ]	drwxr-xr-x	2026-06-23 20:32
__init__.py	40 B	-rw-r--r--	2026-02-28 07:51
cc_ansible.py	10.48 KB	-rw-r--r--	2026-02-28 07:51
cc_apk_configure.py	4.23 KB	-rw-r--r--	2026-02-28 07:51
cc_apt_configure.py	38.25 KB	-rw-r--r--	2026-02-28 07:51
cc_apt_pipelining.py	1.86 KB	-rw-r--r--	2026-02-28 07:51
cc_bootcmd.py	1.65 KB	-rw-r--r--	2026-02-28 07:51
cc_byobu.py	2.73 KB	-rw-r--r--	2026-02-28 07:51
cc_ca_certs.py	8.92 KB	-rw-r--r--	2026-02-28 07:51
cc_chef.py	13.19 KB	-rw-r--r--	2026-02-28 07:51
cc_disable_ec2_metadata.py	1.6 KB	-rw-r--r--	2026-02-28 07:51
cc_disk_setup.py	51.43 KB	-rw-r--r--	2026-02-28 07:51
cc_fan.py	1.95 KB	-rw-r--r--	2026-02-28 07:51
cc_final_message.py	2.5 KB	-rw-r--r--	2026-02-28 07:51
cc_growpart.py	18.37 KB	-rw-r--r--	2026-02-28 07:51
cc_grub_dpkg.py	5.39 KB	-rw-r--r--	2026-03-31 03:56
cc_install_hotplug.py	3.18 KB	-rw-r--r--	2026-02-28 07:51
cc_keyboard.py	1.49 KB	-rw-r--r--	2026-02-28 07:51
cc_keys_to_console.py	2.09 KB	-rw-r--r--	2026-02-28 07:51
cc_landscape.py	3.06 KB	-rw-r--r--	2026-02-28 07:51
cc_locale.py	1.18 KB	-rw-r--r--	2026-02-28 07:51
cc_lxd.py	13.54 KB	-rw-r--r--	2026-03-31 03:56
cc_mcollective.py	4.11 KB	-rw-r--r--	2026-02-28 07:51
cc_mounts.py	18.87 KB	-rw-r--r--	2026-03-31 03:56
cc_ntp.py	19.02 KB	-rw-r--r--	2026-02-28 07:51
cc_package_update_upgrade_install.py	3.82 KB	-rw-r--r--	2026-02-28 07:51
cc_phone_home.py	3.74 KB	-rw-r--r--	2026-02-28 07:51
cc_power_state_change.py	5.99 KB	-rw-r--r--	2026-02-28 07:51
cc_puppet.py	10.72 KB	-rw-r--r--	2026-02-28 07:51
cc_raspberry_pi.py	8.39 KB	-rw-r--r--	2026-02-28 07:51
cc_reset_rmc.py	4.33 KB	-rw-r--r--	2026-02-28 07:51
cc_resizefs.py	10.63 KB	-rw-r--r--	2026-02-28 07:51
cc_resolv_conf.py	3.13 KB	-rw-r--r--	2026-02-28 07:51
cc_rh_subscription.py	16.14 KB	-rw-r--r--	2026-02-28 07:51
cc_rsyslog.py	11.47 KB	-rw-r--r--	2026-02-28 07:51
cc_runcmd.py	1.6 KB	-rw-r--r--	2026-02-28 07:51
cc_salt_minion.py	3.96 KB	-rw-r--r--	2026-02-28 07:51
cc_scripts_per_boot.py	1.27 KB	-rw-r--r--	2026-02-28 07:51
cc_scripts_per_instance.py	1.26 KB	-rw-r--r--	2026-02-28 07:51
cc_scripts_per_once.py	1.24 KB	-rw-r--r--	2026-02-28 07:51
cc_scripts_user.py	1.25 KB	-rw-r--r--	2026-02-28 07:51
cc_scripts_vendor.py	1.26 KB	-rw-r--r--	2026-02-28 07:51
cc_seed_random.py	3 KB	-rw-r--r--	2026-02-28 07:51
cc_set_hostname.py	3.4 KB	-rw-r--r--	2026-02-28 07:51
cc_set_passwords.py	10.32 KB	-rw-r--r--	2026-02-28 07:51
cc_snap.py	3.32 KB	-rw-r--r--	2026-03-31 03:56
cc_spacewalk.py	2.64 KB	-rw-r--r--	2026-02-28 07:51
cc_ssh.py	10.96 KB	-rw-r--r--	2026-02-28 07:51
cc_ssh_authkey_fingerprints.py	3.76 KB	-rw-r--r--	2026-02-28 07:51
cc_ssh_import_id.py	5.44 KB	-rw-r--r--	2026-02-28 07:51
cc_timezone.py	1.13 KB	-rw-r--r--	2026-02-28 07:51
cc_ubuntu_autoinstall.py	1.24 KB	-rw-r--r--	2026-03-31 03:56
cc_ubuntu_drivers.py	4.08 KB	-rw-r--r--	2026-02-28 07:51
cc_ubuntu_pro.py	13.64 KB	-rw-r--r--	2026-02-28 07:51
cc_update_etc_hosts.py	2.46 KB	-rw-r--r--	2026-02-28 07:51
cc_update_hostname.py	2.18 KB	-rw-r--r--	2026-02-28 07:51
cc_users_groups.py	2.86 KB	-rw-r--r--	2026-02-28 07:51
cc_wireguard.py	6.67 KB	-rw-r--r--	2026-02-28 07:51
cc_write_files.py	6.15 KB	-rw-r--r--	2026-02-28 07:51
cc_write_files_deferred.py	1.28 KB	-rw-r--r--	2026-02-28 07:51
cc_yum_add_repo.py	4.53 KB	-rw-r--r--	2026-02-28 07:51
cc_zypper_add_repo.py	4.93 KB	-rw-r--r--	2026-02-28 07:51
modules.py	13.32 KB	-rw-r--r--	2026-02-28 07:51
schema.py	52.26 KB	-rw-r--r--	2026-02-28 07:51

Rename

# Copyright (C) 2024-2025, Raspberry Pi Ltd.
#
# Author: Paul Oberosler <paul.oberosler@raspberrypi.com>
#
# This file is part of cloud-init. See LICENSE file for license information.

import logging
import os
from typing import Union

from cloudinit import subp
from cloudinit.cloud import Cloud
from cloudinit.config import Config
from cloudinit.config.schema import MetaSchema
from cloudinit.settings import PER_INSTANCE

LOG = logging.getLogger(__name__)
RPI_BASE_KEY = "rpi"
RPI_INTERFACES_KEY = "interfaces"
ENABLE_USB_GADGET_KEY = "enable_usb_gadget"
SUPPORTED_INTERFACES = {
    "spi": "do_spi",
    "i2c": "do_i2c",
    "onewire": "do_onewire",
}
SERIAL_INTERFACE = "serial"
RASPI_CONFIG_SERIAL_CONS_FN = "do_serial_cons"
RASPI_CONFIG_SERIAL_HW_FN = "do_serial_hw"
RPI_USB_GADGET_SCRIPT = "/usr/bin/rpi-usb-gadget"
REBOOT_MSG = "Rebooting to apply config.txt changes..."

meta: MetaSchema = {
    "id": "cc_raspberry_pi",
    "distros": ["raspberry-pi-os"],
    "frequency": PER_INSTANCE,
    "activate_by_schema_keys": [RPI_BASE_KEY],
}

def configure_usb_gadget(enable: bool) -> bool:
    """Enable / Disable Raspberry Pi USB gadget mode when
    the script is present.

Returns True when a reboot is required.
    """
    LOG.debug("Enable rpi-usb-gadget mode: %s", enable)

if not os.path.exists(RPI_USB_GADGET_SCRIPT):
        LOG.error("rpi-usb-gadget script not found: %s", RPI_USB_GADGET_SCRIPT)
        return False

try:
        subp.subp(
            [
                RPI_USB_GADGET_SCRIPT,
                "on" if enable else "off",
                # Force even if the device isn't supported
                # to avoid timeout in action request
                # and to allow activation with microSD card
                # in a unsupported device because its maybe faster
                # and then the user might switch it to a
                # supported device later
                "-f",
            ],
            capture=False,
            timeout=15,
        )

return True
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure rpi-usb-gadget: %s", e)
        return False

def is_pifive() -> bool:
    try:
        subp.subp(["/usr/bin/raspi-config", "nonint", "is_pifive"])
        return True
    except subp.ProcessExecutionError:
        return False

def configure_serial_interface(cfg: Union[dict, bool]) -> bool:
    """Configure the serial (UART) interface on Raspberry Pi.

Depending on the configuration, this enables or disables
    the serial console and the hardware UART by invoking
    `raspi-config` non-interactively.

Args:
        cfg: Either a boolean or a dictionary:
            - If a bool, enables or disables the serial
              console and hardware UART together.
            - If a dict, may contain the keys `console`
              and `hardware` with boolean values controlling
              each individually.

Returns:
        True if the configuration was applied successfully,
        False otherwise.
    """

def get_bool_field(cfg_dict: dict, name: str, default=False):
        val = cfg_dict.get(name, default)
        if not isinstance(val, bool):
            LOG.warning(
                "Invalid value for %s.serial.%s: %s",
                RPI_INTERFACES_KEY,
                name,
                val,
            )
            return default
        return val

enable_console = False
    enable_hw = False

if isinstance(cfg, dict):
        enable_console = get_bool_field(cfg, "console")
        enable_hw = get_bool_field(cfg, "hardware")

elif isinstance(cfg, bool):
        # default to enabling console as if < pi5
        # this will also enable the hardware
        enable_console = cfg

if not is_pifive() and enable_console:
        # only pi5 has 2 usable UARTs
        # on other models, enabling the console
        # will also block the other UART
        enable_hw = True

try:
        subp.subp(
            [
                "/usr/bin/raspi-config",
                "nonint",
                RASPI_CONFIG_SERIAL_CONS_FN,
                str(0 if enable_console else 1),
            ]
        )
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure serial console: %s", e)
        return False

try:
        subp.subp(
            [
                "/usr/bin/raspi-config",
                "nonint",
                RASPI_CONFIG_SERIAL_HW_FN,
                str(0 if enable_hw else 1),
            ]
        )
        return True
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure serial hardware: %s", e)
        return False

def configure_interface(iface: str, enable: bool) -> bool:
    """Enable or disable a supported hardware interface.

Supported interfaces are defined in `SUPPORTED_INTERFACES` and
    currently include SPI, I2C, and 1-Wire. The function applies
    the change using `raspi-config` in non-interactive mode.

Args:
        iface: The interface name (e.g. `"spi"` or `"i2c"`).
        enable: True to enable the interface, False to disable it.

Returns:
        True if the interface configuration succeeded, False otherwise.
    """
    assert (
        iface in SUPPORTED_INTERFACES.keys()
    ), f"Unsupported interface: {iface}"

try:
        subp.subp(
            [
                "/usr/bin/raspi-config",
                "nonint",
                SUPPORTED_INTERFACES[iface],
                str(0 if enable else 1),
            ]
        )

return True
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure %s: %s", iface, e)
        return False

def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None:
    if RPI_BASE_KEY not in cfg:
        return
    elif not isinstance(cfg[RPI_BASE_KEY], dict):
        LOG.warning(
            "Invalid value for %s: %s",
            RPI_BASE_KEY,
            cfg[RPI_BASE_KEY],
        )
        return
    elif not cfg[RPI_BASE_KEY]:
        LOG.debug("Empty value for %s. Skipping...", RPI_BASE_KEY)
        return

want_reboot = False

for key in cfg[RPI_BASE_KEY]:
        if key == ENABLE_USB_GADGET_KEY:
            enable = cfg[RPI_BASE_KEY][key]

if isinstance(enable, bool):
                want_reboot |= configure_usb_gadget(enable)
            else:
                raise ValueError(f"Invalid value for {ENABLE_USB_GADGET_KEY}")
            continue
        elif key == RPI_INTERFACES_KEY:
            if not isinstance(cfg[RPI_BASE_KEY][key], dict):
                LOG.warning(
                    "Invalid value for %s: %s",
                    RPI_BASE_KEY,
                    cfg[RPI_BASE_KEY][key],
                )
                return
            elif not cfg[RPI_BASE_KEY][key]:
                LOG.debug("Empty value for %s. Skipping...", key)
                return

subkeys = list(cfg[RPI_BASE_KEY][key].keys())

# check for supported ARM interfaces
            for subkey in subkeys:
                if (
                    subkey not in SUPPORTED_INTERFACES.keys()
                    and subkey != SERIAL_INTERFACE
                ):
                    LOG.warning(
                        "Invalid key for %s: %s", RPI_INTERFACES_KEY, subkey
                    )
                    continue

enable = cfg[RPI_BASE_KEY][key][subkey]

if subkey == SERIAL_INTERFACE:
                    if not isinstance(enable, (dict, bool)):
                        LOG.warning(
                            "Invalid value for %s.%s: %s",
                            RPI_INTERFACES_KEY,
                            subkey,
                            enable,
                        )
                    else:
                        want_reboot |= configure_serial_interface(enable)
                    continue

if isinstance(enable, bool):
                    want_reboot |= configure_interface(subkey, enable)
                else:
                    LOG.warning(
                        "Invalid value for %s.%s: %s",
                        RPI_INTERFACES_KEY,
                        subkey,
                        enable,
                    )
        else:
            LOG.warning("Unsupported key: %s", key)
            continue

if want_reboot is True:
        # Reboot to apply changes to config.txt and modprobe
        cmd = cloud.distro.shutdown_command(
            mode="reboot",
            delay="now",
            message=REBOOT_MSG,
        )
        subp.subp(cmd)