Enforce same frequency on whole block

This commit is contained in:
Christoph Walcher
2026-07-05 14:35:29 +00:00
parent a8fd26ef6d
commit e7671dd854
3 changed files with 64 additions and 97 deletions
+61 -61
View File
@@ -1,3 +1,6 @@
from dataclasses import dataclass
from typing import Any
from esphome import automation, pins
import esphome.codegen as cg
from esphome.components import output
@@ -26,6 +29,7 @@ SetFrequencyAction = zephyr_pwm_ns.class_("SetFrequencyAction", automation.Actio
validate_frequency = cv.All(cv.frequency, cv.float_range(min=1.0, max=1e7))
CONF_ZEPHYR_PWM_BLOCKS = "zephyr_pwm_blocks"
CONFIG_SCHEMA = cv.All(
output.FLOAT_OUTPUT_SCHEMA.extend(
{
@@ -41,6 +45,13 @@ PWM_BLOCK_COUNT = 4
PWM_CHANNELS_PER_BLOCK = 4
@dataclass
class PWMBlock:
id: int
frequency: float
pins: list[Any]
def _final_validate(config: ConfigType) -> ConfigType:
full_config = fv.full_config.get()
zephyr_pwm_conf = [
@@ -48,69 +59,78 @@ def _final_validate(config: ConfigType) -> ConfigType:
for cfg in full_config.get(CONF_OUTPUT, [])
if cfg.get(CONF_PLATFORM) == "zephyr_pwm"
]
if zephyr_pwm_conf and len(zephyr_pwm_conf) > (
PWM_BLOCK_COUNT * PWM_CHANNELS_PER_BLOCK
):
raise cv.Invalid(
f"Only {PWM_BLOCK_COUNT * PWM_CHANNELS_PER_BLOCK} PWM outputs are supported by nrf52"
# Allocate pins to PWM blocks based on frequency
pwm_blocks: list[PWMBlock] = []
for cfg in zephyr_pwm_conf:
pin = cfg[CONF_PIN]
frequency = cfg[CONF_FREQUENCY]
pwm_block = next(
(
block
for block in pwm_blocks
if block.frequency == frequency
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), frequency=frequency, pins=[])
pwm_blocks.append(pwm_block)
pwm_block.pins.append(pin[CONF_NUMBER])
CORE.data[CONF_ZEPHYR_PWM_BLOCKS] = pwm_blocks
return config
FINAL_VALIDATE_SCHEMA = _final_validate
CONF_ZEPHYR_PWM_PINS = "zephyr_pwm_pins"
def _overlay_pwm():
pwm_pins = CORE.data[CONF_ZEPHYR_PWM_PINS]
pwm_blocks: list[PWMBlock] = CORE.data[CONF_ZEPHYR_PWM_BLOCKS]
assert CORE.is_nrf52
pwm_pins_by_block = [
pwm_pins[i : i + PWM_CHANNELS_PER_BLOCK]
for i in range(0, len(pwm_pins), PWM_CHANNELS_PER_BLOCK)
]
overlay_parts = []
overlay_parts.extend(
f"""
&pwm{block_id} {{
&pwm{block.id} {{
status = "okay";
pinctrl-0 = <&pwm{block_id}_default_custom>;
pinctrl-1 = <&pwm{block_id}_sleep_custom>;
pinctrl-0 = <&pwm{block.id}_default_custom>;
pinctrl-1 = <&pwm{block.id}_sleep_custom>;
pinctrl-names = "default", "sleep";
}};"""
for block_id in range(len(pwm_pins_by_block))
for block in pwm_blocks
)
psels_by_block = [
", ".join(
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_pwm_pins)
for channel_id, pin in enumerate(block.pins)
)
for block_pwm_pins in pwm_pins_by_block
]
pinctls = "\n".join(
f"""
pwm{block_id}_default_custom: pwm{block_id}_default_custom {{
pinctls.append(f"""
pwm{block.id}_default_custom: pwm{block.id}_default_custom {{
group1 {{
psels = {block_psels};
psels = {psels};
}};
}};
pwm{block_id}_sleep_custom: pwm{block_id}_sleep_custom {{
pwm{block.id}_sleep_custom: pwm{block.id}_sleep_custom {{
group1 {{
psels = {block_psels};
psels = {psels};
low-power-enable;
}};
}};"""
for block_id, block_psels in enumerate(psels_by_block)
)
}};""")
overlay_parts.append(f"""
&pinctrl {{
{pinctls}
{"\n".join(pinctls)}
}};""")
return "\n".join(overlay_parts)
@@ -119,45 +139,25 @@ async def to_code(config):
assert CORE.is_nrf52
zephyr_add_prj_conf("PWM", True)
pin = config[CONF_PIN]
CORE.data.setdefault(CONF_ZEPHYR_PWM_PINS, [])
pwm_pins = CORE.data[CONF_ZEPHYR_PWM_PINS]
pwm_id = len(pwm_pins)
assert pwm_id < (PWM_BLOCK_COUNT * PWM_CHANNELS_PER_BLOCK), (
f"Only {PWM_BLOCK_COUNT * PWM_CHANNELS_PER_BLOCK} PWM outputs are supported by nrf52"
pwm_blocks: list[PWMBlock] = CORE.data[CONF_ZEPHYR_PWM_BLOCKS]
pwm_block = next(
(block for block in pwm_blocks if pin[CONF_NUMBER] in block.pins), None
)
assert pwm_block is not None, (
f"Pin {pin[CONF_NUMBER]} is not assigned to any PWM block"
)
channel_id = pwm_block.pins.index(pin[CONF_NUMBER])
block_id = pwm_id // PWM_CHANNELS_PER_BLOCK
channel_id = pwm_id % PWM_CHANNELS_PER_BLOCK
pwm_pins.append(pin[CONF_NUMBER])
zephyr_add_overlay_builder(_overlay_pwm)
inverted = pin.get(CONF_INVERTED, False)
period_ns = int(1e9 / pwm_block.frequency)
var = cg.new_Pvariable(
config[CONF_ID],
cg.RawExpression(f"DEVICE_DT_GET_OR_NULL(DT_NODELABEL(pwm{block_id}))"),
cg.RawExpression(f"DEVICE_DT_GET_OR_NULL(DT_NODELABEL(pwm{pwm_block.id}))"),
channel_id,
inverted,
period_ns,
)
cg.add(var.set_frequency(config[CONF_FREQUENCY]))
await cg.register_component(var, config)
await output.register_output(var, config)
@automation.register_action(
"output.zephyr_pwm.set_frequency",
SetFrequencyAction,
cv.Schema(
{
cv.Required(CONF_ID): cv.use_id(ZephyrPWMChannel),
cv.Required(CONF_FREQUENCY): cv.templatable(validate_frequency),
}
),
synchronous=True,
)
async def zephyr_pwm_set_frequency_to_code(config, action_id, template_arg, args):
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_FREQUENCY], args, cg.float_)
cg.add(var.set_frequency(template_))
return var
@@ -30,8 +30,6 @@ void ZephyrPWMChannel::dump_config() {
LOG_FLOAT_OUTPUT(this);
}
void HOT ZephyrPWMChannel::write_state(float state) {
this->last_output_ = state;
uint32_t pulse_width_ns = state * this->period_ns_;
pwm_flags_t flags = this->inverted_ ? PWM_POLARITY_INVERTED : PWM_POLARITY_NORMAL;
int err = pwm_set(this->device_, this->channel_, this->period_ns_, pulse_width_ns, flags);
+3 -34
View File
@@ -13,23 +13,8 @@ namespace esphome::zephyr_pwm {
class ZephyrPWMChannel : public output::FloatOutput, public Component {
public:
explicit ZephyrPWMChannel(const struct device *device, uint8_t channel, bool inverted)
: device_(device), channel_(channel), inverted_(inverted) {}
bool set_frequency(float frequency) {
if (frequency < 1 || frequency > 1e7) {
return false;
}
this->period_ns_ = 1e9f / frequency;
return true;
}
/// Dynamically update frequency
void update_frequency(float frequency) override {
if (!this->set_frequency(frequency)) {
return;
}
this->write_state(this->last_output_);
}
explicit ZephyrPWMChannel(const struct device *device, uint8_t channel, bool inverted, uint32_t period_ns)
: device_(device), channel_(channel), inverted_(inverted), period_ns_(period_ns) {}
void setup() override;
void dump_config() override;
@@ -42,24 +27,8 @@ class ZephyrPWMChannel : public output::FloatOutput, public Component {
const struct device *device_;
uint8_t channel_;
bool inverted_;
uint32_t period_ns_{1000 * 1000}; // default to 1kHz
float last_output_{0.0};
uint32_t period_ns_;
};
template<typename... Ts> class SetFrequencyAction : public Action<Ts...> {
public:
SetFrequencyAction(ZephyrPWMChannel *parent) : parent_(parent) {}
TEMPLATABLE_VALUE(float, frequency);
void play(const Ts &...x) {
float freq = this->frequency_.value(x...);
this->parent_->update_frequency(freq);
}
protected:
ZephyrPWMChannel *parent_;
};
} // namespace esphome::zephyr_pwm
#endif // USE_ZEPHYR