Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove TekSavvy Component, no longer available (#40762) 

The TekSavvy API has been removed.
Per [this forum post](https://community.teksavvy.com/discussion/comment/120) on TekSavvy's
support page.
pull/40789/head
Mike O'Driscoll 2020-09-30 07:14:41 -04:00 committed by GitHub
parent d9ba32dc3f
commit fc0344399b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 0 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

6
homeassistant/components/teksavvy/manifest.json
View File

@ -1,6 +0,0 @@
{
"domain": "teksavvy",
"name": "TekSavvy",
"documentation": "https://www.home-assistant.io/integrations/teksavvy",
"codeowners": []
}

173
homeassistant/components/teksavvy/sensor.py
View File

@ -1,173 +0,0 @@
"""Support for TekSavvy Bandwidth Monitor."""
from datetime import timedelta
import logging
import async_timeout
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_API_KEY,
CONF_MONITORED_VARIABLES,
CONF_NAME,
DATA_GIGABYTES,
HTTP_OK,
PERCENTAGE,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "TekSavvy"
CONF_TOTAL_BANDWIDTH = "total_bandwidth"
MIN_TIME_BETWEEN_UPDATES = timedelta(hours=1)
REQUEST_TIMEOUT = 5 # seconds
SENSOR_TYPES = {
"usage": ["Usage Ratio", PERCENTAGE, "mdi:percent"],
"usage_gb": ["Usage", DATA_GIGABYTES, "mdi:download"],
"limit": ["Data limit", DATA_GIGABYTES, "mdi:download"],
"onpeak_download": ["On Peak Download", DATA_GIGABYTES, "mdi:download"],
"onpeak_upload": ["On Peak Upload", DATA_GIGABYTES, "mdi:upload"],
"onpeak_total": ["On Peak Total", DATA_GIGABYTES, "mdi:download"],
"offpeak_download": ["Off Peak download", DATA_GIGABYTES, "mdi:download"],
"offpeak_upload": ["Off Peak Upload", DATA_GIGABYTES, "mdi:upload"],
"offpeak_total": ["Off Peak Total", DATA_GIGABYTES, "mdi:download"],
"onpeak_remaining": ["Remaining", DATA_GIGABYTES, "mdi:download"],
}
API_HA_MAP = (
("OnPeakDownload", "onpeak_download"),
("OnPeakUpload", "onpeak_upload"),
("OffPeakDownload", "offpeak_download"),
("OffPeakUpload", "offpeak_upload"),
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_MONITORED_VARIABLES): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]
),
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_TOTAL_BANDWIDTH): cv.positive_int,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
}
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the sensor platform."""
websession = async_get_clientsession(hass)
apikey = config.get(CONF_API_KEY)
bandwidthcap = config.get(CONF_TOTAL_BANDWIDTH)
ts_data = TekSavvyData(hass.loop, websession, apikey, bandwidthcap)
ret = await ts_data.async_update()
if ret is False:
_LOGGER.error("Invalid Teksavvy API key: %s", apikey)
return
name = config.get(CONF_NAME)
sensors = []
for variable in config[CONF_MONITORED_VARIABLES]:
sensors.append(TekSavvySensor(ts_data, variable, name))
async_add_entities(sensors, True)
class TekSavvySensor(Entity):
"""Representation of TekSavvy Bandwidth sensor."""
def __init__(self, teksavvydata, sensor_type, name):
"""Initialize the sensor."""
self.client_name = name
self.type = sensor_type
self._name = SENSOR_TYPES[sensor_type][0]
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
self._icon = SENSOR_TYPES[sensor_type][2]
self.teksavvydata = teksavvydata
self._state = None
@property
def name(self):
"""Return the name of the sensor."""
return f"{self.client_name} {self._name}"
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
@property
def icon(self):
"""Icon to use in the frontend, if any."""
return self._icon
async def async_update(self):
"""Get the latest data from TekSavvy and update the state."""
await self.teksavvydata.async_update()
if self.type in self.teksavvydata.data:
self._state = round(self.teksavvydata.data[self.type], 2)
class TekSavvyData:
"""Get data from TekSavvy API."""
def __init__(self, loop, websession, api_key, bandwidth_cap):
"""Initialize the data object."""
self.loop = loop
self.websession = websession
self.api_key = api_key
self.bandwidth_cap = bandwidth_cap
# Set unlimited users to infinite, otherwise the cap.
self.data = (
{"limit": self.bandwidth_cap}
if self.bandwidth_cap > 0
else {"limit": float("inf")}
)
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Get the TekSavvy bandwidth data from the web service."""
headers = {"TekSavvy-APIKey": self.api_key}
_LOGGER.debug("Updating TekSavvy data")
url = (
"https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?$filter=IsCurrent%20eq%20true"
)
with async_timeout.timeout(REQUEST_TIMEOUT):
req = await self.websession.get(url, headers=headers)
if req.status != HTTP_OK:
_LOGGER.error("Request failed with status: %u", req.status)
return False
try:
data = await req.json()
for (api, ha_name) in API_HA_MAP:
self.data[ha_name] = float(data["value"][0][api])
on_peak_download = self.data["onpeak_download"]
on_peak_upload = self.data["onpeak_upload"]
off_peak_download = self.data["offpeak_download"]
off_peak_upload = self.data["offpeak_upload"]
limit = self.data["limit"]
# Support "unlimited" users
if self.bandwidth_cap > 0:
self.data["usage"] = 100 * on_peak_download / self.bandwidth_cap
else:
self.data["usage"] = 0
self.data["usage_gb"] = on_peak_download
self.data["onpeak_total"] = on_peak_download + on_peak_upload
self.data["offpeak_total"] = off_peak_download + off_peak_upload
self.data["onpeak_remaining"] = limit - on_peak_download
return True
except ValueError:
_LOGGER.error("JSON Decode Failed")
return False

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

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

