diff --git a/CODEOWNERS b/CODEOWNERS
index 700d68b9449..38de5b1fe6f 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -153,6 +153,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
diff --git a/homeassistant/components/nextbus/__init__.py b/homeassistant/components/nextbus/__init__.py
new file mode 100644
index 00000000000..4891af77b28
--- /dev/null
+++ b/homeassistant/components/nextbus/__init__.py
@@ -0,0 +1 @@
+"""NextBus sensor."""
diff --git a/homeassistant/components/nextbus/manifest.json b/homeassistant/components/nextbus/manifest.json
new file mode 100644
index 00000000000..63bdbf8a928
--- /dev/null
+++ b/homeassistant/components/nextbus/manifest.json
@@ -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"]
+}
diff --git a/homeassistant/components/nextbus/sensor.py b/homeassistant/components/nextbus/sensor.py
new file mode 100644
index 00000000000..acf8028e31f
--- /dev/null
+++ b/homeassistant/components/nextbus/sensor.py
@@ -0,0 +1,268 @@
+"""NextBus sensor."""
+import logging
+from itertools import chain
+
+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 listify(maybe_list):
+    """Return list version of whatever value is passed in.
+
+    This is used to provide a consistent way of interacting with the JSON
+    results from the API. There are several attributes that will either missing
+    if there are no values, a single dictionary if there is only one value, and
+    a list if there are multiple.
+    """
+    if maybe_list is None:
+        return []
+    if isinstance(maybe_list, list):
+        return maybe_list
+    return [maybe_list]
+
+
+def maybe_first(maybe_list):
+    """Return the first item out of a list or returns back the input."""
+    if isinstance(maybe_list, list) and maybe_list:
+        return maybe_list[0]
+
+    return maybe_list
+
+
+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(
+            client,
+            agency,
+            route,
+            stop,
+            name,
+        ),
+    ], 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, client, agency, route, stop, name=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
+        results = self._client.get_predictions_for_multi_stops(
+            [{
+                'stop_tag': int(self.stop),
+                'route_tag': self.route,
+            }],
+            self.agency,
+        )
+
+        self._log_debug('Predictions results: %s', results)
+
+        if 'Error' in results:
+            self._log_debug('Could not get predictions: %s', results)
+
+        if not results.get('predictions'):
+            self._log_debug('No predictions available')
+            self._state = None
+            # Remove attributes that may now be outdated
+            self._attributes.pop('upcoming', None)
+            return
+
+        results = results['predictions']
+
+        # Set detailed attributes
+        self._attributes.update({
+            'agency': results.get('agencyTitle'),
+            'route': results.get('routeTitle'),
+            'stop': results.get('stopTitle'),
+        })
+
+        # List all messages in the attributes
+        messages = listify(results.get('message', []))
+        self._log_debug('Messages: %s', messages)
+        self._attributes['message'] = ' -- '.join((
+            message.get('text', '')
+            for message in messages
+        ))
+
+        # List out all directions in the attributes
+        directions = listify(results.get('direction', []))
+        self._attributes['direction'] = ', '.join((
+            direction.get('title', '')
+            for direction in directions
+        ))
+
+        # Chain all predictions together
+        predictions = list(chain(*[
+            listify(direction.get('prediction', []))
+            for direction in directions
+        ]))
+
+        # Short circuit if we don't have any actual bus predictions
+        if not predictions:
+            self._log_debug('No upcoming predictions available')
+            self._state = None
+            self._attributes['upcoming'] = 'No upcoming predictions'
+            return
+
+        # Generate list of upcoming times
+        self._attributes['upcoming'] = ', '.join(
+            p['minutes'] for p in predictions
+        )
+
+        latest_prediction = maybe_first(predictions)
+        self._state = utc_from_timestamp(
+            int(latest_prediction['epochTime']) / 1000
+        ).isoformat()
diff --git a/requirements_all.txt b/requirements_all.txt
index d841baa54f9..1540a936b5b 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -936,6 +936,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
 
diff --git a/tests/components/nextbus/__init__.py b/tests/components/nextbus/__init__.py
new file mode 100644
index 00000000000..609e0bb574b
--- /dev/null
+++ b/tests/components/nextbus/__init__.py
@@ -0,0 +1 @@
+"""The tests for the nexbus component."""
diff --git a/tests/components/nextbus/test_sensor.py b/tests/components/nextbus/test_sensor.py
new file mode 100644
index 00000000000..ece2a1d8092
--- /dev/null
+++ b/tests/components/nextbus/test_sensor.py
@@ -0,0 +1,329 @@
+"""The tests for the nexbus sensor component."""
+from copy import deepcopy
+
+import pytest
+
+import homeassistant.components.sensor as sensor
+import homeassistant.components.nextbus.sensor as nextbus
+
+from tests.common import (assert_setup_component,
+                          async_setup_component,
+                          MockDependency)
+
+
+VALID_AGENCY = 'sf-muni'
+VALID_ROUTE = 'F'
+VALID_STOP = '5650'
+VALID_AGENCY_TITLE = 'San Francisco Muni'
+VALID_ROUTE_TITLE = 'F-Market & Wharves'
+VALID_STOP_TITLE = 'Market St & 7th St'
+SENSOR_ID_SHORT = 'sensor.sf_muni_f'
+
+CONFIG_BASIC = {
+    'sensor': {
+        'platform': 'nextbus',
+        'agency': VALID_AGENCY,
+        'route': VALID_ROUTE,
+        'stop': VALID_STOP,
+    }
+}
+
+CONFIG_INVALID_MISSING = {
+    'sensor': {
+        'platform': 'nextbus',
+    }
+}
+
+BASIC_RESULTS = {
+    'predictions': {
+        'agencyTitle': VALID_AGENCY_TITLE,
+        'routeTitle': VALID_ROUTE_TITLE,
+        'stopTitle': VALID_STOP_TITLE,
+        'direction': {
+            'title': 'Outbound',
+            'prediction': [
+                {'minutes': '1', 'epochTime': '1553807371000'},
+                {'minutes': '2', 'epochTime': '1553807372000'},
+                {'minutes': '3', 'epochTime': '1553807373000'},
+            ],
+        }
+    }
+}
+
+
+async def assert_setup_sensor(hass, config, count=1):
+    """Set up the sensor and assert it's been created."""
+    with assert_setup_component(count):
+        assert await async_setup_component(hass, sensor.DOMAIN, config)
+
+
+@pytest.fixture
+def mock_nextbus():
+    """Create a mock py_nextbus module."""
+    with MockDependency('py_nextbus') as py_nextbus:
+        yield py_nextbus
+
+
+@pytest.fixture
+def mock_nextbus_predictions(mock_nextbus):
+    """Create a mock of NextBusClient predictions."""
+    instance = mock_nextbus.NextBusClient.return_value
+    instance.get_predictions_for_multi_stops.return_value = BASIC_RESULTS
+
+    yield instance.get_predictions_for_multi_stops
+
+
+@pytest.fixture
+def mock_nextbus_lists(mock_nextbus):
+    """Mock all list functions in nextbus to test validate logic."""
+    instance = mock_nextbus.NextBusClient.return_value
+    instance.get_agency_list.return_value = {
+        'agency': [
+            {'tag': 'sf-muni', 'title': 'San Francisco Muni'},
+        ]
+    }
+    instance.get_route_list.return_value = {
+        'route': [
+            {'tag': 'F', 'title': 'F - Market & Wharves'},
+        ]
+    }
+    instance.get_route_config.return_value = {
+        'route': {
+            'stop': [
+                {'tag': '5650', 'title': 'Market St & 7th St'},
+            ]
+        }
+    }
+
+
+async def test_valid_config(hass, mock_nextbus, mock_nextbus_lists):
+    """Test that sensor is set up properly with valid config."""
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+
+async def test_invalid_config(hass, mock_nextbus, mock_nextbus_lists):
+    """Checks that component is not setup when missing information."""
+    await assert_setup_sensor(hass, CONFIG_INVALID_MISSING, count=0)
+
+
+async def test_validate_tags(hass, mock_nextbus, mock_nextbus_lists):
+    """Test that additional validation against the API is successful."""
+    client = mock_nextbus.NextBusClient()
+    # with self.subTest('Valid everything'):
+    assert nextbus.validate_tags(
+        client,
+        VALID_AGENCY,
+        VALID_ROUTE,
+        VALID_STOP,
+    )
+    # with self.subTest('Invalid agency'):
+    assert not nextbus.validate_tags(
+        client,
+        'not-valid',
+        VALID_ROUTE,
+        VALID_STOP,
+    )
+
+    # with self.subTest('Invalid route'):
+    assert not nextbus.validate_tags(
+        client,
+        VALID_AGENCY,
+        '0',
+        VALID_STOP,
+    )
+
+    # with self.subTest('Invalid stop'):
+    assert not nextbus.validate_tags(
+        client,
+        VALID_AGENCY,
+        VALID_ROUTE,
+        0,
+    )
+
+
+async def test_verify_valid_state(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify all attributes are set from a valid response."""
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+    mock_nextbus_predictions.assert_called_once_with(
+        [{'stop_tag': int(VALID_STOP), 'route_tag': VALID_ROUTE}],
+        VALID_AGENCY,
+    )
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.state == '2019-03-28T21:09:31+00:00'
+    assert state.attributes['agency'] == VALID_AGENCY_TITLE
+    assert state.attributes['route'] == VALID_ROUTE_TITLE
+    assert state.attributes['stop'] == VALID_STOP_TITLE
+    assert state.attributes['direction'] == 'Outbound'
+    assert state.attributes['upcoming'] == '1, 2, 3'
+
+
+async def test_message_dict(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify that a single dict message is rendered correctly."""
+    mock_nextbus_predictions.return_value = {
+        'predictions': {
+            'agencyTitle': VALID_AGENCY_TITLE,
+            'routeTitle': VALID_ROUTE_TITLE,
+            'stopTitle': VALID_STOP_TITLE,
+            'message': {'text': 'Message'},
+            'direction': {
+                'title': 'Outbound',
+                'prediction': [
+                    {'minutes': '1', 'epochTime': '1553807371000'},
+                    {'minutes': '2', 'epochTime': '1553807372000'},
+                    {'minutes': '3', 'epochTime': '1553807373000'},
+                ],
+            }
+        }
+    }
+
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.attributes['message'] == 'Message'
+
+
+async def test_message_list(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify that a list of messages are rendered correctly."""
+    mock_nextbus_predictions.return_value = {
+        'predictions': {
+            'agencyTitle': VALID_AGENCY_TITLE,
+            'routeTitle': VALID_ROUTE_TITLE,
+            'stopTitle': VALID_STOP_TITLE,
+            'message': [{'text': 'Message 1'}, {'text': 'Message 2'}],
+            'direction': {
+                'title': 'Outbound',
+                'prediction': [
+                    {'minutes': '1', 'epochTime': '1553807371000'},
+                    {'minutes': '2', 'epochTime': '1553807372000'},
+                    {'minutes': '3', 'epochTime': '1553807373000'},
+                ],
+            }
+        }
+    }
+
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.attributes['message'] == 'Message 1 -- Message 2'
+
+
+async def test_direction_list(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify that a list of messages are rendered correctly."""
+    mock_nextbus_predictions.return_value = {
+        'predictions': {
+            'agencyTitle': VALID_AGENCY_TITLE,
+            'routeTitle': VALID_ROUTE_TITLE,
+            'stopTitle': VALID_STOP_TITLE,
+            'message': [{'text': 'Message 1'}, {'text': 'Message 2'}],
+            'direction': [
+                {
+                    'title': 'Outbound',
+                    'prediction': [
+                        {'minutes': '1', 'epochTime': '1553807371000'},
+                        {'minutes': '2', 'epochTime': '1553807372000'},
+                        {'minutes': '3', 'epochTime': '1553807373000'},
+                    ],
+                },
+                {
+                    'title': 'Outbound 2',
+                    'prediction': {
+                        'minutes': '4',
+                        'epochTime': '1553807374000',
+                    },
+                },
+            ],
+        }
+    }
+
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.state == '2019-03-28T21:09:31+00:00'
+    assert state.attributes['agency'] == VALID_AGENCY_TITLE
+    assert state.attributes['route'] == VALID_ROUTE_TITLE
+    assert state.attributes['stop'] == VALID_STOP_TITLE
+    assert state.attributes['direction'] == 'Outbound, Outbound 2'
+    assert state.attributes['upcoming'] == '1, 2, 3, 4'
+
+
+async def test_custom_name(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify that a custom name can be set via config."""
+    config = deepcopy(CONFIG_BASIC)
+    config['sensor']['name'] = 'Custom Name'
+
+    await assert_setup_sensor(hass, config)
+    state = hass.states.get('sensor.custom_name')
+    assert state is not None
+
+
+async def test_no_predictions(
+        hass,
+        mock_nextbus,
+        mock_nextbus_predictions,
+        mock_nextbus_lists,
+):
+    """Verify there are no exceptions when no predictions are returned."""
+    mock_nextbus_predictions.return_value = {}
+
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.state == 'unknown'
+
+
+async def test_verify_no_upcoming(
+        hass,
+        mock_nextbus,
+        mock_nextbus_lists,
+        mock_nextbus_predictions,
+):
+    """Verify attributes are set despite no upcoming times."""
+    mock_nextbus_predictions.return_value = {
+        'predictions': {
+            'agencyTitle': VALID_AGENCY_TITLE,
+            'routeTitle': VALID_ROUTE_TITLE,
+            'stopTitle': VALID_STOP_TITLE,
+            'direction': {
+                'title': 'Outbound',
+                'prediction': [],
+            }
+        }
+    }
+
+    await assert_setup_sensor(hass, CONFIG_BASIC)
+
+    state = hass.states.get(SENSOR_ID_SHORT)
+    assert state is not None
+    assert state.state == 'unknown'
+    assert state.attributes['upcoming'] == 'No upcoming predictions'