Compare commits

...
Author SHA1 Message Date
J. Nick Koston d7bda29ada Merge branch 'dev' into jesserockz-2026-446 2026-10-12 15:07:27 -10:00
J. Nick Koston 86073304dc Merge branch 'dev' into jesserockz-2026-446 2026-09-24 14:21:53 +01:00
Jesse Hills 63e1adfd5f [schema] Keep the field's own range when spreading a named schema
Address esphbot review: emitting min/max means named scalar schemas now carry
bounds (e.g. core.uint8_t is 0..255), and shrink()'s spread copied those over a
tighter bound the field defined itself. cv.All(cv.uint8_t, cv.Range(min=1,
max=6)) dumped 0..255 instead of 1..6, and All(uint32_t, Range(min=1)) lost its
min. The spread now reconciles the two to the intersection (larger min, smaller
max); non-numeric bounds keep the field's own value.
2026-09-15 07:13:29 +12:00
Jesse Hills 59044484e2 [schema] Unit-type number-with-unit validators; leave dimensions untyped
Address esphbot review: validators that accept a number optionally carrying a
unit (percentage, possibly_negative_percentage, temperature, temperature_delta,
color_temperature, percentage_int, validate_bytes) are dumped as their base
numeric type plus the canonical unit (e.g. {type: float, unit: "%"}), like
float_with_unit, so the editor accepts both the bare-number and the unit-string
form instead of rejecting one. cv.dimensions accepts a "WIDTHxHEIGHT" string or
a [w, h] list, so it is left untyped rather than misrepresented as a string.
2026-09-14 13:52:22 +12:00
Jesse Hills b40c497d3e Merge branch 'dev' into jesserockz-2026-446 2026-09-14 13:41:47 +12:00
Jesse Hills d4fb77af75 [schema] Emit unit for float fields; leave union validators untyped
Address review feedback:
- float_with_unit fields now dump as type "float" with the canonical unit
  (e.g. {"type": "float", "unit": "Hz"}) rather than the quantity name as the
  type. This keeps the type vocabulary closed and lets editors label the field
  with its unit directly instead of mapping each quantity name. The unit is the
  first alternative of the accepted-units regex; current/voltage list A/V first
  (no validation change).
- Drop scalar typing of validate_bytes and date_time: both accept multiple YAML
  forms (unit-suffixed strings or numbers, and a formatted string or a mapping)
  that a single scalar type would wrongly reject in the editor.
2026-07-16 07:51:26 +12:00
Jesse Hills c0b59ea10b Merge remote-tracking branch 'origin/dev' into jesserockz-2026-446 2026-07-15 14:44:56 +12:00
Jesse Hills 98f6e74385 [schema] Type config_validation validators and range bounds in language schema dump
The language schema dump left many config_validation validators untyped
(icon, mac_address, percentage, update_interval, lambdas, encryption keys,
...), so the visual editor and dashboard could not tell what YAML those
fields accept. Type them via convert() and schema_extractor decorators,
emit float_with_unit quantities (frequency, voltage, current, ...) as the
field type, and attach min/max bounds detected from range validators next
to the type.
2026-07-15 14:44:28 +12:00
4 changed files with 411 additions and 9 deletions

No files matched your search

