FM / config

/snap/core24/1587/usr/lib/python3/dist-packages/cloudinit/config/

/snap/core24/1587/usr/lib/python3/dist-packages/cloudinit/config

Name	Size	Mode	Actions
schemas/	-	0755	rm
__pycache__/	-	0755	rm
cc_ansible.py	10727	0644	edit dl rm
cc_apk_configure.py	4331	0644	edit dl rm
cc_apt_configure.py	39170	0644	edit dl rm
cc_apt_pipelining.py	1909	0644	edit dl rm
cc_bootcmd.py	1690	0644	edit dl rm
cc_byobu.py	2800	0644	edit dl rm
cc_ca_certs.py	9126	0644	edit dl rm
cc_chef.py	13502	0644	edit dl rm
cc_disable_ec2_metadata.py	1642	0644	edit dl rm
cc_disk_setup.py	52622	0644	edit dl rm
cc_fan.py	1997	0644	edit dl rm
cc_final_message.py	2561	0644	edit dl rm
cc_growpart.py	18809	0644	edit dl rm
cc_grub_dpkg.py	5524	0644	edit dl rm
cc_install_hotplug.py	3260	0644	edit dl rm
cc_keyboard.py	1527	0644	edit dl rm
cc_keys_to_console.py	2138	0644	edit dl rm
cc_landscape.py	3131	0644	edit dl rm
cc_locale.py	1211	0644	edit dl rm
cc_lxd.py	13798	0644	edit dl rm
cc_mcollective.py	4207	0644	edit dl rm
cc_mounts.py	19322	0644	edit dl rm
cc_ntp.py	19528	0644	edit dl rm
cc_package_update_upgrade_install.py	3909	0644	edit dl rm
cc_phone_home.py	3829	0644	edit dl rm
cc_power_state_change.py	6133	0644	edit dl rm
cc_puppet.py	10976	0644	edit dl rm
cc_raspberry_pi.py	6599	0644	edit dl rm
cc_reset_rmc.py	4432	0644	edit dl rm
cc_resizefs.py	10880	0644	edit dl rm
cc_resolv_conf.py	3206	0644	edit dl rm
cc_rh_subscription.py	16545	0644	edit dl rm
cc_rsyslog.py	11745	0644	edit dl rm
cc_runcmd.py	1640	0644	edit dl rm
cc_salt_minion.py	4053	0644	edit dl rm
cc_scripts_per_boot.py	1298	0644	edit dl rm
cc_scripts_per_instance.py	1295	0644	edit dl rm
cc_scripts_per_once.py	1271	0644	edit dl rm
cc_scripts_user.py	1282	0644	edit dl rm
cc_scripts_vendor.py	1292	0644	edit dl rm
cc_seed_random.py	3068	0644	edit dl rm
cc_set_hostname.py	3481	0644	edit dl rm
cc_set_passwords.py	10566	0644	edit dl rm
cc_snap.py	3436	0644	edit dl rm
cc_spacewalk.py	2708	0644	edit dl rm
cc_ssh.py	11224	0644	edit dl rm
cc_ssh_authkey_fingerprints.py	3850	0644	edit dl rm
cc_ssh_import_id.py	5571	0644	edit dl rm
cc_timezone.py	1157	0644	edit dl rm
cc_ubuntu_autoinstall.py	1304	0644	edit dl rm
cc_ubuntu_drivers.py	4174	0644	edit dl rm
cc_ubuntu_pro.py	13971	0644	edit dl rm
cc_update_etc_hosts.py	2522	0644	edit dl rm
cc_update_hostname.py	2230	0644	edit dl rm
cc_users_groups.py	2930	0644	edit dl rm
cc_wireguard.py	6825	0644	edit dl rm
cc_write_files.py	6296	0644	edit dl rm
cc_write_files_deferred.py	1308	0644	edit dl rm
cc_yum_add_repo.py	4641	0644	edit dl rm
cc_zypper_add_repo.py	5052	0644	edit dl rm
modules.py	13643	0644	edit dl rm
schema.py	52049	0644	edit dl rm
__init__.py	40	0644	edit dl rm

Edit: /snap/core24/1587/usr/lib/python3/dist-packages/cloudinit/config/cc_raspberry_pi.py (6599B)

# 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
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_RPI_CONNECT_KEY = "enable_rpi_connect"
SUPPORTED_INTERFACES = {
    "spi": "do_spi",
    "i2c": "do_i2c",
    "serial": "do_serial",
    "onewire": "do_onewire",
}
RASPI_CONFIG_SERIAL_CONS_FN = "do_serial_cons"
RASPI_CONFIG_SERIAL_HW_FN = "do_serial_hw"

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

def configure_rpi_connect(enable: bool) -> None:
    LOG.debug("Configuring rpi-connect: %s", enable)

num = 0 if enable else 1

try:
        subp.subp(["/usr/bin/raspi-config", "do_rpi_connect", str(num)])
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure rpi-connect: %s", e)

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], instCfg: Config, cloud: Cloud
) -> None:
    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),
            ]
        )

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

# Reboot to apply changes
        cmd = cloud.distro.shutdown_command(
            mode="reboot",
            delay="now",
            message="Rebooting to apply serial console changes",
        )
        subp.subp(cmd)
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure serial console: %s", e)

def configure_interface(iface: str, enable: bool) -> None:
    assert (
        iface in SUPPORTED_INTERFACES.keys() and iface != "serial"
    ), f"Unsupported interface: {iface}"

try:
        subp.subp(
            [
                "/usr/bin/raspi-config",
                "nonint",
                SUPPORTED_INTERFACES[iface],
                str(0 if enable else 1),
            ]
        )
    except subp.ProcessExecutionError as e:
        LOG.error("Failed to configure %s: %s", iface, e)

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

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

if isinstance(enable, bool):
                configure_rpi_connect(enable)
            else:
                LOG.warning(
                    "Invalid value for %s: %s", ENABLE_RPI_CONNECT_KEY, enable
                )
            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())
            # Move " serial" to the end if it exists
            if "serial" in subkeys:
                subkeys.append(subkeys.pop(subkeys.index("serial")))

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

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

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

if isinstance(enable, bool):
                    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

cmd: