Compare commits

...
Author SHA1 Message Date
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 332 additions and 9 deletions
+9 -2
View File
@@ -1364,8 +1364,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)
@@ -1389,8 +1396,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)
+129 -7
View File
@@ -120,6 +120,56 @@ 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.dimensions,
cv.none,
)
_CV_INTEGER_VALIDATORS = (cv.hex_int, cv.percentage_int, cv.mqtt_qos)
_CV_FLOAT_VALIDATORS = (
cv.percentage,
cv.possibly_negative_percentage,
cv.temperature,
cv.temperature_delta,
cv.color_temperature,
)
_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):
@@ -625,22 +675,48 @@ 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)
target_s |= 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:
@@ -651,6 +727,7 @@ def shrink():
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
# remove this dangling again
pop_str_path_schema(x)
@@ -897,7 +974,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):
@@ -930,16 +1012,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 schema in _CV_FLOAT_VALIDATORS:
config_var[S_TYPE] = "float"
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
@@ -1013,6 +1123,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")
+178
View File
@@ -305,3 +305,181 @@ 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.percentage, "float"),
(cv.temperature, "float"),
(cv.color_temperature, "float"),
(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
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"]
assert visual["schema"]["config_vars"]["min_temperature"]["type"] == "float"
# 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
# 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
@@ -1641,6 +1641,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)