+9 -2
View File
@@ -1404,8 +1404,15 @@ def float_with_unit(quantity, regex_suffix, optional_unit=False):
pattern = re.compile(
f"^([-+]?[0-9]*\\.?[0-9]*)\\s*(\\w*?){regex_suffix}$", re.UNICODE
)
# First alternative of the accepted-units regex is the canonical unit
# (e.g. "(Hz|HZ|hz)?" -> "Hz"); reported to the schema dump so editors can
# label the numeric field with its unit. None for an empty suffix.
unit = regex_suffix.lstrip("(").split(")", 1)[0].split("|", 1)[0] or None
@schema_extractor("float")
def validator(value):
if value is SCHEMA_EXTRACT:
return unit
if optional_unit:
try:
return float_(value)
@@ -1429,8 +1436,8 @@ def float_with_unit(quantity, regex_suffix, optional_unit=False):
bps = float_with_unit("bits per second", "(bps|bits/s|bit/s)?")
frequency = float_with_unit("frequency", "(Hz|HZ|hz)?")
resistance = float_with_unit("resistance", "(Ω|Ω|ohm|Ohm|OHM)?")
current = float_with_unit("current", "(a|A|amp|Amp|amps|Amps|ampere|Ampere)?")
voltage = float_with_unit("voltage", "(v|V|volt|Volts)?")
current = float_with_unit("current", "(A|a|amp|Amp|amps|Amps|ampere|Ampere)?")
voltage = float_with_unit("voltage", "(V|v|volt|Volts)?")
distance = float_with_unit("distance", "(m)")
framerate = float_with_unit("framerate", "(FPS|fps|Fps|FpS|Hz)")
angle = float_with_unit("angle", "(°|deg)", optional_unit=True)
+169 -7
View File
@@ -120,6 +120,61 @@ from esphome.util import Registry # noqa: E402
# pylint: enable=wrong-import-position
# Scalar ``cv.*`` validators the dumper describes by identity. Extending these
# tuples -- rather than decorating each validator with a schema_extractor --
# mirrors how ``cv.boolean`` / ``cv.string`` / ``cv.int_`` are already handled
# in ``convert()`` and keeps runtime validation completely untouched. Grouped by
# the ``type`` emitted into the schema dump so a language server / visual editor
# knows what YAML each field accepts instead of treating it as free-form.
_CV_STRING_VALIDATORS = (
cv.icon,
cv.mac_address,
cv.url,
cv.publish_topic,
cv.subscribe_topic,
cv.mqtt_payload,
cv.uuid,
cv.ssid,
cv.domain,
cv.domain_name,
cv.hostname,
cv.entity_id,
cv.git_ref,
cv.string_no_slash,
cv.version_number,
cv.validate_esphome_version,
cv.validate_id_name,
cv._validate_entity_name,
cv.validate_source_shorthand,
cv.ipv4address,
cv.ipv6network,
cv.ipv4address_multi_broadcast,
cv.time_of_day,
cv.directory,
cv.file_,
cv.none,
)
_CV_INTEGER_VALIDATORS = (cv.hex_int, cv.mqtt_qos)
# Validators that accept a number optionally carrying a unit suffix (e.g.
# "50%", "25°C", "10kB"). Dumped as their base numeric type with the canonical
# unit attached, mirroring float_with_unit, so editors accept both the
# bare-number and the unit-string form instead of rejecting one of them.
_CV_UNIT_VALIDATORS = {
cv.percentage: ("float", "%"),
cv.possibly_negative_percentage: ("float", "%"),
cv.temperature: ("float", "°C"),
cv.temperature_delta: ("float", "°C"),
cv.color_temperature: ("float", "mireds"),
cv.percentage_int: ("integer", "%"),
cv.validate_bytes: ("integer", "B"),
}
_CV_TIME_VALIDATORS = (
cv.update_interval,
cv.time_period_str_unit,
cv.time_period_str_colon,
)
_CV_LAMBDA_VALIDATORS = (cv.lambda_, cv.returning_lambda)
def sort_obj(obj):
if isinstance(obj, dict):
@@ -589,6 +644,37 @@ def is_platform_schema(schema_name):
return component in components and components[component].is_platform_component
def _tighter_bound(own, ref, choose):
"""Return the tighter of a field's own range bound and a referenced schema's.
``choose`` is ``max`` for a lower bound and ``min`` for an upper bound, so an
``All(named_type, field_range)`` collapses to the intersection of the two
ranges. Non-numeric bounds (e.g. a stringified TimePeriod) can't be compared,
so the field's own value is kept.
"""
if own is None:
return ref
if ref is None:
return own
if isinstance(own, (int, float)) and isinstance(ref, (int, float)):
return choose(own, ref)
return own
def _apply_tighter_bounds(target_s, own_min, own_max, key_s):
"""Reconcile a field's own min/max (captured before the spread) with the
referenced schema's bounds, so spreading a named schema never widens
(clobbers) a tighter bound the field defined itself, e.g.
``All(uint8_t, Range(min=1, max=6))`` must stay 1..6, not uint8_t's 0..255.
"""
for bound, own, choose in (("min", own_min, max), ("max", own_max, min)):
merged = _tighter_bound(own, key_s.get(bound), choose)
if merged is None:
target_s.pop(bound, None)
else:
target_s[bound] = merged
def shrink():
"""Shrink the extending schemas which has just an end type, e.g. at this point
ota / port is type schema with extended pointing to core.port, this should instead be
@@ -638,22 +724,50 @@ def shrink():
# then are all simple types, integer and strings
for x, paths in referenced_schemas.items():
key_s = get_str_path_schema(x)
if key_s and key_s.get(S_TYPE) in ["enum", "registry", "integer", "string"]:
if key_s[S_TYPE] == "registry":
# Spread scalar leaf schemas (a single ``type`` with no nested schema or
# config vars) onto each referencing field so the type is inline. This
# covers enum/registry/integer/string plus float_with_unit quantities
# (e.g. a ``core.frequency`` schema typed ``frequency``), time and
# lambda -- but never structural schemas, which stay as references.
key_type = key_s.get(S_TYPE) if key_s else None
if (
key_type is not None
and key_type not in ("schema", "typed", "trigger", "pin", "use_id")
and S_SCHEMA not in key_s
and S_CONFIG_VARS not in key_s
):
if key_type == "registry":
print("Spreading registry: " + x)
for target in paths:
target_s = get_arr_path_schema(target)
if S_SCHEMA not in target_s:
print("skipping simple spread for " + ".".join(target))
continue
assert target_s[S_SCHEMA][S_EXTENDS] == [x]
extends = target_s[S_SCHEMA][S_EXTENDS]
if x not in extends:
# Already handled on an earlier visit (a field can list the
# same schema reference more than once).
continue
if len(extends) > 1:
# The field references several schemas at once (e.g. a value
# that extends both hex_uint8_t and uint8_t). Drop this
# reference and let the remaining one(s) describe the type,
# rather than forcing a single-extends spread here.
extends.remove(x)
continue
assert extends == [x]
target_s.pop(S_SCHEMA)
own_min, own_max = target_s.get("min"), target_s.get("max")
target_s |= key_s
_apply_tighter_bounds(target_s, own_min, own_max, key_s)
if key_s[S_TYPE] in ["integer", "string"]:
target_s["data_type"] = x.split(".")[1]
# remove this dangling again
pop_str_path_schema(x)
elif not key_s:
elif not key_s or set(key_s) <= {"min", "max"}:
# An untyped named schema, or one carrying only range bounds (e.g.
# positive_float = All(float_, Range(min=0)) has no scalar type but a
# min). Spread its data_type name and any bounds onto each field.
for target in paths:
target_s = get_arr_path_schema(target)
if S_SCHEMA not in target_s:
@@ -661,9 +775,12 @@ def shrink():
target_s[S_EXTENDS].remove(x)
continue
assert x in target_s[S_SCHEMA][S_EXTENDS]
own_min, own_max = target_s.get("min"), target_s.get("max")
target_s.pop(S_SCHEMA)
target_s.pop(S_TYPE) # undefined
target_s["data_type"] = x.split(".")[1]
target_s.update(key_s) # carry min/max bounds, if any
_apply_tighter_bounds(target_s, own_min, own_max, key_s)
# remove this dangling again
pop_str_path_schema(x)
@@ -910,7 +1027,12 @@ def convert(schema, config_var, path):
if isinstance(schema, cv.SensitiveValidator):
config_var["sensitive"] = True
config_var["sensitive_source"] = "explicit"
convert(schema.inner, config_var, f"{path}/sensitive")
if isinstance(schema, cv.BindKeyValidator):
# Its inner is the bound ``_validate`` method (a hex-key string);
# walking it yields no type, so describe it directly.
config_var[S_TYPE] = "string"
else:
convert(schema.inner, config_var, f"{path}/sensitive")
return
if isinstance(schema, cv.All):
@@ -943,16 +1065,44 @@ def convert(schema, config_var, path):
if DUMP_RAW:
config_var["raw"] = repr_schema
# A numeric range constraint (from cv.int_range / cv.float_range / a bare
# vol.Range in an All) contributes bounds, not a type. Attach them at the
# config var level, next to ``type``, so editors can validate the range.
if isinstance(schema, vol.Range):
# min/max may be non-numeric (e.g. a TimePeriod for a time-period range);
# keep numbers as-is and stringify anything else so the dump stays JSON
# serializable.
if schema.min is not None:
config_var["min"] = (
schema.min if isinstance(schema.min, (int, float)) else str(schema.min)
)
if schema.max is not None:
config_var["max"] = (
schema.max if isinstance(schema.max, (int, float)) else str(schema.max)
)
return
# pylint: disable=comparison-with-callable
if schema == cv.boolean:
config_var[S_TYPE] = "boolean"
elif schema == automation.validate_potentially_and_condition:
config_var[S_TYPE] = "registry"
config_var["registry"] = "condition"
elif schema in (cv.int_, cv.int_range):
elif schema in (cv.int_, cv.int_range) or schema in _CV_INTEGER_VALIDATORS:
config_var[S_TYPE] = "integer"
elif schema in (cv.string, cv.string_strict, cv.valid_name):
elif schema in (cv.string, cv.string_strict, cv.valid_name) or (
schema in _CV_STRING_VALIDATORS
):
config_var[S_TYPE] = "string"
elif callable(schema) and schema in _CV_UNIT_VALIDATORS:
config_var[S_TYPE], config_var["unit"] = _CV_UNIT_VALIDATORS[schema]
elif schema in _CV_TIME_VALIDATORS:
config_var[S_TYPE] = "time"
elif schema in _CV_LAMBDA_VALIDATORS:
config_var[S_TYPE] = "lambda"
elif schema == cv.entity_category:
config_var[S_TYPE] = "enum"
config_var["values"] = dict.fromkeys(cv.ENTITY_CATEGORIES)
elif isinstance(schema, vol.Schema):
# test: esphome/project
@@ -1026,6 +1176,18 @@ def convert(schema, config_var, path):
config_var[S_TYPE] = "registry"
config_var["registry"] = "light.effects"
config_var["filter"] = data[0]
elif schema_type == "float":
config_var[S_TYPE] = "float"
# cv.float_with_unit reports its canonical unit (e.g. "Hz", "V") for
# SCHEMA_EXTRACT; surface it next to the type so editors can label
# the field. Other float sources return None (no unit).
if isinstance(data, str):
config_var["unit"] = data
elif schema_type in ("string", "integer", "time", "lambda"):
# Scalar validators (e.g. cv.date_time) that declare their result
# type via schema_extractor. ``data`` is unused (the decorated
# validator returns None for SCHEMA_EXTRACT).
config_var[S_TYPE] = schema_type
elif schema_type == "templatable":
config_var["templatable"] = True
convert(data, config_var, path + "/templat")
+217
View File
@@ -368,3 +368,220 @@ def test_lvgl_style_schemas_are_named_and_deduped(lvgl_schema: dict) -> None:
_count(lvgl_schema)
assert refs > 100, f"STYLE_SCHEMA should be referenced via extends, got {refs}"
# ---------------------------------------------------------------------------
# Typing of esphome.config_validation validators.
#
# These validators used to fall through convert() with no ``type``, leaving the
# visual editor / dashboard unable to tell what YAML the field accepts. They are
# now described either by identity (scalar leaf validators) or via a
# schema_extractor decorator (factory-produced closures like float_with_unit).
# ---------------------------------------------------------------------------
def _convert(validator: object) -> dict:
config_var: dict = {}
_bls.convert(validator, config_var, "/x")
return config_var
@pytest.mark.parametrize(
("validator", "expected"),
[
(cv.icon, "string"),
(cv.mac_address, "string"),
(cv.url, "string"),
(cv.uuid, "string"),
(cv.directory, "string"),
(cv.mqtt_qos, "integer"),
(cv.hex_int, "integer"),
(cv.update_interval, "time"),
(cv.time_period_str_colon, "time"),
(cv.lambda_, "lambda"),
(cv.returning_lambda, "lambda"),
],
)
def test_convert_types_scalar_cv_validators(validator: object, expected: str) -> None:
assert _convert(validator).get("type") == expected
@pytest.mark.parametrize(
("validator", "expected_type", "expected_unit"),
[
(cv.percentage, "float", "%"),
(cv.possibly_negative_percentage, "float", "%"),
(cv.temperature, "float", "°C"),
(cv.temperature_delta, "float", "°C"),
(cv.color_temperature, "float", "mireds"),
(cv.percentage_int, "integer", "%"),
(cv.validate_bytes, "integer", "B"),
],
)
def test_convert_types_unit_validators(
validator: object, expected_type: str, expected_unit: str
) -> None:
# Number-with-unit validators accept both a bare number and a unit-suffixed
# string, so they carry their base numeric type plus the canonical unit.
entry = _convert(validator)
assert entry.get("type") == expected_type
assert entry.get("unit") == expected_unit
def test_convert_dimensions_left_untyped() -> None:
# cv.dimensions accepts either a "WIDTHxHEIGHT" string or a [w, h] list;
# a single scalar type would reject one form, so it is left untyped.
entry = _convert(cv.dimensions)
assert "type" not in entry
def test_convert_entity_category_is_enum() -> None:
entry = _convert(cv.entity_category)
assert entry["type"] == "enum"
assert set(entry["values"]) == set(cv.ENTITY_CATEGORIES)
def test_convert_bind_key_is_sensitive_string() -> None:
entry = _convert(cv.bind_key)
assert entry["type"] == "string"
assert entry["sensitive"] is True
@pytest.mark.parametrize("scalar", ["string", "integer", "float", "time", "lambda"])
def test_convert_scalar_schema_extractor(scalar: str) -> None:
"""A validator that declares a scalar type via schema_extractor is typed.
Mirrors cv.float_with_unit / cv.date_time, whose decorated closures return
None for SCHEMA_EXTRACT and are keyed into hidden_schemas by repr.
"""
from esphome import schema_extractors as ejs
def decorated(value: object) -> None:
return None
ejs.hidden_schemas[repr(decorated)] = scalar
try:
assert _convert(decorated).get("type") == scalar
finally:
del ejs.hidden_schemas[repr(decorated)]
def test_convert_float_with_unit_reports_unit() -> None:
"""A float schema_extractor whose probe returns a unit types float + unit."""
import voluptuous as vol
from esphome import schema_extractors as ejs
def frequency_validator(value: object) -> object:
return "Hz" if value is ejs.SCHEMA_EXTRACT else value
ejs.hidden_schemas[repr(frequency_validator)] = "float"
try:
entry = _convert(frequency_validator)
assert entry.get("type") == "float"
assert entry.get("unit") == "Hz"
finally:
del ejs.hidden_schemas[repr(frequency_validator)]
# A float source with no unit is a bare float (no unit key).
def plain_float(value: object) -> object:
return None if value is ejs.SCHEMA_EXTRACT else value
ejs.hidden_schemas[repr(plain_float)] = "float"
try:
entry = _convert(plain_float)
assert entry.get("type") == "float"
assert "unit" not in entry
finally:
del ejs.hidden_schemas[repr(plain_float)]
# A vol.Range in the All contributes min/max next to type, not a type.
assert _convert(vol.Range(min=45.0, max=66.0)) == {"min": 45.0, "max": 66.0}
def test_convert_range_stringifies_non_numeric_bounds() -> None:
"""A time-period range keeps JSON-serializable string bounds."""
import voluptuous as vol
entry = _convert(vol.Range(min=cv.time_period("1s"), max=cv.time_period("10s")))
assert isinstance(entry["min"], str) and isinstance(entry["max"], str)
@pytest.fixture(scope="module")
def full_schema_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
"""Run the full build once (fresh interpreter, see ``lvgl_schema``).
PYTHONPATH points at this worktree so the subprocess imports the local
esphome (with the config_validation changes) rather than an editable install
that may resolve to a different checkout.
"""
import os
out_dir = tmp_path_factory.mktemp("cv_types_schema")
repo_root = SCRIPT_PATH.parent.parent
subprocess.run(
[sys.executable, str(SCRIPT_PATH), "--output-path", str(out_dir)],
check=True,
capture_output=True,
text=True,
cwd=str(repo_root),
env={**os.environ, "PYTHONPATH": str(repo_root)},
)
return out_dir
def test_cv_types_end_to_end(full_schema_dir: Path) -> None:
"""The full build types config_validation fields end-to-end.
Also covers the shrink() spread of a field that references two typed schemas
at once (hex_uint8_t + uint8_t), which previously tripped an assertion.
"""
core = json.loads((full_schema_dir / "esphome.json").read_text())["core"]
entity = core["schemas"]["ENTITY_BASE_SCHEMA"]["schema"]["config_vars"]
assert entity["icon"]["type"] == "string"
assert entity["entity_category"]["type"] == "enum"
climate = json.loads((full_schema_dir / "climate.json").read_text())["climate"]
visual = climate["schemas"]["_CLIMATE_SCHEMA"]["schema"]["config_vars"]["visual"]
min_temp = visual["schema"]["config_vars"]["min_temperature"]
assert min_temp["type"] == "float"
assert min_temp["unit"] == "°C"
# message_type references both hex_uint8_t and uint8_t; shrink() must spread
# it to integer instead of tripping the single-extends assertion.
remote = json.loads((full_schema_dir / "remote_receiver.json").read_text())
abbwelcome = remote["remote_receiver.binary_sensor"]["schemas"]["CONFIG_SCHEMA"][
"schema"
]["config_vars"]["abbwelcome"]["schema"]["config_vars"]
assert abbwelcome["message_type"]["type"] == "integer"
# cv.All(cv.frequency, cv.float_range(45, 66)) -> float + unit + min/max
# inline, next to type.
ade = json.loads((full_schema_dir / "ade7880.json").read_text())
freq = ade["ade7880.sensor"]["schemas"]["CONFIG_SCHEMA"]["schema"]["config_vars"][
"frequency"
]
assert freq["type"] == "float"
assert freq["unit"] == "Hz"
assert freq["min"] == 45.0
assert freq["max"] == 66.0
# cv.All(cv.uint8_t, cv.Range(min=1, max=6)): spreading the uint8_t named
# schema (0..255) must not clobber the field's own tighter 1..6 bound.
tm1637 = json.loads((full_schema_dir / "tm1637.json").read_text())
length = tm1637["tm1637.display"]["schemas"]["CONFIG_SCHEMA"]["schema"][
"config_vars"
]["length"]
assert length["type"] == "integer"
assert length["data_type"] == "uint8_t"
assert length["min"] == 1
assert length["max"] == 6
# positive_float = All(float_, Range(min=0)): a bounds-only named schema
# spreads its data_type name and its min onto the field.
light = json.loads((full_schema_dir / "light.json").read_text())["light"]
gamma = light["schemas"]["BRIGHTNESS_ONLY_LIGHT_SCHEMA"]["schema"]["config_vars"][
"gamma_correct"
]
assert gamma["data_type"] == "positive_float"
assert gamma["min"] == 0
@@ -1736,6 +1736,22 @@ def test_templatable_schema_extract() -> None:
assert cv.templatable(cv.int_)(SCHEMA_EXTRACT) is cv.int_
@pytest.mark.parametrize(
("validator", "unit"),
[
(cv.float_with_unit("frequency", "(Hz|HZ|hz)?"), "Hz"),
(cv.frequency, "Hz"),
(cv.voltage, "V"),
(cv.current, "A"),
(cv.decibel, "dB"),
],
)
def test_float_with_unit_schema_extract(validator: object, unit: str) -> None:
# For the SCHEMA_EXTRACT sentinel the validator returns its canonical unit
# (not the parsed value) so build_language_schema can label the float field.
assert validator(SCHEMA_EXTRACT) == unit
def test_templatable_lambda() -> None:
result = cv.templatable(cv.int_)(Lambda("return 5;"))
assert isinstance(result, Lambda)