ambient.py

"""
ambient.py

Custom component for Home Assistant that scrapes the Ambient Weather Station page, collects data then creates sensors and events

Documentation: https://gist.github.com/JerryWorkman/6222d0a7b260a73323b884cf8d3cb8ea

Ambient Weather WS-1400-IP OBSERVER
http://www.ambientweather.com/amws1400ip.html

To install beautiful soup on a raspian/Debian/Ubuntu:
sudo apt install python3-bs4
source /srv/hass/hass_venv/bin/activate
pip3 install bs4
deactivate

Note: A Rasberry Pi 2 or smaller does not have enough resources to handle beautiful soup

Configuration:

Add the following to your configuration.yaml file.

sensor ambient:
  platform: ambient
    host: hostname	#hostname or ipaddress of weather station
    exclude: None   #(optional) default: receiver_time,yearly_rain,indoor_sensor,outdoor_sensor1,outdoor_sensor2,unnamed_device
    prefix: Ambient #text prefixed to sensor name

Note: a group with the name of the prefix (e.g. Ambient) is created with all sensors
If you wish to createyour own group I suggest the following:

group:
  Ambient Weather Station Sensors:
    view: yes
    entities:
      - sensor.ambient_monthly_rain
      - sensor.ambient_wind_gust
      - sensor.ambient_daily_rain
      - sensor.ambient_solar_radiation
      - sensor.ambient_hourly_rain_rate
      - sensor.ambient_weekly_rain
      - sensor.ambient_uv
      - sensor.ambient_outdoor_temperature
      - sensor.ambient_wind_speed
      - sensor.ambient_wind_direction
      - sensor.ambient_uvi
      - sensor.ambient_outdoor_humidity

"""
import logging
import sched
import time
from urllib.request import urlopen, Request
from urllib.error import URLError
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (CONF_HOST, STATE_UNKNOWN)
from homeassistant.helpers.entity import Entity
import homeassistant.helpers.config_validation as cv
import homeassistant.loader as loader
from bs4 import BeautifulSoup
import pprint
pp = pprint.PrettyPrinter(indent=4)

_LOGGER = logging.getLogger(__name__)

REQUIREMENTS = ['beautifulsoup4==4.5.1']
DOMAIN = 'sensor.ambient'
CONF_EXCLUDE = 'exclude'
CONF_PREFIX = 'prefix'
DEFAULT_HOST = None
DEFAULT_EXCLUDE = 'receiver_time,yearly_rain,indoor_sensor,outdoor_sensor1,outdoor_sensor2,unnamed_device'
DEFAULT_PREFIX = 'Ambient'

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_HOST, default=DEFAULT_HOST): cv.string,
    vol.Optional(CONF_EXCLUDE, default=DEFAULT_EXCLUDE): cv.string,
    vol.Optional(CONF_PREFIX, default=DEFAULT_PREFIX): cv.string,
})


def _log_error(msg):
    _LOGGER.error('%s: %s', DOMAIN, str(msg))


def _debug(msg):
    _LOGGER.debug(msg)


