Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add new nextbus sensor #20197

Merged
merged 4 commits into from Apr 27, 2019
Merged
Changes from 2 commits
Commits
File filter...
Filter file types
Jump to…
Jump to file or symbol
Failed to load files and symbols.

Always

Just for now

@@ -151,6 +151,7 @@ homeassistant/components/nello/* @pschmitt
homeassistant/components/ness_alarm/* @nickw444
homeassistant/components/nest/* @awarecan
homeassistant/components/netdata/* @fabaff
homeassistant/components/nextbus/* @vividboarder
homeassistant/components/nissan_leaf/* @filcole
homeassistant/components/nmbs/* @thibmaek
homeassistant/components/no_ip/* @fabaff
@@ -0,0 +1 @@
"""NextBus sensor."""
@@ -0,0 +1,8 @@
{
"domain": "nextbus",
"name": "NextBus",
"documentation": "https://www.home-assistant.io/components/nextbus",
"dependencies": [],
"codeowners": ["@vividboarder"],
"requirements": ["py_nextbus==0.1.2"]
}
@@ -0,0 +1,236 @@
"""NextBus sensor."""
import logging

import voluptuous as vol

import homeassistant.helpers.config_validation as cv
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_NAME
from homeassistant.const import DEVICE_CLASS_TIMESTAMP
from homeassistant.helpers.entity import Entity
from homeassistant.util.dt import utc_from_timestamp

_LOGGER = logging.getLogger(__name__)

DOMAIN = 'nextbus'

CONF_AGENCY = 'agency'
CONF_ROUTE = 'route'
CONF_STOP = 'stop'

ICON = 'mdi:bus'

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_AGENCY): cv.string,
vol.Required(CONF_ROUTE): cv.string,
vol.Required(CONF_STOP): cv.string,
vol.Optional(CONF_NAME): cv.string,
})


def validate_value(value_name, value, value_list):
"""Validate tag value is in the list of items and logs error if not."""
valid_values = {
v['tag']: v['title']
for v in value_list
}
if value not in valid_values:
_LOGGER.error(
'Invalid %s tag `%s`. Please use one of the following: %s',
value_name,
value,
', '.join(
'{}: {}'.format(title, tag)
for tag, title in valid_values.items()
)
)
return False

return True


def validate_tags(client, agency, route, stop):
"""Validate provided tags."""
# Validate agencies
if not validate_value(
'agency',
agency,
client.get_agency_list()['agency'],
):
return False

# Validate the route
if not validate_value(
'route',
route,
client.get_route_list(agency)['route'],
):
return False

# Validate the stop
route_config = client.get_route_config(route, agency)['route']
if not validate_value(
'stop',
stop,
route_config['stop'],
):
return False

return True


def setup_platform(hass, config, add_entities, discovery_info=None):
"""Load values from configuration and initialize the platform."""
agency = config[CONF_AGENCY]
route = config[CONF_ROUTE]
stop = config[CONF_STOP]
name = config.get(CONF_NAME)

from py_nextbus import NextBusClient
client = NextBusClient(output_format='json')

# Ensures that the tags provided are valid, also logs out valid values
if not validate_tags(client, agency, route, stop):
_LOGGER.error('Invalid config value(s)')
return

add_entities([
NextBusDepartureSensor(
agency,
route,
stop,
name,
client=client,
),
], True)


class NextBusDepartureSensor(Entity):
"""Sensor class that displays upcoming NextBus times.
To function, this requires knowing the agency tag as well as the tags for
both the route and the stop.
This is possibly a little convoluted to provide as it requires making a
request to the service to get these values. Perhaps it can be simplifed in
the future using fuzzy logic and matching.
"""

def __init__(self, agency, route, stop, name=None, client=None):
This conversation was marked as resolved by ViViDboarder

This comment has been minimized.

Copy link
@MartinHjelmare

MartinHjelmare Apr 24, 2019

Member

client shouldn't be a keyword argument. Make it a positional argument. It will never be None.

"""Initialize sensor with all required config."""
self.agency = agency
self.route = route
self.stop = stop
self._custom_name = name
# Maybe pull a more user friendly name from the API here
self._name = '{} {}'.format(agency, route)
self._client = client

# set up default state attributes
self._state = None
self._attributes = {}

def _log_debug(self, message, *args):
"""Log debug message with prefix."""
_LOGGER.debug(':'.join((
self.agency,
self.route,
self.stop,
message,
)), *args)

@property
def name(self):
"""Return sensor name.
Uses an auto generated name based on the data from the API unless a
custom name is provided in the configuration.
"""
if self._custom_name:
return self._custom_name

return self._name

@property
def device_class(self):
"""Return the device class."""
return DEVICE_CLASS_TIMESTAMP

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

@property
def device_state_attributes(self):
"""Return additional state attributes."""
return self._attributes

@property
def icon(self):
"""Return icon to be used for this sensor."""
# Would be nice if we could determine if the line is a train or bus
# however that doesn't seem to be available to us. Using bus for now.
return ICON

def update(self):
"""Update sensor with new departures times."""
# Note: using Multi because there is a bug with the single stop impl
predictions = self._client.get_predictions_for_multi_stops(
[{
'stop_tag': int(self.stop),
'route_tag': self.route,
}],
self.agency,
)

self._log_debug('Predictions results: %s', predictions)

if 'Error' in predictions:
self._log_debug('Could not get predictions: %s', predictions)

if not predictions.get('predictions'):
self._log_debug('No predictions available')
self._state = None
# Remove attributes that may now be outdated
self._attributes.pop('upcoming', None)
return

predictions = predictions['predictions']

# Set detailed attributes
self._attributes.update({
'agency': predictions.get('agencyTitle'),
'route': predictions.get('routeTitle'),
'direction': predictions.get('direction', {}).get('title'),
'stop': predictions.get('stopTitle'),
})

# Sometimes message will return a dict, others a list of dicts
message = predictions.get('message')
if isinstance(message, dict):
self._attributes['message'] = message.get('text', '')
elif isinstance(message, list):
self._attributes['message'] = ' -- '.join(
m.get('text', '')
for m in message
)
else:
self._attributes.pop('message', None)

upcoming = predictions.get('direction', {}).get('prediction')
if not upcoming:
self._log_debug('No upcoming predictions available')
self._state = None
self._attributes['upcoming'] = 'No upcoming predictions'
return

self._attributes['upcoming'] = ', '.join(
p['minutes'] for p in upcoming
)

latest_prediction = upcoming[0]

self._state = utc_from_timestamp(
int(latest_prediction['epochTime']) / 1000
).isoformat()
@@ -939,6 +939,9 @@ pyW215==0.6.0
# homeassistant.components.w800rf32
pyW800rf32==0.1

# homeassistant.components.nextbus
py_nextbus==0.1.2

# homeassistant.components.noaa_tides
# py_noaa==0.3.0

@@ -0,0 +1 @@
"""The tests for the nexbus component."""
ProTip! Use n and p to navigate between commits in a pull request.
You can’t perform that action at this time.