Initial version

.gitignore

@@ -0,0 +1 @@
+.DS_Store

README.md

@@ -0,0 +1,32 @@
+![Version](https://img.shields.io/github/v/release/MarcoGos/davis_vantage?include_prereleases)
+
+# Davis Vantage
+
+This is a custom integration for the Davis Vantage Pro2 and Vue.
+
+## Installation
+
+Via HACS:
+
+- Add the following custom repository as an integration:
+ - MarcoGos/davis_vantage
+- Restart Home Assistant
+- Add the integration to Home Assistant
+
+## Setup
+
+During the setup of the integration the serial port of the weather station needs to be provided.
+
+Examples:
+- tcp:192.168.0.18:1111
+- serial:/dev/ttyUSB0:19200:8N1
+
+![Setup](/assets/setup.png)
+
+## What to expect
+
+The following sensors will be registered:
+
+![Sensors](/assets/sensors.png)
+
+The sensor information is updated every 30 seconds.

custom_components/davis_vantage/__init__.py

@@ -0,0 +1,69 @@
+"""The Davis Vantage integration."""
+from __future__ import annotations
+from typing import Any
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceEntryType
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.const import Platform
+
+from .client import DavisVantageClient
+from .const import DOMAIN, NAME, VERSION, MANUFACTURER
+from .coordinator import DavisVantageDataUpdateCoordinator
+
+PLATFORMS: list[Platform] = [
+ Platform.SENSOR,
+ Platform.BINARY_SENSOR
+]
+
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
+async def async_setup(hass: HomeAssistant, config: Any) -> bool:
+ return True
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+ """Set up Davis Vantage from a config entry."""
+ if hass.data.get(DOMAIN) is None:
+ hass.data.setdefault(DOMAIN, {})
+
+ _LOGGER.debug(f"entry.data: {entry.data}")
+
+ protocol = entry.data.get("protocol", "")
+ link = entry.data.get("link", "")
+ rain_collector = entry.data.get("rain_collector", "0.01""")
+ windrose8 = entry.data.get("windrose8", False)
+
+ client = DavisVantageClient(hass, protocol, link, rain_collector, windrose8)
+
+ device_info = DeviceInfo(
+ entry_type=DeviceEntryType.SERVICE,
+ identifiers={(DOMAIN, entry.entry_id)},
+ manufacturer=MANUFACTURER,
+ name=NAME,
+ sw_version=VERSION
+ )
+
+ hass.data[DOMAIN][entry.entry_id] = coordinator = DavisVantageDataUpdateCoordinator(
+ hass=hass, client=client, device_info=device_info)
+
+ await coordinator.async_config_entry_first_refresh()
+
+ await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+ entry.async_on_unload(entry.add_update_listener(async_reload_entry))
+
+ return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+ """Unload a config entry."""
+ if unloaded := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+ hass.data[DOMAIN].pop(entry.entry_id)
+ return unloaded
+
+
+async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+ """Reload config entry."""
+ await async_unload_entry(hass, entry)
+ await async_setup_entry(hass, entry)

custom_components/davis_vantage/binary_sensor.py

@@ -0,0 +1,73 @@
+from homeassistant.components.binary_sensor import (
+ DOMAIN as BINARY_SENSOR_DOMAIN,
+ BinarySensorEntity,
+ BinarySensorEntityDescription
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DEFAULT_NAME, DOMAIN
+from .coordinator import DavisVantageDataUpdateCoordinator
+
+DESCRIPTIONS: list[BinarySensorEntityDescription] = [
+ BinarySensorEntityDescription(
+ key="IsRaining",
+ name="Is Raining"
+ )
+]
+
+async def async_setup_entry(
+ hass: HomeAssistant,
+ entry: ConfigEntry,
+ async_add_entities: AddEntitiesCallback,
+) -> None:
+ """Set up Davis Vantage sensors based on a config entry."""
+ coordinator = hass.data[DOMAIN][entry.entry_id]
+
+ entities: list[DavisVantageBinarySensor] = []
+
+ # Add all binary sensors described above.
+ for description in DESCRIPTIONS:
+ entities.append(
+ DavisVantageBinarySensor(
+ coordinator=coordinator,
+ entry_id=entry.entry_id,
+ description=description,
+ )
+ )
+
+ async_add_entities(entities)
+
+
+class DavisVantageBinarySensor(CoordinatorEntity[DavisVantageDataUpdateCoordinator], BinarySensorEntity):
+ """Defines a Davis Vantage sensor."""
+
+ _attr_has_entity_name = True
+
+ def __init__(
+ self,
+ coordinator: DavisVantageDataUpdateCoordinator,
+ entry_id: str,
+ description: BinarySensorEntityDescription,
+ ) -> None:
+ """Initialize Davis Vantage sensor."""
+ super().__init__(coordinator=coordinator)
+
+ self.entity_id = (
+ f"{BINARY_SENSOR_DOMAIN}.{DEFAULT_NAME}_{description.name}".lower()
+ )
+ self.entity_description = description
+ self._attr_name = description.name # type: ignore
+ self._attr_unique_id = f"{entry_id}-{DEFAULT_NAME} {self.name}"
+ self._attr_device_info = coordinator.device_info
+
+ @property
+ def is_on(self) -> bool | None:
+ """Return the is_on of the sensor."""
+ key = self.entity_description.key
+ data = self.coordinator.data
+ if key not in data:
+ return None
+ return data.get(key, False) # type: ignore

custom_components/davis_vantage/client.py

@@ -0,0 +1,95 @@
+from typing import Any
+from datetime import datetime
+import logging
+from pyvantagepro import VantagePro2
+from pyvantagepro.parser import LoopDataParserRevB
+
+from homeassistant.core import HomeAssistant
+
+from .utils import *
+from .const import RAIN_COLLECTOR_METRIC, PROTOCOL_NETWORK, PROTOCOL_SERIAL
+
+TIMEOUT = 10
+
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
+class DavisVantageClient:
+ def __init__(
+ self,
+ hass: HomeAssistant,
+ protocol: str,
+ link: str,
+ rain_collector: str,
+ windrose8: bool
+ ) -> None:
+ self._hass = hass
+ self._protocol = protocol
+ self._link = link
+ self._windrose8 = windrose8
+ self._rain_collector = rain_collector
+ self._last_data = LoopDataParserRevB
+
+ async def async_get_current_data(self) -> LoopDataParserRevB | None:
+ """Get current date from weather station."""
+ data = self._last_data
+ try:
+ vantagepro2 = VantagePro2.from_url(self.get_link()) # type: ignore
+ # vantagepro2.link.open() # type: ignore
+ data = vantagepro2.get_current_data()
+ if data:
+ self.add_additional_info(data)
+ self.convert_values(data)
+ data['LastError'] = "No error"
+ self._last_data = data
+ else:
+ data['LastError'] = "Couldn't acquire data"
+ except Exception as e:
+ _LOGGER.warning(f"Couldn't acquire data from {self._link}")
+ data['LastError'] = f"Couldn't acquire data: {e}" # type: ignore
+ # finally:
+ # vantagepro2.link.close() # type: ignore
+ return data # type: ignore
+
+ async def async_get_davis_time(self) -> datetime | None:
+ """Get time from weather station."""
+ data = None
+ try:
+ vantagepro2 = VantagePro2.from_url(self.get_link()) # type: ignore
+ # vantagepro2.link.open() # type: ignore
+ data = vantagepro2.gettime()
+ except Exception:
+ _LOGGER.warning(f"Couldn't acquire data from {self._link}")
+ # finally:
+ # vantagepro2.link.close() # type: ignore
+ return data
+
+ def add_additional_info(self, data: dict[str, Any]) -> None:
+ data['HeatIndex'] = calc_heat_index(data['TempOut'], data['HumOut'])
+ data['WindChill'] = calc_wind_chill(data['TempOut'], data['WindSpeed'])
+ data['FeelsLike'] = calc_feels_like(data['TempOut'], data['HumOut'], data['WindSpeed'])
+ data['WindDirRose'] = get_wind_rose(data['WindDir'], self._windrose8)
+ data['DewPoint'] = calc_dew_point(data['TempOut'], data['HumOut'])
+ data['WindSpeedBft'] = convert_kmh_to_bft(convert_to_kmh(data['WindSpeed10Min']))
+ data['IsRaining'] = data['RainRate'] > 0
+
+ def convert_values(self, data: dict[str, Any]) -> None:
+ data['Datetime'] = convert_to_iso_datetime(data['Datetime'], ZoneInfo(self._hass.config.time_zone))
+ data['BarTrend'] = get_baro_trend(data['BarTrend'])
+ data['UV'] = get_uv(data['UV'])
+ data['ForecastRuleNo'] = get_forecast_string(data['ForecastRuleNo'])
+ data['RainCollector'] = self._rain_collector
+ data['WindRoseSetup'] = 8 if self._windrose8 else 16
+ if data['RainCollector'] == RAIN_COLLECTOR_METRIC:
+ self.correct_rain_values(data)
+
+ def correct_rain_values(self, data: dict[str, Any]):
+ data['RainDay'] *= 2/2.54
+ data['RainMonth'] *= 2/2.54
+ data['RainYear'] *= 2/2.54
+ data['RainRate'] *= 2/2.54
+
+ def get_link(self) -> str | None:
+ if self._protocol == PROTOCOL_NETWORK:
+ return f"tcp:{self._link}"
+ if self._protocol == PROTOCOL_SERIAL:
+ return f"serial:{self._link}:19200:8N1"