jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove BOM integration because it uses webscraping (#41941)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Paulus Schoutsen 2020-10-17 13:56:11 +02:00 committed by GitHub
parent ee1b6d3195
commit 39ba0fc7ee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 0 additions and 725 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -98,9 +98,6 @@ omit =
homeassistant/components/bme680/sensor.py
homeassistant/components/bmp280/sensor.py
homeassistant/components/bmw_connected_drive/*
homeassistant/components/bom/camera.py
homeassistant/components/bom/sensor.py
homeassistant/components/bom/weather.py
homeassistant/components/braviatv/__init__.py
homeassistant/components/braviatv/const.py
homeassistant/components/braviatv/media_player.py

1
CODEOWNERS
View File

@ -60,7 +60,6 @@ homeassistant/components/blebox/* @gadgetmobile
homeassistant/components/blink/* @fronzbot
homeassistant/components/bmp280/* @belidzs
homeassistant/components/bmw_connected_drive/* @gerard33 @rikroe
homeassistant/components/bom/* @maddenp
homeassistant/components/bond/* @prystupa
homeassistant/components/braviatv/* @bieniu
homeassistant/components/broadlink/* @danielhiversen @felipediel

1
homeassistant/components/bom/__init__.py
View File

@ -1 +0,0 @@
"""The bom component."""

132
homeassistant/components/bom/camera.py
View File

@ -1,132 +0,0 @@
"""Provide animated GIF loops of BOM radar imagery."""
from bomradarloop import BOMRadarLoop
import voluptuous as vol
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
from homeassistant.const import CONF_ID, CONF_NAME
from homeassistant.helpers import config_validation as cv
CONF_DELTA = "delta"
CONF_FRAMES = "frames"
CONF_LOCATION = "location"
CONF_OUTFILE = "filename"
LOCATIONS = [
"Adelaide",
"Albany",
"AliceSprings",
"Bairnsdale",
"Bowen",
"Brisbane",
"Broome",
"Cairns",
"Canberra",
"Carnarvon",
"Ceduna",
"Dampier",
"Darwin",
"Emerald",
"Esperance",
"Geraldton",
"Giles",
"Gladstone",
"Gove",
"Grafton",
"Gympie",
"HallsCreek",
"Hobart",
"Kalgoorlie",
"Katherine",
"Learmonth",
"Longreach",
"Mackay",
"Marburg",
"Melbourne",
"Mildura",
"Moree",
"MorningtonIs",
"MountIsa",
"MtGambier",
"Namoi",
"Newcastle",
"Newdegate",
"NorfolkIs",
"NWTasmania",
"Perth",
"PortHedland",
"Rainbow",
"SellicksHill",
"SouthDoodlakine",
"Sydney",
"Townsville",
"WaggaWagga",
"Warrego",
"Warruwi",
"Watheroo",
"Weipa",
"WillisIs",
"Wollongong",
"Woomera",
"Wyndham",
"Yarrawonga",
]
def _validate_schema(config):
if config.get(CONF_LOCATION) is None:
if not all(config.get(x) for x in (CONF_ID, CONF_DELTA, CONF_FRAMES)):
raise vol.Invalid(
f"Specify '{CONF_ID}', '{CONF_DELTA}' and '{CONF_FRAMES}' when '{CONF_LOCATION}' is unspecified"
)
return config
LOCATIONS_MSG = f"Set '{CONF_LOCATION}' to one of: {', '.join(sorted(LOCATIONS))}"
XOR_MSG = f"Specify exactly one of '{CONF_ID}' or '{CONF_LOCATION}'"
PLATFORM_SCHEMA = vol.All(
PLATFORM_SCHEMA.extend(
{
vol.Exclusive(CONF_ID, "xor", msg=XOR_MSG): cv.string,
vol.Exclusive(CONF_LOCATION, "xor", msg=XOR_MSG): vol.In(
LOCATIONS, msg=LOCATIONS_MSG
),
vol.Optional(CONF_DELTA): cv.positive_int,
vol.Optional(CONF_FRAMES): cv.positive_int,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_OUTFILE): cv.string,
}
),
_validate_schema,
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up BOM radar-loop camera component."""
location = config.get(CONF_LOCATION) or f"ID {config.get(CONF_ID)}"
name = config.get(CONF_NAME) or f"BOM Radar Loop - {location}"
args = [
config.get(x)
for x in (CONF_LOCATION, CONF_ID, CONF_DELTA, CONF_FRAMES, CONF_OUTFILE)
]
add_entities([BOMRadarCam(name, *args)])
class BOMRadarCam(Camera):
"""A camera component producing animated BOM radar-imagery GIFs."""
def __init__(self, name, location, radar_id, delta, frames, outfile):
"""Initialize the component."""
super().__init__()
self._name = name
self._cam = BOMRadarLoop(location, radar_id, delta, frames, outfile)
def camera_image(self):
"""Return the current BOM radar-loop image."""
return self._cam.current
@property
def name(self):
"""Return the component name."""
return self._name

7
homeassistant/components/bom/manifest.json
View File

@ -1,7 +0,0 @@
{
"domain": "bom",
"name": "Australian Bureau of Meteorology (BOM)",
"documentation": "https://www.home-assistant.io/integrations/bom",
"requirements": ["bomradarloop==0.1.5"],
"codeowners": ["@maddenp"]
}

355
homeassistant/components/bom/sensor.py
View File

@ -1,355 +0,0 @@
"""Support for Australian BOM (Bureau of Meteorology) weather service."""
import datetime
import ftplib
import gzip
import io
import json
import logging
import os
import re
import zipfile
import requests
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_MONITORED_CONDITIONS,
CONF_NAME,
LENGTH_KILOMETERS,
LENGTH_METERS,
LENGTH_MILLIMETERS,
PERCENTAGE,
PRESSURE_MBAR,
SPEED_KILOMETERS_PER_HOUR,
TEMP_CELSIUS,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
import homeassistant.util.dt as dt_util
_LOGGER = logging.getLogger(__name__)
ATTR_LAST_UPDATE = "last_update"
ATTR_SENSOR_ID = "sensor_id"
ATTR_STATION_ID = "station_id"
ATTR_STATION_NAME = "station_name"
ATTR_ZONE_ID = "zone_id"
ATTRIBUTION = "Data provided by the Australian Bureau of Meteorology"
CONF_STATION = "station"
CONF_ZONE_ID = "zone_id"
CONF_WMO_ID = "wmo_id"
MIN_TIME_BETWEEN_UPDATES = datetime.timedelta(seconds=60)
SENSOR_TYPES = {
"wmo": ["wmo", None],
"name": ["Station Name", None],
"history_product": ["Zone", None],
"local_date_time": ["Local Time", None],
"local_date_time_full": ["Local Time Full", None],
"aifstime_utc": ["UTC Time Full", None],
"lat": ["Lat", None],
"lon": ["Long", None],
"apparent_t": ["Feels Like C", TEMP_CELSIUS],
"cloud": ["Cloud", None],
"cloud_base_m": ["Cloud Base", None],
"cloud_oktas": ["Cloud Oktas", None],
"cloud_type_id": ["Cloud Type ID", None],
"cloud_type": ["Cloud Type", None],
"delta_t": ["Delta Temp C", TEMP_CELSIUS],
"gust_kmh": ["Wind Gust kmh", SPEED_KILOMETERS_PER_HOUR],
"gust_kt": ["Wind Gust kt", "kt"],
"air_temp": ["Air Temp C", TEMP_CELSIUS],
"dewpt": ["Dew Point C", TEMP_CELSIUS],
"press": ["Pressure mb", PRESSURE_MBAR],
"press_qnh": ["Pressure qnh", "qnh"],
"press_msl": ["Pressure msl", "msl"],
"press_tend": ["Pressure Tend", None],
"rain_trace": ["Rain Today", LENGTH_MILLIMETERS],
"rel_hum": ["Relative Humidity", PERCENTAGE],
"sea_state": ["Sea State", None],
"swell_dir_worded": ["Swell Direction", None],
"swell_height": ["Swell Height", LENGTH_METERS],
"swell_period": ["Swell Period", None],
"vis_km": [f"Visability {LENGTH_KILOMETERS}", LENGTH_KILOMETERS],
"weather": ["Weather", None],
"wind_dir": ["Wind Direction", None],
"wind_spd_kmh": ["Wind Speed kmh", SPEED_KILOMETERS_PER_HOUR],
"wind_spd_kt": ["Wind Speed kt", "kt"],
}
def validate_station(station):
"""Check that the station ID is well-formed."""
if station is None:
return
station = station.replace(".shtml", "")
if not re.fullmatch(r"ID[A-Z]\d\d\d\d\d\.\d\d\d\d\d", station):
raise vol.error.Invalid("Malformed station ID")
return station
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Inclusive(CONF_ZONE_ID, "Deprecated partial station ID"): cv.string,
vol.Inclusive(CONF_WMO_ID, "Deprecated partial station ID"): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_STATION): validate_station,
vol.Required(CONF_MONITORED_CONDITIONS, default=[]): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]
),
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the BOM sensor."""
station = config.get(CONF_STATION)
zone_id, wmo_id = config.get(CONF_ZONE_ID), config.get(CONF_WMO_ID)
if station is not None:
if zone_id and wmo_id:
_LOGGER.warning(
"Using configuration %s, not %s and %s for BOM sensor",
CONF_STATION,
CONF_ZONE_ID,
CONF_WMO_ID,
)
elif zone_id and wmo_id:
station = f"{zone_id}.{wmo_id}"
else:
station = closest_station(
config.get(CONF_LATITUDE),
config.get(CONF_LONGITUDE),
hass.config.config_dir,
)
if station is None:
_LOGGER.error("Could not get BOM weather station from lat/lon")
return
bom_data = BOMCurrentData(station)
try:
bom_data.update()
except ValueError as err:
_LOGGER.error("Received error from BOM Current: %s", err)
return
add_entities(
[
BOMCurrentSensor(bom_data, variable, config.get(CONF_NAME))
for variable in config[CONF_MONITORED_CONDITIONS]
]
)
class BOMCurrentSensor(Entity):
"""Implementation of a BOM current sensor."""
def __init__(self, bom_data, condition, stationname):
"""Initialize the sensor."""
self.bom_data = bom_data
self._condition = condition
self.stationname = stationname
@property
def name(self):
"""Return the name of the sensor."""
if self.stationname is None:
return f"BOM {SENSOR_TYPES[self._condition][0]}"
return f"BOM {self.stationname} {SENSOR_TYPES[self._condition][0]}"
@property
def state(self):
"""Return the state of the sensor."""
return self.bom_data.get_reading(self._condition)
@property
def device_state_attributes(self):
"""Return the state attributes of the device."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
ATTR_LAST_UPDATE: self.bom_data.last_updated,
ATTR_SENSOR_ID: self._condition,
ATTR_STATION_ID: self.bom_data.latest_data["wmo"],
ATTR_STATION_NAME: self.bom_data.latest_data["name"],
ATTR_ZONE_ID: self.bom_data.latest_data["history_product"],
}
@property
def unit_of_measurement(self):
"""Return the units of measurement."""
return SENSOR_TYPES[self._condition][1]
def update(self):
"""Update current conditions."""
self.bom_data.update()
class BOMCurrentData:
"""Get data from BOM."""
def __init__(self, station_id):
"""Initialize the data object."""
self._zone_id, self._wmo_id = station_id.split(".")
self._data = None
self.last_updated = None
def _build_url(self):
"""Build the URL for the requests."""
url = (
f"http://www.bom.gov.au/fwo/{self._zone_id}"
f"/{self._zone_id}.{self._wmo_id}.json"
)
_LOGGER.debug("BOM URL: %s", url)
return url
@property
def latest_data(self):
"""Return the latest data object."""
if self._data:
return self._data[0]
return None
def get_reading(self, condition):
"""Return the value for the given condition.
BOM weather publishes condition readings for weather (and a few other
conditions) at intervals throughout the day. To avoid a `-` value in
the frontend for these conditions, we traverse the historical data
for the latest value that is not `-`.
Iterators are used in this method to avoid iterating needlessly
through the entire BOM provided dataset.
"""
condition_readings = (entry[condition] for entry in self._data)
reading = next((x for x in condition_readings if x != "-"), None)
if isinstance(reading, (int, float)):
return round(reading, 2)
return reading
def should_update(self):
"""Determine whether an update should occur.
BOM provides updated data every 30 minutes. We manually define
refreshing logic here rather than a throttle to keep updates
in lock-step with BOM.
If 35 minutes has passed since the last BOM data update, then
an update should be done.
"""
if self.last_updated is None:
# Never updated before, therefore an update should occur.
return True
now = dt_util.utcnow()
update_due_at = self.last_updated + datetime.timedelta(minutes=35)
return now > update_due_at
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Get the latest data from BOM."""
if not self.should_update():
_LOGGER.debug(
"BOM was updated %s minutes ago, skipping update as"
" < 35 minutes, Now: %s, LastUpdate: %s",
(dt_util.utcnow() - self.last_updated),
dt_util.utcnow(),
self.last_updated,
)
return
try:
result = requests.get(self._build_url(), timeout=10).json()
self._data = result["observations"]["data"]
# set lastupdate using self._data[0] as the first element in the
# array is the latest date in the json
self.last_updated = dt_util.as_utc(
datetime.datetime.strptime(
str(self._data[0]["local_date_time_full"]), "%Y%m%d%H%M%S"
)
)
return
except ValueError as err:
_LOGGER.error("Check BOM %s", err.args)
self._data = None
raise
def _get_bom_stations():
"""Return {CONF_STATION: (lat, lon)} for all stations, for auto-config.
This function does several MB of internet requests, so please use the
caching version to minimize latency and hit-count.
"""
latlon = {}
with io.BytesIO() as file_obj:
with ftplib.FTP("ftp.bom.gov.au") as ftp:
ftp.login()
ftp.cwd("anon2/home/ncc/metadata/sitelists")
ftp.retrbinary("RETR stations.zip", file_obj.write)
file_obj.seek(0)
with zipfile.ZipFile(file_obj) as zipped:
with zipped.open("stations.txt") as station_txt:
for _ in range(4):
station_txt.readline() # skip header
while True:
line = station_txt.readline().decode().strip()
if len(line) < 120:
break # end while loop, ignoring any footer text
wmo, lat, lon = (
line[a:b].strip() for a, b in [(128, 134), (70, 78), (79, 88)]
)
if wmo != "..":
latlon[wmo] = (float(lat), float(lon))
zones = {}
pattern = (
r'<a href="/products/(?P<zone>ID[A-Z]\d\d\d\d\d)/'
r'(?P=zone)\.(?P<wmo>\d\d\d\d\d).shtml">'
)
for state in ("nsw", "vic", "qld", "wa", "tas", "nt"):
url = f"http://www.bom.gov.au/{state}/observations/{state}all.shtml"
for zone_id, wmo_id in re.findall(pattern, requests.get(url).text):
zones[wmo_id] = zone_id
return {f"{zones[k]}.{k}": latlon[k] for k in set(latlon) & set(zones)}
def bom_stations(cache_dir):
"""Return {CONF_STATION: (lat, lon)} for all stations, for auto-config.
Results from internet requests are cached as compressed JSON, making
subsequent calls very much faster.
"""
cache_file = os.path.join(cache_dir, ".bom-stations.json.gz")
if not os.path.isfile(cache_file):
stations = _get_bom_stations()
with gzip.open(cache_file, "wt") as cache:
json.dump(stations, cache, sort_keys=True)
return stations
with gzip.open(cache_file, "rt") as cache:
return {k: tuple(v) for k, v in json.load(cache).items()}
def closest_station(lat, lon, cache_dir):
"""Return the ZONE_ID.WMO_ID of the closest station to our lat/lon."""
if lat is None or lon is None or not os.path.isdir(cache_dir):
return
stations = bom_stations(cache_dir)
def comparable_dist(wmo_id):
"""Create a psudeo-distance from latitude/longitude."""
station_lat, station_lon = stations[wmo_id]
return (lat - station_lat) ** 2 + (lon - station_lon) ** 2
return min(stations, key=comparable_dist)

115
homeassistant/components/bom/weather.py
View File

@ -1,115 +0,0 @@
"""Support for Australian BOM (Bureau of Meteorology) weather service."""
import logging
import voluptuous as vol
from homeassistant.components.weather import PLATFORM_SCHEMA, WeatherEntity
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, TEMP_CELSIUS
from homeassistant.helpers import config_validation as cv
# Reuse data and API logic from the sensor implementation
from .sensor import CONF_STATION, BOMCurrentData, closest_station, validate_station
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{vol.Optional(CONF_NAME): cv.string, vol.Optional(CONF_STATION): validate_station}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the BOM weather platform."""
station = config.get(CONF_STATION) or closest_station(
config.get(CONF_LATITUDE), config.get(CONF_LONGITUDE), hass.config.config_dir
)
if station is None:
_LOGGER.error("Could not get BOM weather station from lat/lon")
return False
bom_data = BOMCurrentData(station)
try:
bom_data.update()
except ValueError as err:
_LOGGER.error("Received error from BOM_Current: %s", err)
return False
add_entities([BOMWeather(bom_data, config.get(CONF_NAME))], True)
class BOMWeather(WeatherEntity):
"""Representation of a weather condition."""
def __init__(self, bom_data, stationname=None):
"""Initialise the platform with a data instance and station name."""
self.bom_data = bom_data
self.stationname = stationname or self.bom_data.latest_data.get("name")
def update(self):
"""Update current conditions."""
self.bom_data.update()
@property
def name(self):
"""Return the name of the sensor."""
return f"BOM {self.stationname or '(unknown station)'}"
@property
def condition(self):
"""Return the current condition."""
return self.bom_data.get_reading("weather") or self.bom_data.get_reading(
"cloud"
)
# Now implement the WeatherEntity interface
@property
def temperature(self):
"""Return the platform temperature."""
return self.bom_data.get_reading("air_temp")
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return TEMP_CELSIUS
@property
def pressure(self):
"""Return the mean sea-level pressure."""
return self.bom_data.get_reading("press_msl")
@property
def humidity(self):
"""Return the relative humidity."""
return self.bom_data.get_reading("rel_hum")
@property
def wind_speed(self):
"""Return the wind speed."""
return self.bom_data.get_reading("wind_spd_kmh")
@property
def wind_bearing(self):
"""Return the wind bearing."""
directions = [
"N",
"NNE",
"NE",
"ENE",
"E",
"ESE",
"SE",
"SSE",
"S",
"SSW",
"SW",
"WSW",
"W",
"WNW",
"NW",
"NNW",
]
wind = {name: idx * 360 / 16 for idx, name in enumerate(directions)}
return wind.get(self.bom_data.get_reading("wind_dir"))
@property
def attribution(self):
"""Return the attribution."""
return "Data provided by the Australian Bureau of Meteorology"

3
requirements_all.txt
View File

@ -369,9 +369,6 @@ blockchain==1.4.4
# homeassistant.components.bme680
# bme680==1.0.5
# homeassistant.components.bom
bomradarloop==0.1.5
# homeassistant.components.bond
bond-api==0.1.8

3
requirements_test_all.txt
View File

@ -197,9 +197,6 @@ blebox_uniapi==1.3.2
# homeassistant.components.blink
blinkpy==0.16.3
# homeassistant.components.bom
bomradarloop==0.1.5
# homeassistant.components.bond
bond-api==0.1.8

1
tests/components/bom/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the bom component."""

104
tests/components/bom/test_sensor.py
View File

@ -1,104 +0,0 @@
"""The tests for the BOM Weather sensor platform."""
import json
import re
import unittest
from urllib.parse import urlparse
import requests
from homeassistant.components import sensor
from homeassistant.components.bom.sensor import BOMCurrentData
from homeassistant.setup import setup_component
from tests.async_mock import patch
from tests.common import assert_setup_component, get_test_home_assistant, load_fixture
VALID_CONFIG = {
"platform": "bom",
"station": "IDN60901.94767",
"name": "Fake",
"monitored_conditions": ["apparent_t", "press", "weather"],
}
def mocked_requests(*args, **kwargs):
"""Mock requests.get invocations."""
class MockResponse:
"""Class to represent a mocked response."""
def __init__(self, json_data, status_code):
"""Initialize the mock response class."""
self.json_data = json_data
self.status_code = status_code
def json(self):
"""Return the json of the response."""
return self.json_data
@property
def content(self):
"""Return the content of the response."""
return self.json()
def raise_for_status(self):
"""Raise an HTTPError if status is not 200."""
if self.status_code != 200:
raise requests.HTTPError(self.status_code)
url = urlparse(args[0])
if re.match(r"^/fwo/[\w]+/[\w.]+\.json", url.path):
return MockResponse(json.loads(load_fixture("bom_weather.json")), 200)
raise NotImplementedError(f"Unknown route {url.path}")
class TestBOMWeatherSensor(unittest.TestCase):
"""Test the BOM Weather sensor."""
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = VALID_CONFIG
self.addCleanup(self.hass.stop)
@patch("requests.get", side_effect=mocked_requests)
def test_setup(self, mock_get):
"""Test the setup with custom settings."""
with assert_setup_component(1, sensor.DOMAIN):
assert setup_component(self.hass, sensor.DOMAIN, {"sensor": VALID_CONFIG})
self.hass.block_till_done()
fake_entities = [
"bom_fake_feels_like_c",
"bom_fake_pressure_mb",
"bom_fake_weather",
]
for entity_id in fake_entities:
state = self.hass.states.get(f"sensor.{entity_id}")
assert state is not None
@patch("requests.get", side_effect=mocked_requests)
def test_sensor_values(self, mock_get):
"""Test retrieval of sensor values."""
assert setup_component(self.hass, sensor.DOMAIN, {"sensor": VALID_CONFIG})
self.hass.block_till_done()
weather = self.hass.states.get("sensor.bom_fake_weather").state
assert "Fine" == weather
pressure = self.hass.states.get("sensor.bom_fake_pressure_mb").state
assert "1021.7" == pressure
feels_like = self.hass.states.get("sensor.bom_fake_feels_like_c").state
assert "25.0" == feels_like
class TestBOMCurrentData(unittest.TestCase):
"""Test the BOM data container."""
def test_should_update_initial(self):
"""Test that the first update always occurs."""
bom_data = BOMCurrentData("IDN60901.94767")
assert bom_data.should_update() is True