[nrf52] feat: add support for zephyr pwm (#16483)

Co-authored-by: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com>
Co-authored-by: tomaszduda23 <tomaszduda23@gmail.com>

CODEOWNERS

@@ -628,6 +628,7 @@ esphome/components/xpt2046/touchscreen/* @nielsnl68 @numo68
 esphome/components/xxtea/* @clydebarrow
 esphome/components/zephyr/* @tomaszduda23
 esphome/components/zephyr_mcumgr/ota/* @tomaszduda23
+esphome/components/zephyr_pwm/* @wiomoc
 esphome/components/zhlt01/* @cfeenstra1024
 esphome/components/zigbee/* @luar123 @tomaszduda23
 esphome/components/zio_ultrasonic/* @kahrendt

esphome/components/adc/sensor.py

@@ -10,8 +10,8 @@ from esphome.components.esp32 import (
 from esphome.components.nrf52.const import AIN_TO_GPIO, EXTRA_ADC
 from esphome.components.zephyr import (
     zephyr_add_overlay,
+    zephyr_add_overlay_builder,
     zephyr_add_prj_conf,
-    zephyr_add_user,
 )
 from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
@@ -113,6 +113,18 @@ CONFIG_SCHEMA = cv.All(
 CONF_ADC_CHANNEL_ID = "adc_channel_id"
 
 
+def _overlay_io_channels():
+    channel_count = CORE.data[CONF_ADC_CHANNEL_ID]
+    entries = ", ".join(f"<&adc {channel_id}>" for channel_id in range(channel_count))
+    return f"""
+            / {{
+                zephyr,user {{
+                    io-channels = {entries};
+                }};
+            }};
+            """
+
+
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
@@ -173,9 +185,8 @@ async def to_code(config):
         if isinstance(pin_number, int):
             GPIO_TO_AIN = {v: k for k, v in AIN_TO_GPIO.items()}
             pin_number = GPIO_TO_AIN[pin_number]
-        zephyr_add_user("io-channels", f"<&adc {channel_id}>")
+        zephyr_add_overlay_builder(_overlay_io_channels)
-        zephyr_add_overlay(
+        zephyr_add_overlay(f"""
-            f"""
                 &adc {{
                     #address-cells = <1>;
                     #size-cells = <0>;
@@ -190,8 +201,7 @@ async def to_code(config):
                         zephyr,oversampling = <8>;
                     }};
                 }};
-            """
+            """)
-        )
 
 
 FILTER_SOURCE_FILES = filter_source_files_from_platform(

esphome/components/zephyr/__init__.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from pathlib import Path
 import textwrap
 from typing import TypedDict
@@ -16,10 +17,10 @@ from .const import (
     KEY_EXTRA_BUILD_FILES,
     KEY_KCONFIG,
     KEY_OVERLAY,
+    KEY_OVERLAY_BUILDER,
     KEY_PM_STATIC,
     KEY_PRJ_CONF,
     KEY_SYSBUILD,
-    KEY_USER,
     KEY_ZEPHYR,
     zephyr_ns,
 )
@@ -73,9 +74,9 @@ class ZephyrData(TypedDict):
     overlay: dict[str, str]
     extra_build_files: dict[str, Path]
     pm_static: list[Section]
-    user: dict[str, list[str]]
     kconfig: str
     sysbuild: bool
+    overlay_builder: list[Callable[[], str]]
 
 
 def zephyr_set_core_data(config: ConfigType) -> None:
@@ -86,9 +87,9 @@ def zephyr_set_core_data(config: ConfigType) -> None:
         overlay={
             "": "",
         },  # set empty to make sure that overlay is cleared after config change
+        overlay_builder=[],
         extra_build_files={},
         pm_static=[],
-        user={},
         kconfig="",
         # When OTA is disabled, the image is built without a bootloader even if the
         # config says `bootloader: mcuboot`, so the image can be smaller. This was
@@ -132,6 +133,12 @@ def zephyr_add_overlay(content: str, image: str = "") -> None:
     data[KEY_OVERLAY][image] += textwrap.dedent(content)
 
 
+def zephyr_add_overlay_builder(func: Callable[[], str]) -> None:
+    data = zephyr_data()
+    if func not in data[KEY_OVERLAY_BUILDER]:
+        data[KEY_OVERLAY_BUILDER].append(func)
+
+
 def add_extra_build_file(filename: str, path: Path) -> bool:
     """Add an extra build file to the project."""
     extra_build_files = zephyr_data()[KEY_EXTRA_BUILD_FILES]
@@ -222,13 +229,6 @@ def zephyr_add_pm_static(sections: list[Section]) -> None:
     zephyr_data()[KEY_PM_STATIC].extend(sections)
 
 
-def zephyr_add_user(key, value):
-    user = zephyr_data()[KEY_USER]
-    if key not in user:
-        user[key] = []
-    user[key] += [value]
-
-
 def _write_file_if_changed_or_remove_when_empty(path: Path, content: str) -> bool:
     """Write content to path, or remove a stale file when content is empty.
 
@@ -243,20 +243,9 @@ def _write_file_if_changed_or_remove_when_empty(path: Path, content: str) -> boo
 
 
 def copy_files() -> None:
-    user = zephyr_data()[KEY_USER]
+    for builder_func in zephyr_data()[KEY_OVERLAY_BUILDER]:
-    if user:
+        overlay_contents = builder_func()
-        entries = " ".join(
+        zephyr_add_overlay(overlay_contents)
-            f"{key} = {', '.join(value)};" for key, value in user.items()
-        )
-        zephyr_add_overlay(
-            f"""
-                / {{
-                    zephyr,user {{
-                        {entries}
-                    }};
-                }};
-            """
-        )
 
     changed = False
 

esphome/components/zephyr/const.py

@@ -7,12 +7,12 @@ BOOTLOADER_MCUBOOT = "mcuboot"
 KEY_BOOTLOADER: Final = "bootloader"
 KEY_EXTRA_BUILD_FILES: Final = "extra_build_files"
 KEY_OVERLAY: Final = "overlay"
+KEY_OVERLAY_BUILDER: Final = "overlay_builder"
 KEY_PM_STATIC: Final = "pm_static"
 KEY_KCONFIG: Final = "kconfig"
 KEY_PRJ_CONF: Final = "prj_conf"
 KEY_ZEPHYR = "zephyr"
 KEY_BOARD: Final = "board"
-KEY_USER: Final = "user"
 KEY_SYSBUILD: Final = "sysbuild"
 
 zephyr_ns = cg.esphome_ns.namespace("zephyr")

esphome/components/zephyr_pwm/__init__.py

@@ -0,0 +1 @@
+CODEOWNERS = ["@wiomoc"]

esphome/components/zephyr_pwm/output.py

@@ -0,0 +1,177 @@
+from dataclasses import dataclass, field
+
+from esphome import pins
+import esphome.codegen as cg
+from esphome.components import output
+from esphome.components.zephyr import zephyr_add_overlay_builder, zephyr_add_prj_conf
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_ALLOW_OTHER_USES,
+    CONF_FREQUENCY,
+    CONF_ID,
+    CONF_INVERTED,
+    CONF_NUMBER,
+    CONF_OUTPUT,
+    CONF_PIN,
+    CONF_PLATFORM,
+)
+from esphome.core import CORE
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+DEPENDENCIES = ["zephyr"]
+DOMAIN = "zephyr_pwm"
+
+zephyr_pwm_ns = cg.esphome_ns.namespace("zephyr_pwm")
+ZephyrPWMChannel = zephyr_pwm_ns.class_(
+    "ZephyrPWMChannel", output.FloatOutput, cg.Component
+)
+validate_frequency = cv.All(cv.frequency, cv.float_range(min=3.815, max=1e7))
+
+
+def _pin_schema(value):
+    value = pins.internal_gpio_output_pin_schema(value)
+    if value.get(CONF_ALLOW_OTHER_USES, False):
+        raise cv.Invalid("allow_other_uses is not supported for zephyr_pwm pins")
+    return value
+
+
+CONFIG_SCHEMA = cv.All(
+    output.FLOAT_OUTPUT_SCHEMA.extend(
+        {
+            cv.Required(CONF_ID): cv.declare_id(ZephyrPWMChannel),
+            cv.Required(CONF_PIN): _pin_schema,
+            cv.Optional(CONF_FREQUENCY, default="1kHz"): validate_frequency,
+        }
+    ).extend(cv.COMPONENT_SCHEMA),
+    cv.only_on_nrf52,
+)
+
+PWM_BLOCK_COUNT = 4
+PWM_CHANNELS_PER_BLOCK = 4
+
+
+@dataclass
+class PWMBlock:
+    id: int
+    period_ns: int
+    pins: list[int]
+
+
+@dataclass
+class ZephyrPWMData:
+    pwm_blocks: list[PWMBlock] = field(default_factory=list)
+
+
+def _get_data() -> ZephyrPWMData:
+    if DOMAIN not in CORE.data:
+        CORE.data[DOMAIN] = ZephyrPWMData()
+    return CORE.data[DOMAIN]
+
+
+def _allocate_blocks() -> None:
+    full_config = fv.full_config.get()
+    zephyr_pwm_conf = [
+        cfg
+        for cfg in full_config.get(CONF_OUTPUT, [])
+        if cfg.get(CONF_PLATFORM) == DOMAIN
+    ]
+
+    pwm_blocks: list[PWMBlock] = []
+    for cfg in zephyr_pwm_conf:
+        pin_number = cfg[CONF_PIN][CONF_NUMBER]
+        period_ns = int(1e9 / cfg[CONF_FREQUENCY])
+        pwm_block = next(
+            (
+                block
+                for block in pwm_blocks
+                if block.period_ns == period_ns
+                and len(block.pins) < PWM_CHANNELS_PER_BLOCK
+            ),
+            None,
+        )
+        if pwm_block is None:
+            if len(pwm_blocks) >= PWM_BLOCK_COUNT:
+                raise cv.Invalid(
+                    f"Only {PWM_BLOCK_COUNT} PWM blocks with a distinct frequency and {PWM_CHANNELS_PER_BLOCK} channels each are supported by nrf52"
+                )
+            pwm_block = PWMBlock(id=len(pwm_blocks), period_ns=period_ns, pins=[])
+            pwm_blocks.append(pwm_block)
+        pwm_block.pins.append(pin_number)
+
+    _get_data().pwm_blocks = pwm_blocks
+
+
+def _final_validate(config: ConfigType) -> ConfigType:
+    _allocate_blocks()
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
+def _overlay_pwm():
+    pwm_blocks: list[PWMBlock] = _get_data().pwm_blocks
+
+    assert CORE.is_nrf52
+
+    overlay_parts = []
+
+    overlay_parts.extend(
+        f"""
+        &pwm{block.id} {{
+            status = "okay";
+            pinctrl-0 = <&pwm{block.id}_default_custom>;
+            pinctrl-1 = <&pwm{block.id}_sleep_custom>;
+            pinctrl-names = "default", "sleep";
+        }};"""
+        for block in pwm_blocks
+    )
+
+    pinctls = []
+    for block in pwm_blocks:
+        psels = ", ".join(
+            f"<NRF_PSEL(PWM_OUT{channel_id}, {pin // 32}, {pin % 32})>"
+            for channel_id, pin in enumerate(block.pins)
+        )
+        pinctls.append(f"""
+            pwm{block.id}_default_custom: pwm{block.id}_default_custom {{
+                group1 {{
+                    psels = {psels};
+                }};
+            }};
+            pwm{block.id}_sleep_custom: pwm{block.id}_sleep_custom {{
+                group1 {{
+                    psels = {psels};
+                    low-power-enable;
+                }};
+            }};""")
+
+    overlay_parts.append(f"""
+        &pinctrl {{
+         {"\n".join(pinctls)}
+        }};""")
+    return "\n".join(overlay_parts)
+
+
+async def to_code(config):
+    zephyr_add_prj_conf("PWM", True)
+    pin = config[CONF_PIN]
+    pwm_blocks: list[PWMBlock] = _get_data().pwm_blocks
+    pwm_block = next(
+        (block for block in pwm_blocks if pin[CONF_NUMBER] in block.pins), None
+    )
+    channel_id = pwm_block.pins.index(pin[CONF_NUMBER])
+
+    zephyr_add_overlay_builder(_overlay_pwm)
+
+    pin_inverted = pin.get(CONF_INVERTED, False)
+    var = cg.new_Pvariable(
+        config[CONF_ID],
+        cg.RawExpression(f"DEVICE_DT_GET_OR_NULL(DT_NODELABEL(pwm{pwm_block.id}))"),
+        channel_id,
+        pin_inverted,
+        pwm_block.period_ns,
+    )
+    await cg.register_component(var, config)
+    await output.register_output(var, config)

esphome/components/zephyr_pwm/zephyr_pwm.cpp

@@ -0,0 +1,39 @@
+#ifdef USE_ZEPHYR
+
+#include "zephyr_pwm.h"
+
+#include <zephyr/drivers/pwm.h>
+
+namespace esphome::zephyr_pwm {
+
+static const char *const TAG = "zephyr_pwm";
+
+void ZephyrPWMChannel::setup() {
+  if (!device_is_ready(this->device_)) {
+    ESP_LOGE(TAG, "PWM is not ready.");
+    this->mark_failed();
+    return;
+  }
+}
+
+void ZephyrPWMChannel::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Zephyr PWM:\n"
+                "  Channel: %u\n"
+                "  Period: %u ns",
+                this->channel_, this->period_ns_);
+  LOG_FLOAT_OUTPUT(this);
+}
+void HOT ZephyrPWMChannel::write_state(float state) {
+  uint32_t pulse_width_ns = state * this->period_ns_;
+  pwm_flags_t flags = this->pin_inverted_ ? PWM_POLARITY_INVERTED : PWM_POLARITY_NORMAL;
+  int err = pwm_set(this->device_, this->channel_, this->period_ns_, pulse_width_ns, flags);
+  if (err != 0) {
+    ESP_LOGE(TAG, "Failed to set PWM output: channel=%u, period=%u ns, pulse_width=%u ns, error=%d", this->channel_,
+             this->period_ns_, pulse_width_ns, err);
+  }
+}
+
+}  // namespace esphome::zephyr_pwm
+
+#endif  // USE_ZEPHYR

esphome/components/zephyr_pwm/zephyr_pwm.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#ifdef USE_ZEPHYR
+#include "esphome/core/defines.h"
+#include "esphome/components/output/float_output.h"
+
+#include <zephyr/device.h>
+
+namespace esphome::zephyr_pwm {
+
+class ZephyrPWMChannel : public output::FloatOutput, public Component {
+ public:
+  explicit ZephyrPWMChannel(const struct device *device, uint8_t channel, bool pin_inverted, uint32_t period_ns)
+      : device_(device), channel_(channel), pin_inverted_(pin_inverted), period_ns_(period_ns) {}
+
+  void setup() override;
+  void dump_config() override;
+  /// HARDWARE setup_priority
+  float get_setup_priority() const override { return setup_priority::HARDWARE; }
+
+ protected:
+  void write_state(float state) override;
+
+  const struct device *device_;
+  uint8_t channel_;
+  bool pin_inverted_;
+  uint32_t period_ns_;
+};
+}  // namespace esphome::zephyr_pwm
+
+#endif  // USE_ZEPHYR

tests/components/light/test.nrf52-adafruit.yaml

@@ -1,19 +1,23 @@
-esphome:
-  on_boot:
-    then:
-      - light.toggle: test_binary_light
-
 output:
   - platform: gpio
     id: light_test_binary
-    pin: 0
+    pin: 12
+  - platform: zephyr_pwm
+    id: test_ledc_1
+    pin: 13
+  - platform: zephyr_pwm
+    id: test_ledc_2
+    pin:
+      number: 14
+      inverted: true
+  - platform: zephyr_pwm
+    id: test_ledc_3
+    pin: 15
+  - platform: zephyr_pwm
+    id: test_ledc_4
+    pin: 16
+  - platform: zephyr_pwm
+    id: test_ledc_5
+    pin: 17
 
-light:
+<<: !include common.yaml
-  - platform: binary
-    id: test_binary_light
-    name: Binary Light
-    output: light_test_binary
-    effects:
-      - strobe:
-    on_state:
-      - logger.log: Binary light state changed

tests/components/light/test.nrf52-mcumgr.yaml

@@ -1,19 +1,23 @@
-esphome:
-  on_boot:
-    then:
-      - light.toggle: test_binary_light
-
 output:
   - platform: gpio
     id: light_test_binary
-    pin: 0
+    pin: 12
+  - platform: zephyr_pwm
+    id: test_ledc_1
+    pin: 13
+  - platform: zephyr_pwm
+    id: test_ledc_2
+    pin:
+      number: 14
+      inverted: true
+  - platform: zephyr_pwm
+    id: test_ledc_3
+    pin: 15
+  - platform: zephyr_pwm
+    id: test_ledc_4
+    pin: 16
+  - platform: zephyr_pwm
+    id: test_ledc_5
+    pin: 17
 
-light:
+<<: !include common.yaml
-  - platform: binary
-    id: test_binary_light
-    name: Binary Light
-    output: light_test_binary
-    effects:
-      - strobe:
-    on_state:
-      - logger.log: Binary light state changed

tests/components/zephyr_pwm/common.yaml

@@ -0,0 +1,9 @@
+output:
+  - platform: zephyr_pwm
+    id: pwm_output_1
+    pin: P0.02
+  - platform: zephyr_pwm
+    id: pwm_output_2
+    pin:
+      number: 10
+      inverted: true

tests/components/zephyr_pwm/test.nrf52-adafruit.yaml

@@ -0,0 +1 @@
+<<: !include common.yaml

tests/unit_tests/test_nrf52_upload.py

@@ -13,9 +13,9 @@ from esphome.components.zephyr.const import (
     KEY_EXTRA_BUILD_FILES,
     KEY_KCONFIG,
     KEY_OVERLAY,
+    KEY_OVERLAY_BUILDER,
     KEY_PM_STATIC,
     KEY_PRJ_CONF,
-    KEY_USER,
     KEY_ZEPHYR,
 )
 import esphome.config_validation as cv
@@ -53,9 +53,9 @@ def _setup_nrf52_core(
         KEY_BOOTLOADER: bootloader,
         KEY_PRJ_CONF: {},
         KEY_OVERLAY: {"": ""},
+        KEY_OVERLAY_BUILDER: [],
         KEY_EXTRA_BUILD_FILES: {},
         KEY_PM_STATIC: [],
-        KEY_USER: {},
         KEY_KCONFIG: "",
     }