class AmbientServer():
    """Ambient weather server scraper"""
    def __init__(self, hass, AddDevices, host, exclude, prefix):
        self._hass = hass
        self._AddDevices = AddDevices
        self._url = "http://%s/livedata.htm" % host
        self._exclude = exclude
        self._prefix = prefix
        self._timeout = 30
        self._data = {}
        self.get_weather_data()
        self.scrape_page()
        self._sch = sched.scheduler(time.time, time.sleep)
        self._sch.enter(60, 1, self.get_weather_data, (self._sch,))

    @property
    def data(self):
        return self._data

    def get_value(self, name):
        rec = self._data.get(name, None)
        if rec == None:
            return None
        return rec['value']

    def get_friendly_name(self, name):
        rec = self._data.get(name, None)
        if rec == None:
            return None
        return rec['friendly_name']

    def get_units(self, name):
        ut = {'speed': 'mph', 'rain': 'in', 'speed': 'mph', 'gust': 'mph', 'direction': 'deg',
              'rain': 'inches', 'temp': 'deg', 'humi': 'pct', 'solar': 'W/m^2', 'uv': 'mW/cm2',
              'uvi': 'mW/m2',}
        for key, value in ut.items():
            if key in name:
                return value
        return ""

    def check_server(self):
        """ Tests a web server using a HEAD request """
        request = Request(self._url)
        request.get_method = lambda: 'HEAD'
        try:
            response = urlopen(request)
            return response.info()
        except Exception:
            return False

    def validate_sensor(self, haname, friendly_name, value):
        """make sure sensor is valid and needed"""
        exclude = self._exclude.strip().replace(" ", "").lower().split(",")
        if not (friendly_name.endswith("ID") or friendly_name.endswith("Default")
                or str(value)=='' or '-' in str(value) or haname==''):
            ascii_str = self._prefix.encode('ascii', 'ignore')
            l = len(ascii_str) + 1
            haname = haname[l:]
            if not ((haname in exclude) or (len(haname) < 2)):
                return True
        return False

    def make_haname(self, friendly_name):
        return friendly_name.strip().replace(" ", "_").lower()

    def scrape_page(self):
        """scrape page and create sensors"""
        _debug("creating sensors")
        entity_ids = []
        for name in self._data.keys():
            state = self._data.get(name, STATE_UNKNOWN)
            friendly_name = self.get_friendly_name(name)
            haname = self.make_haname(friendly_name)
            units = self.get_units(haname)
            if self.validate_sensor(haname, friendly_name, state['value']):
                self._AddDevices([AmbientSensor(self._hass, self, name, haname, units, friendly_name)])
                entity_ids.append("sensor." + haname)

        group = loader.get_component('group')
        gp = group.Group.create_group(self._hass, self._prefix, entity_ids=entity_ids, view=False)


    def get_weather_data(self):
        _debug("Opening %s" % self._url)
        if not self.check_server():
            _LOGGER.error("%s is not responding.", self._url)
            return
        html = urlopen(self._url, timeout=self._timeout).read() #give the server plenty time to respond
        if not html:
            _log_error("Unable to fetch data from %s" % self._url)
            return
        soup = BeautifulSoup(html, 'html.parser')
        for inp in soup.findAll('input'):
            try:
                name = inp['name']
                value = inp['value']
                fname = inp.parent.parent.contents[1].get_text()
                fname =  self._prefix + " " + fname.replace(" ID", "").replace(".", "").replace(":", "").strip()
                haname = self.make_haname(fname)
                if self.validate_sensor(haname, fname, value):
                    if value.isnumeric():
                        value = float(value)
                    self._data[haname] = {"friendly_name": fname, "value": value}
            except KeyError:
                pass


class AmbientSensor(Entity):
    """Representation of a web scrape sensor."""

    def __init__(self, hass, server, name, haname, units, friendly_name):
        """Initialize a web scrape sensor."""
        self._hass = hass
        self._server = server
        self._name = name
        self._haname = haname
        self._unit_of_measurement = units
        self._friendly_name = friendly_name
        self._state = STATE_UNKNOWN
        self.update()

    @property
    def name(self):
        """Return the name of the sensor."""
        return self._friendly_name

    @property
    def friendly_name(self):
        """Return the friendly_name of the sensor."""
        return self._friendly_name

    @property
    def unit_of_measurement(self):
        """Return the unit the value is expressed in."""
        return self._unit_of_measurement

    @property
    def state(self):
        """Return the state of the device."""
        return self._state

    def update(self):
        """Get the latest data from the source and updates the state."""
        value = self._server.get_value(self._haname)
        _debug("update %s: %s" % (self._name, value))
        self._state = value


# pylint: disable=unused-argument
def setup_platform(hass, config, add_devices, discovery_info=None):
    """set up Ambient platform."""
    host = config.get(CONF_HOST)
    if not host:
        _LOGGER.error("host is a required parameter")
        return False
    exclude = config.get(CONF_EXCLUDE, DEFAULT_EXCLUDE)
    prefix = config.get(CONF_PREFIX, DEFAULT_PREFIX)
    AmbientServer(hass, add_devices, host, exclude, prefix)