Add elevation option for observer obstruction vs horizon

custom_components/sun2/__init__.py

@@ -24,7 +24,7 @@
 from homeassistant.helpers.typing import ConfigType
 
 from .const import DOMAIN, SIG_HA_LOC_UPDATED
-from .helpers import LocData, LocParams, Sun2Data
+from .helpers import LocData, Sun2Data
 
 PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
 _OLD_UNIQUE_ID = re.compile(r"[0-9a-f]{32}-([0-9a-f]{32})")
@@ -36,14 +36,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 
     def update_local_loc_data() -> LocData:
         """Update local location data from HA's config."""
-        cast(Sun2Data, hass.data[DOMAIN]).locations[None] = loc_data = LocData(
-            LocParams(
-                hass.config.elevation,
-                hass.config.latitude,
-                hass.config.longitude,
-                str(hass.config.time_zone),
-            )
-        )
+        loc_data = LocData.from_hass_config(hass)
+        cast(Sun2Data, hass.data[DOMAIN]).locations[None] = loc_data
         return loc_data
 
     async def process_config(

custom_components/sun2/binary_sensor.py

@@ -29,7 +29,6 @@
     Num,
     Sun2Entity,
     Sun2EntityParams,
-    get_loc_params,
     nearest_second,
     sun2_dev_info,
     translate,
@@ -275,12 +274,12 @@ async def async_setup_entry(
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up the sensor platform."""
-    config = entry.options
+    options = entry.options
     async_add_entities(
         _sensors(
-            get_loc_params(config),
+            LocParams.from_entry_options(options),
             Sun2EntityParams(entry, sun2_dev_info(hass, entry)),
-            config.get(CONF_BINARY_SENSORS, []),
+            options.get(CONF_BINARY_SENSORS, []),
             hass,
         ),
         True,

custom_components/sun2/config.py

@@ -1,6 +1,8 @@
 """Sun2 config validation."""
 from __future__ import annotations
 
+from collections.abc import Mapping
+import logging
 from typing import cast
 
 from astral import SunDirection
@@ -23,21 +25,23 @@
 from homeassistant.helpers.typing import ConfigType
 
 from .const import (
+    CONF_ABOVE_GROUND,
     CONF_DIRECTION,
+    CONF_DISTANCE,
     CONF_ELEVATION_AT_TIME,
+    CONF_OBS_ELV,
+    CONF_RELATIVE_HEIGHT,
+    CONF_SUNRISE_OBSTRUCTION,
+    CONF_SUNSET_OBSTRUCTION,
     CONF_TIME_AT_ELEVATION,
     DOMAIN,
 )
 from .helpers import init_translations
 
-PACKAGE_MERGE_HINT = "list"
+_LOGGER = logging.getLogger(__name__)
 
-LOC_PARAMS = {
-    vol.Inclusive(CONF_ELEVATION, "location"): vol.Coerce(float),
-    vol.Inclusive(CONF_LATITUDE, "location"): cv.latitude,
-    vol.Inclusive(CONF_LONGITUDE, "location"): cv.longitude,
-    vol.Inclusive(CONF_TIME_ZONE, "location"): cv.time_zone,
-}
+PACKAGE_MERGE_HINT = "list"
+SUN_DIRECTIONS = [dir.lower() for dir in SunDirection.__members__]
 
 _SUN2_BINARY_SENSOR_SCHEMA = vol.Schema(
     {
@@ -51,8 +55,9 @@
     }
 )
 
-ELEVATION_AT_TIME_SCHEMA_BASE = vol.Schema(
+_ELEVATION_AT_TIME_SCHEMA = vol.Schema(
     {
+        vol.Required(CONF_UNIQUE_ID): cv.string,
         vol.Required(CONF_ELEVATION_AT_TIME): vol.Any(
             vol.All(cv.string, cv.entity_domain("input_datetime")),
             cv.time,
@@ -62,14 +67,9 @@
     }
 )
 
-_ELEVATION_AT_TIME_SCHEMA = ELEVATION_AT_TIME_SCHEMA_BASE.extend(
-    {vol.Required(CONF_UNIQUE_ID): cv.string}
-)
-
-SUN_DIRECTIONS = [dir.lower() for dir in SunDirection.__members__]
-
-TIME_AT_ELEVATION_SCHEMA_BASE = vol.Schema(
+_TIME_AT_ELEVATION_SCHEMA = vol.Schema(
     {
+        vol.Required(CONF_UNIQUE_ID): cv.string,
         vol.Required(CONF_TIME_AT_ELEVATION): vol.All(
             vol.Coerce(float), vol.Range(min=-90, max=90), msg="invalid elevation"
         ),
@@ -79,10 +79,6 @@
     }
 )
 
-_TIME_AT_ELEVATION_SCHEMA = TIME_AT_ELEVATION_SCHEMA_BASE.extend(
-    {vol.Required(CONF_UNIQUE_ID): cv.string}
-)
-
 
 def _sensor(config: ConfigType) -> ConfigType:
     """Validate sensor config."""
@@ -93,15 +89,21 @@ def _sensor(config: ConfigType) -> ConfigType:
     raise vol.Invalid(f"expected {CONF_ELEVATION_AT_TIME} or {CONF_TIME_AT_ELEVATION}")
 
 
-_SUN2_LOCATION_CONFIG = vol.Schema(
+_SUN2_LOCATION_SCHEMA = vol.Schema(
     {
         vol.Required(CONF_UNIQUE_ID): cv.string,
         vol.Inclusive(CONF_LOCATION, "location"): cv.string,
+        vol.Inclusive(CONF_LATITUDE, "location"): cv.latitude,
+        vol.Inclusive(CONF_LONGITUDE, "location"): cv.longitude,
+        vol.Inclusive(CONF_TIME_ZONE, "location"): cv.time_zone,
+        vol.Optional(CONF_ELEVATION): vol.Coerce(float),
+        vol.Optional(CONF_OBS_ELV): vol.Any(
+            vol.Coerce(float), dict, msg="expected a float or a dictionary"
+        ),
         vol.Optional(CONF_BINARY_SENSORS): vol.All(
             cv.ensure_list, [_SUN2_BINARY_SENSOR_SCHEMA]
         ),
         vol.Optional(CONF_SENSORS): vol.All(cv.ensure_list, [_sensor]),
-        **LOC_PARAMS,
     }
 )
 
@@ -119,18 +121,110 @@ def _unique_locations_names(configs: list[dict]) -> list[dict]:
         vol.Optional(DOMAIN): vol.All(
             lambda config: config or [],
             cv.ensure_list,
-            [_SUN2_LOCATION_CONFIG],
+            [_SUN2_LOCATION_SCHEMA],
             _unique_locations_names,
         ),
     },
     extra=vol.ALLOW_EXTRA,
 )
 
 
+_OBSTRUCTION_CONFIG = {
+    vol.Required(CONF_DISTANCE): vol.Coerce(float),
+    vol.Required(CONF_RELATIVE_HEIGHT): vol.Coerce(float),
+}
+_OBS_ELV_DICT = {
+    vol.Optional(CONF_ABOVE_GROUND): vol.Coerce(float),
+    vol.Optional(CONF_SUNRISE_OBSTRUCTION): _OBSTRUCTION_CONFIG,
+    vol.Optional(CONF_SUNSET_OBSTRUCTION): _OBSTRUCTION_CONFIG,
+}
+_OBS_ELV_KEYS = [option.schema for option in _OBS_ELV_DICT]
+_OBS_ELV_INVALID_LEN_MSG = f"use exactly two of: {', '.join(_OBS_ELV_KEYS)}"
+_OBS_ELV_DICT_SCHEMA = vol.All(
+    vol.Schema(_OBS_ELV_DICT), vol.Length(2, 2, msg=_OBS_ELV_INVALID_LEN_MSG)
+)
+
+
+def _obs_elv(
+    obstruction: Mapping[str, float] | None, above_ground: float | None
+) -> float | list[float]:
+    """Determine observer elevation from obstruction or elevation above ground level."""
+    if obstruction:
+        return [obstruction[CONF_RELATIVE_HEIGHT], obstruction[CONF_DISTANCE]]
+    assert above_ground is not None
+    return above_ground
+
+
+def _validate_and_convert_observer(
+    loc_config: ConfigType, idx: int, home_elevation: int
+) -> None:
+    """Validate observer elevation option in location config.
+
+    If deprecated elevation option is present, warn or raise exception,
+    but leave as-is (i.e., do not convert to observer elevation option.)
+    Just continue to use elevation option until user replaces deprecated
+    option with new option.
+
+    Otherwise, convert to list[float | list[float]] where
+    list[0] is east (sunrise) observer_elevation,
+    list[1] is west (sunset) observer_elevation,
+    observer_elevation is float or list[float] where
+    float is elevation above ground level or
+    list[0] is height of obstruction relative to observer
+    list[1] is distance to obstruction from observer
+    """
+    east_obs_elv: float | list[float]
+    west_obs_elv: float | list[float]
+
+    try:
+        if CONF_ELEVATION in loc_config:
+            cv.has_at_most_one_key(CONF_ELEVATION, CONF_OBS_ELV)(loc_config)
+            # Pass in copy of config so elevation option does not get removed.
+            cv.deprecated(CONF_ELEVATION, CONF_OBS_ELV)(dict(loc_config))
+            return
+
+        if CONF_OBS_ELV not in loc_config:
+            # TODO: Make this a repair issue???
+            _LOGGER.warning(
+                "New config option %s missing @ data[%s][%i], "
+                "will use system general elevation setting",
+                CONF_OBS_ELV,
+                DOMAIN,
+                idx,
+            )
+            east_obs_elv = west_obs_elv = float(home_elevation)
+
+        elif isinstance(obs := loc_config[CONF_OBS_ELV], float):
+            east_obs_elv = west_obs_elv = obs
+
+        else:
+            try:
+                _OBS_ELV_DICT_SCHEMA(obs)
+            except vol.Invalid as err:
+                err.prepend([CONF_OBS_ELV])
+                raise
+            above_ground = obs.get(CONF_ABOVE_GROUND)
+            east_obs_elv = _obs_elv(obs.get(CONF_SUNRISE_OBSTRUCTION), above_ground)
+            west_obs_elv = _obs_elv(obs.get(CONF_SUNSET_OBSTRUCTION), above_ground)
+
+    except vol.Invalid as err:
+        err.prepend([DOMAIN, idx])
+        raise
+
+    loc_config[CONF_OBS_ELV] = [east_obs_elv, west_obs_elv]
+
+
 async def async_validate_config(
     hass: HomeAssistant, config: ConfigType
 ) -> ConfigType | None:
     """Validate configuration."""
     await init_translations(hass)
 
-    return cast(ConfigType, _SUN2_CONFIG_SCHEMA(config))
+    config = _SUN2_CONFIG_SCHEMA(config)
+    if DOMAIN not in config:
+        return config
+
+    home_elevation = hass.config.elevation
+    for idx, loc_config in enumerate(config[DOMAIN]):
+        _validate_and_convert_observer(loc_config, idx, home_elevation)
+    return config