204
tests/components/teksavvy/test_sensor.py
View File

@ -1,204 +0,0 @@
"""Tests for the TekSavvy sensor platform."""
from homeassistant.bootstrap import async_setup_component
from homeassistant.components.teksavvy.sensor import TekSavvyData
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
DATA_GIGABYTES,
HTTP_NOT_FOUND,
PERCENTAGE,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
async def test_capped_setup(hass, aioclient_mock):
"""Test the default setup."""
config = {
"platform": "teksavvy",
"api_key": "NOTAKEY",
"total_bandwidth": 400,
"monitored_variables": [
"usage",
"usage_gb",
"limit",
"onpeak_download",
"onpeak_upload",
"onpeak_total",
"offpeak_download",
"offpeak_upload",
"offpeak_total",
"onpeak_remaining",
],
}
result = (
'{"odata.metadata":"http://api.teksavvy.com/web/Usage/$metadata'
'#UsageSummaryRecords","value":[{'
'"StartDate":"2018-01-01T00:00:00",'
'"EndDate":"2018-01-31T00:00:00",'
'"OID":"999999","IsCurrent":true,'
'"OnPeakDownload":226.75,'
'"OnPeakUpload":8.82,'
'"OffPeakDownload":36.24,"OffPeakUpload":1.58'
"}]}"
)
aioclient_mock.get(
"https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text=result,
)
await async_setup_component(hass, "sensor", {"sensor": config})
await hass.async_block_till_done()
state = hass.states.get("sensor.teksavvy_data_limit")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "400"
state = hass.states.get("sensor.teksavvy_off_peak_download")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "36.24"
state = hass.states.get("sensor.teksavvy_off_peak_upload")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "1.58"
state = hass.states.get("sensor.teksavvy_off_peak_total")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "37.82"
state = hass.states.get("sensor.teksavvy_on_peak_download")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "226.75"
state = hass.states.get("sensor.teksavvy_on_peak_upload")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "8.82"
state = hass.states.get("sensor.teksavvy_on_peak_total")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "235.57"
state = hass.states.get("sensor.teksavvy_usage_ratio")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == PERCENTAGE
assert state.state == "56.69"
state = hass.states.get("sensor.teksavvy_usage")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "226.75"
state = hass.states.get("sensor.teksavvy_remaining")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "173.25"
async def test_unlimited_setup(hass, aioclient_mock):
"""Test the default setup."""
config = {
"platform": "teksavvy",
"api_key": "NOTAKEY",
"total_bandwidth": 0,
"monitored_variables": [
"usage",
"usage_gb",
"limit",
"onpeak_download",
"onpeak_upload",
"onpeak_total",
"offpeak_download",
"offpeak_upload",
"offpeak_total",
"onpeak_remaining",
],
}
result = (
'{"odata.metadata":"http://api.teksavvy.com/web/Usage/$metadata'
'#UsageSummaryRecords","value":[{'
'"StartDate":"2018-01-01T00:00:00",'
'"EndDate":"2018-01-31T00:00:00",'
'"OID":"999999","IsCurrent":true,'
'"OnPeakDownload":226.75,'
'"OnPeakUpload":8.82,'
'"OffPeakDownload":36.24,"OffPeakUpload":1.58'
"}]}"
)
aioclient_mock.get(
"https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text=result,
)
await async_setup_component(hass, "sensor", {"sensor": config})
await hass.async_block_till_done()
state = hass.states.get("sensor.teksavvy_data_limit")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "inf"
state = hass.states.get("sensor.teksavvy_off_peak_download")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "36.24"
state = hass.states.get("sensor.teksavvy_off_peak_upload")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "1.58"
state = hass.states.get("sensor.teksavvy_off_peak_total")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "37.82"
state = hass.states.get("sensor.teksavvy_on_peak_download")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "226.75"
state = hass.states.get("sensor.teksavvy_on_peak_upload")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "8.82"
state = hass.states.get("sensor.teksavvy_on_peak_total")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "235.57"
state = hass.states.get("sensor.teksavvy_usage")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "226.75"
state = hass.states.get("sensor.teksavvy_usage_ratio")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == PERCENTAGE
assert state.state == "0"
state = hass.states.get("sensor.teksavvy_remaining")
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == DATA_GIGABYTES
assert state.state == "inf"
async def test_bad_return_code(hass, aioclient_mock):
"""Test handling a return code that isn't HTTP OK."""
aioclient_mock.get(
"https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
status=HTTP_NOT_FOUND,
)
tsd = TekSavvyData(hass.loop, async_get_clientsession(hass), "notakey", 400)
result = await tsd.async_update()
assert result is False
async def test_bad_json_decode(hass, aioclient_mock):
"""Test decoding invalid json result."""
aioclient_mock.get(
"https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text="this is not json",
)
tsd = TekSavvyData(hass.loop, async_get_clientsession(hass), "notakey", 400)
result = await tsd.async_update()
assert result is False