jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 12:17:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #263 from balloob/dev

Browse Source 
Update master with latest changes
This commit is contained in:
Paulus Schoutsen 2015-08-17 19:55:50 -07:00
commit dfc7f8b0c6
63 changed files with 2085 additions and 924 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.coveragerc
View File

@ -16,7 +16,11 @@ omit =
homeassistant/components/modbus.py
homeassistant/components/*/modbus.py
homeassistant/components/mqtt.py
homeassistant/components/*/tellstick.py
homeassistant/components/*/vera.py
homeassistant/components/verisure.py
homeassistant/components/*/verisure.py
homeassistant/components/wink.py
homeassistant/components/*/wink.py
@ -24,9 +28,6 @@ omit =
homeassistant/components/zwave.py
homeassistant/components/*/zwave.py
homeassistant/components/*/tellstick.py
homeassistant/components/*/vera.py
homeassistant/components/browser.py
homeassistant/components/camera/*
homeassistant/components/device_tracker/asuswrt.py

10
README.md
View File

@ -8,11 +8,11 @@ Check out [the website](https://home-assistant.io) for installation instructions
Examples of devices it can interface it:
* Monitoring connected devices to a wireless router: [OpenWrt](https://openwrt.org/), [Tomato](http://www.polarcloud.com/tomato), [Netgear](http://netgear.com), [DD-WRT](http://www.dd-wrt.com/site/index), [TPLink](http://www.tp-link.us/), [ASUSWRT](http://event.asus.com/2013/nw/ASUSWRT/)
* [Philips Hue](http://meethue.com) lights, [WeMo](http://www.belkin.com/us/Products/home-automation/c/wemo-home-automation/) switches, and [Tellstick](http://www.telldus.se/products/tellstick) devices and sensors
* [Google Chromecasts](http://www.google.com/intl/en/chrome/devices/chromecast), [Music Player Daemon](http://www.musicpd.org/) and [Kodi (XBMC)](http://kodi.tv/)
* Support for [ISY994](https://www.universal-devices.com/residential/isy994i-series/) (Insteon and X10 devices), [Z-Wave](http://www.z-wave.com/), [Nest Thermostats](https://nest.com/), and [Modbus](http://www.modbus.org/)
* Integrate data from the [Bitcoin](https://bitcoin.org) network, local meteorological data from [OpenWeatherMap](http://openweathermap.org/), [Transmission](http://www.transmissionbt.com/) or [SABnzbd](http://sabnzbd.org).
* Monitoring connected devices to a wireless router: [OpenWrt](https://openwrt.org/), [Tomato](http://www.polarcloud.com/tomato), [Netgear](http://netgear.com), [DD-WRT](http://www.dd-wrt.com/site/index), [TPLink](http://www.tp-link.us/), and [ASUSWRT](http://event.asus.com/2013/nw/ASUSWRT/)
* [Philips Hue](http://meethue.com) lights, [WeMo](http://www.belkin.com/us/Products/home-automation/c/wemo-home-automation/) switches, [Efergy](https://efergy.com) plugs, [Edimax](http://www.edimax.com/) switches, RFXtrx sensors, and [Tellstick](http://www.telldus.se/products/tellstick) devices and sensors
* [Google Chromecasts](http://www.google.com/intl/en/chrome/devices/chromecast), [Music Player Daemon](http://www.musicpd.org/), [Logitech Squeezebox](https://en.wikipedia.org/wiki/Squeezebox_%28network_music_player%29), and [Kodi (XBMC)](http://kodi.tv/)
* Support for [ISY994](https://www.universal-devices.com/residential/isy994i-series/) (Insteon and X10 devices), [Z-Wave](http://www.z-wave.com/), [Nest Thermostats](https://nest.com/), [Arduino](https://www.arduino.cc/), [Raspberry Pi](https://www.raspberrypi.org/), and [Modbus](http://www.modbus.org/)
* Integrate data from the [Bitcoin](https://bitcoin.org) network, meteorological data from [OpenWeatherMap](http://openweathermap.org/) and [Forecast.io](https://forecast.io/), [Transmission](http://www.transmissionbt.com/), or [SABnzbd](http://sabnzbd.org).
* [See full list of supported devices](https://home-assistant.io/components/)
Built home automation on top of your devices:

802
homeassistant/__init__.py
View File

@ -1,802 +0,0 @@
"""
homeassistant
~~~~~~~~~~~~~
Home Assistant is a Home Automation framework for observing the state
of entities and react to changes.
"""
import os
import time
import logging
import threading
import enum
import re
import functools as ft
from collections import namedtuple
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP,
SERVICE_HOMEASSISTANT_STOP, EVENT_TIME_CHANGED, EVENT_STATE_CHANGED,
EVENT_CALL_SERVICE, ATTR_NOW, ATTR_DOMAIN, ATTR_SERVICE, MATCH_ALL,
EVENT_SERVICE_EXECUTED, ATTR_SERVICE_CALL_ID, EVENT_SERVICE_REGISTERED,
TEMP_CELCIUS, TEMP_FAHRENHEIT, ATTR_FRIENDLY_NAME)
import homeassistant.util as util
import homeassistant.util.dt as date_util
DOMAIN = "homeassistant"
# How often time_changed event should fire
TIMER_INTERVAL = 1 # seconds
# How long we wait for the result of a service call
SERVICE_CALL_LIMIT = 10 # seconds
# Define number of MINIMUM worker threads.
# During bootstrap of HA (see bootstrap._setup_component()) worker threads
# will be added for each component that polls devices.
MIN_WORKER_THREAD = 2
# Pattern for validating entity IDs (format: <domain>.<entity>)
ENTITY_ID_PATTERN = re.compile(r"^(?P<domain>\w+)\.(?P<entity>\w+)$")
_LOGGER = logging.getLogger(__name__)
# Temporary to support deprecated methods
_MockHA = namedtuple("MockHomeAssistant", ['bus'])
class HomeAssistant(object):
""" Core class to route all communication to right components. """
def __init__(self):
self.pool = pool = create_worker_pool()
self.bus = EventBus(pool)
self.services = ServiceRegistry(self.bus, pool)
self.states = StateMachine(self.bus)
self.config = Config()
def start(self):
""" Start home assistant. """
_LOGGER.info(
"Starting Home Assistant (%d threads)", self.pool.worker_count)
create_timer(self)
self.bus.fire(EVENT_HOMEASSISTANT_START)
def block_till_stopped(self):
""" Will register service homeassistant/stop and
will block until called. """
request_shutdown = threading.Event()
def stop_homeassistant(service):
""" Stops Home Assistant. """
request_shutdown.set()
self.services.register(
DOMAIN, SERVICE_HOMEASSISTANT_STOP, stop_homeassistant)
while not request_shutdown.isSet():
try:
time.sleep(1)
except KeyboardInterrupt:
break
self.stop()
def stop(self):
""" Stops Home Assistant and shuts down all threads. """
_LOGGER.info("Stopping")
self.bus.fire(EVENT_HOMEASSISTANT_STOP)
# Wait till all responses to homeassistant_stop are done
self.pool.block_till_done()
self.pool.stop()
def track_point_in_time(self, action, point_in_time):
"""Deprecated method as of 8/4/2015 to track point in time."""
_LOGGER.warning(
'hass.track_point_in_time is deprecated. '
'Please use homeassistant.helpers.event.track_point_in_time')
import homeassistant.helpers.event as helper
helper.track_point_in_time(self, action, point_in_time)
def track_point_in_utc_time(self, action, point_in_time):
"""Deprecated method as of 8/4/2015 to track point in UTC time."""
_LOGGER.warning(
'hass.track_point_in_utc_time is deprecated. '
'Please use homeassistant.helpers.event.track_point_in_utc_time')
import homeassistant.helpers.event as helper
helper.track_point_in_utc_time(self, action, point_in_time)
def track_utc_time_change(self, action,
year=None, month=None, day=None,
hour=None, minute=None, second=None):
"""Deprecated method as of 8/4/2015 to track UTC time change."""
# pylint: disable=too-many-arguments
_LOGGER.warning(
'hass.track_utc_time_change is deprecated. '
'Please use homeassistant.helpers.event.track_utc_time_change')
import homeassistant.helpers.event as helper
helper.track_utc_time_change(self, action, year, month, day, hour,
minute, second)
def track_time_change(self, action,
year=None, month=None, day=None,
hour=None, minute=None, second=None, utc=False):
"""Deprecated method as of 8/4/2015 to track time change."""
# pylint: disable=too-many-arguments
_LOGGER.warning(
'hass.track_time_change is deprecated. '
'Please use homeassistant.helpers.event.track_time_change')
import homeassistant.helpers.event as helper
helper.track_time_change(self, action, year, month, day, hour,
minute, second)
class JobPriority(util.OrderedEnum):
""" Provides priorities for bus events. """
# pylint: disable=no-init,too-few-public-methods
EVENT_CALLBACK = 0
EVENT_SERVICE = 1
EVENT_STATE = 2
EVENT_TIME = 3
EVENT_DEFAULT = 4
@staticmethod
def from_event_type(event_type):
""" Returns a priority based on event type. """
if event_type == EVENT_TIME_CHANGED:
return JobPriority.EVENT_TIME
elif event_type == EVENT_STATE_CHANGED:
return JobPriority.EVENT_STATE
elif event_type == EVENT_CALL_SERVICE:
return JobPriority.EVENT_SERVICE
elif event_type == EVENT_SERVICE_EXECUTED:
return JobPriority.EVENT_CALLBACK
else:
return JobPriority.EVENT_DEFAULT
class EventOrigin(enum.Enum):
""" Distinguish between origin of event. """
# pylint: disable=no-init,too-few-public-methods
local = "LOCAL"
remote = "REMOTE"
def __str__(self):
return self.value
# pylint: disable=too-few-public-methods
class Event(object):
""" Represents an event within the Bus. """
__slots__ = ['event_type', 'data', 'origin', 'time_fired']
def __init__(self, event_type, data=None, origin=EventOrigin.local,
time_fired=None):
self.event_type = event_type
self.data = data or {}
self.origin = origin
self.time_fired = date_util.strip_microseconds(
time_fired or date_util.utcnow())
def as_dict(self):
""" Returns a dict representation of this Event. """
return {
'event_type': self.event_type,
'data': dict(self.data),
'origin': str(self.origin),
'time_fired': date_util.datetime_to_str(self.time_fired),
}
def __repr__(self):
# pylint: disable=maybe-no-member
if self.data:
return "<Event {}[{}]: {}>".format(
self.event_type, str(self.origin)[0],
util.repr_helper(self.data))
else:
return "<Event {}[{}]>".format(self.event_type,
str(self.origin)[0])
def __eq__(self, other):
return (self.__class__ == other.__class__ and
self.event_type == other.event_type and
self.data == other.data and
self.origin == other.origin and
self.time_fired == other.time_fired)
class EventBus(object):
""" Class that allows different components to communicate via services
and events.
"""
def __init__(self, pool=None):
self._listeners = {}
self._lock = threading.Lock()
self._pool = pool or create_worker_pool()
@property
def listeners(self):
""" Dict with events that is being listened for and the number
of listeners.
"""
with self._lock:
return {key: len(self._listeners[key])
for key in self._listeners}
def fire(self, event_type, event_data=None, origin=EventOrigin.local):
""" Fire an event. """
if not self._pool.running:
raise HomeAssistantError('Home Assistant has shut down.')
with self._lock:
# Copy the list of the current listeners because some listeners
# remove themselves as a listener while being executed which
# causes the iterator to be confused.
get = self._listeners.get
listeners = get(MATCH_ALL, []) + get(event_type, [])
event = Event(event_type, event_data, origin)
if event_type != EVENT_TIME_CHANGED:
_LOGGER.info("Bus:Handling %s", event)
if not listeners:
return
job_priority = JobPriority.from_event_type(event_type)
for func in listeners:
self._pool.add_job(job_priority, (func, event))
def listen(self, event_type, listener):
""" Listen for all events or events of a specific type.
To listen to all events specify the constant ``MATCH_ALL``
as event_type.
"""
with self._lock:
if event_type in self._listeners:
self._listeners[event_type].append(listener)
else:
self._listeners[event_type] = [listener]
def listen_once(self, event_type, listener):
""" Listen once for event of a specific type.
To listen to all events specify the constant ``MATCH_ALL``
as event_type.
Returns registered listener that can be used with remove_listener.
"""
@ft.wraps(listener)
def onetime_listener(event):
""" Removes listener from eventbus and then fires listener. """
if hasattr(onetime_listener, 'run'):
return
# Set variable so that we will never run twice.
# Because the event bus might have to wait till a thread comes
# available to execute this listener it might occur that the
# listener gets lined up twice to be executed.
# This will make sure the second time it does nothing.
onetime_listener.run = True
self.remove_listener(event_type, onetime_listener)
listener(event)
self.listen(event_type, onetime_listener)
return onetime_listener
def remove_listener(self, event_type, listener):
""" Removes a listener of a specific event_type. """
with self._lock:
try:
self._listeners[event_type].remove(listener)
# delete event_type list if empty
if not self._listeners[event_type]:
self._listeners.pop(event_type)
except (KeyError, ValueError):
# KeyError is key event_type listener did not exist
# ValueError if listener did not exist within event_type
pass
class State(object):
"""
Object to represent a state within the state machine.
entity_id: the entity that is represented.
state: the state of the entity
attributes: extra information on entity and state
last_changed: last time the state was changed, not the attributes.
last_updated: last time this object was updated.
"""
__slots__ = ['entity_id', 'state', 'attributes',
'last_changed', 'last_updated']
# pylint: disable=too-many-arguments
def __init__(self, entity_id, state, attributes=None, last_changed=None,
last_updated=None):
if not ENTITY_ID_PATTERN.match(entity_id):
raise InvalidEntityFormatError((
"Invalid entity id encountered: {}. "
"Format should be <domain>.<object_id>").format(entity_id))
self.entity_id = entity_id.lower()
self.state = state
self.attributes = attributes or {}
self.last_updated = date_util.strip_microseconds(
last_updated or date_util.utcnow())
# Strip microsecond from last_changed else we cannot guarantee
# state == State.from_dict(state.as_dict())
# This behavior occurs because to_dict uses datetime_to_str
# which does not preserve microseconds
self.last_changed = date_util.strip_microseconds(
last_changed or self.last_updated)
@property
def domain(self):
""" Returns domain of this state. """
return util.split_entity_id(self.entity_id)[0]
@property
def object_id(self):
""" Returns object_id of this state. """
return util.split_entity_id(self.entity_id)[1]
@property
def name(self):
""" Name to represent this state. """
return (
self.attributes.get(ATTR_FRIENDLY_NAME) or
self.object_id.replace('_', ' '))
def copy(self):
""" Creates a copy of itself. """
return State(self.entity_id, self.state,
dict(self.attributes), self.last_changed)
def as_dict(self):
""" Converts State to a dict to be used within JSON.
Ensures: state == State.from_dict(state.as_dict()) """
return {'entity_id': self.entity_id,
'state': self.state,
'attributes': self.attributes,
'last_changed': date_util.datetime_to_str(self.last_changed),
'last_updated': date_util.datetime_to_str(self.last_updated)}
@classmethod
def from_dict(cls, json_dict):
""" Static method to create a state from a dict.
Ensures: state == State.from_json_dict(state.to_json_dict()) """
if not (json_dict and
'entity_id' in json_dict and
'state' in json_dict):
return None
last_changed = json_dict.get('last_changed')
if last_changed:
last_changed = date_util.str_to_datetime(last_changed)
last_updated = json_dict.get('last_updated')
if last_updated:
last_updated = date_util.str_to_datetime(last_updated)
return cls(json_dict['entity_id'], json_dict['state'],
json_dict.get('attributes'), last_changed, last_updated)
def __eq__(self, other):
return (self.__class__ == other.__class__ and
self.entity_id == other.entity_id and
self.state == other.state and
self.attributes == other.attributes)
def __repr__(self):
attr = "; {}".format(util.repr_helper(self.attributes)) \
if self.attributes else ""
return "<state {}={}{} @ {}>".format(
self.entity_id, self.state, attr,
date_util.datetime_to_local_str(self.last_changed))
class StateMachine(object):
""" Helper class that tracks the state of different entities. """
def __init__(self, bus):
self._states = {}
self._bus = bus
self._lock = threading.Lock()
def entity_ids(self, domain_filter=None):
""" List of entity ids that are being tracked. """
if domain_filter is None:
return list(self._states.keys())
domain_filter = domain_filter.lower()
return [state.entity_id for key, state
in self._states.items()
if util.split_entity_id(key)[0] == domain_filter]
def all(self):
""" Returns a list of all states. """
with self._lock:
return [state.copy() for state in self._states.values()]
def get(self, entity_id):
""" Returns the state of the specified entity. """
state = self._states.get(entity_id.lower())
# Make a copy so people won't mutate the state
return state.copy() if state else None
def is_state(self, entity_id, state):
""" Returns True if entity exists and is specified state. """
entity_id = entity_id.lower()
return (entity_id in self._states and
self._states[entity_id].state == state)
def remove(self, entity_id):
""" Removes an entity from the state machine.
Returns boolean to indicate if an entity was removed. """
entity_id = entity_id.lower()
with self._lock:
return self._states.pop(entity_id, None) is not None
def set(self, entity_id, new_state, attributes=None):
""" Set the state of an entity, add entity if it does not exist.
Attributes is an optional dict to specify attributes of this state.
If you just update the attributes and not the state, last changed will
not be affected.
"""
entity_id = entity_id.lower()
new_state = str(new_state)
attributes = attributes or {}
with self._lock:
old_state = self._states.get(entity_id)
is_existing = old_state is not None
same_state = is_existing and old_state.state == new_state
same_attr = is_existing and old_state.attributes == attributes
if same_state and same_attr:
return
# If state did not exist or is different, set it
last_changed = old_state.last_changed if same_state else None
state = State(entity_id, new_state, attributes, last_changed)
self._states[entity_id] = state
event_data = {'entity_id': entity_id, 'new_state': state}
if old_state:
event_data['old_state'] = old_state
self._bus.fire(EVENT_STATE_CHANGED, event_data)
def track_change(self, entity_ids, action, from_state=None, to_state=None):
"""
DEPRECATED AS OF 8/4/2015
"""
_LOGGER.warning(
'hass.states.track_change is deprecated. '
'Use homeassistant.helpers.event.track_state_change instead.')
import homeassistant.helpers.event as helper
helper.track_state_change(_MockHA(self._bus), entity_ids, action,
from_state, to_state)
# pylint: disable=too-few-public-methods
class ServiceCall(object):
""" Represents a call to a service. """
__slots__ = ['domain', 'service', 'data']
def __init__(self, domain, service, data=None):
self.domain = domain
self.service = service
self.data = data or {}
def __repr__(self):
if self.data:
return "<ServiceCall {}.{}: {}>".format(
self.domain, self.service, util.repr_helper(self.data))
else:
return "<ServiceCall {}.{}>".format(self.domain, self.service)
class ServiceRegistry(object):
""" Offers services over the eventbus. """
def __init__(self, bus, pool=None):
self._services = {}
self._lock = threading.Lock()
self._pool = pool or create_worker_pool()
self._bus = bus
self._cur_id = 0
bus.listen(EVENT_CALL_SERVICE, self._event_to_service_call)
@property
def services(self):
""" Dict with per domain a list of available services. """
with self._lock:
return {domain: list(self._services[domain].keys())
for domain in self._services}
def has_service(self, domain, service):
""" Returns True if specified service exists. """
return service in self._services.get(domain, [])
def register(self, domain, service, service_func):
""" Register a service. """
with self._lock:
if domain in self._services:
self._services[domain][service] = service_func
else:
self._services[domain] = {service: service_func}
self._bus.fire(
EVENT_SERVICE_REGISTERED,
{ATTR_DOMAIN: domain, ATTR_SERVICE: service})
def call(self, domain, service, service_data=None, blocking=False):
"""
Calls specified service.
Specify blocking=True to wait till service is executed.
Waits a maximum of SERVICE_CALL_LIMIT.
If blocking = True, will return boolean if service executed
succesfully within SERVICE_CALL_LIMIT.
This method will fire an event to call the service.
This event will be picked up by this ServiceRegistry and any
other ServiceRegistry that is listening on the EventBus.
Because the service is sent as an event you are not allowed to use
the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.
"""
call_id = self._generate_unique_id()
event_data = service_data or {}
event_data[ATTR_DOMAIN] = domain
event_data[ATTR_SERVICE] = service
event_data[ATTR_SERVICE_CALL_ID] = call_id
if blocking:
executed_event = threading.Event()
def service_executed(call):
"""
Called when a service is executed.
Will set the event if matches our service call.
"""
if call.data[ATTR_SERVICE_CALL_ID] == call_id:
executed_event.set()
self._bus.listen(EVENT_SERVICE_EXECUTED, service_executed)
self._bus.fire(EVENT_CALL_SERVICE, event_data)
if blocking:
success = executed_event.wait(SERVICE_CALL_LIMIT)
self._bus.remove_listener(
EVENT_SERVICE_EXECUTED, service_executed)
return success
def _event_to_service_call(self, event):
""" Calls a service from an event. """
service_data = dict(event.data)
domain = service_data.pop(ATTR_DOMAIN, None)
service = service_data.pop(ATTR_SERVICE, None)
if not self.has_service(domain, service):
return
service_handler = self._services[domain][service]
service_call = ServiceCall(domain, service, service_data)
# Add a job to the pool that calls _execute_service
self._pool.add_job(JobPriority.EVENT_SERVICE,
(self._execute_service,
(service_handler, service_call)))
def _execute_service(self, service_and_call):
""" Executes a service and fires a SERVICE_EXECUTED event. """
service, call = service_and_call
service(call)
self._bus.fire(
EVENT_SERVICE_EXECUTED,
{ATTR_SERVICE_CALL_ID: call.data[ATTR_SERVICE_CALL_ID]})
def _generate_unique_id(self):
""" Generates a unique service call id. """
self._cur_id += 1
return "{}-{}".format(id(self), self._cur_id)
class Config(object):
""" Configuration settings for Home Assistant. """
# pylint: disable=too-many-instance-attributes
def __init__(self):
self.latitude = None
self.longitude = None
self.temperature_unit = None
self.location_name = None
self.time_zone = None
# List of loaded components
self.components = []
# Remote.API object pointing at local API
self.api = None
# Directory that holds the configuration
self.config_dir = os.path.join(os.getcwd(), 'config')
def path(self, *path):
""" Returns path to the file within the config dir. """
return os.path.join(self.config_dir, *path)
def temperature(self, value, unit):
""" Converts temperature to user preferred unit if set. """
if not (unit in (TEMP_CELCIUS, TEMP_FAHRENHEIT) and
self.temperature_unit and unit != self.temperature_unit):
return value, unit
try:
if unit == TEMP_CELCIUS:
# Convert C to F
return round(float(value) * 1.8 + 32.0, 1), TEMP_FAHRENHEIT
# Convert F to C
return round((float(value)-32.0)/1.8, 1), TEMP_CELCIUS
except ValueError:
# Could not convert value to float
return value, unit
def as_dict(self):
""" Converts config to a dictionary. """
time_zone = self.time_zone or date_util.UTC
return {
'latitude': self.latitude,
'longitude': self.longitude,
'temperature_unit': self.temperature_unit,
'location_name': self.location_name,
'time_zone': time_zone.zone,
'components': self.components,
}
class HomeAssistantError(Exception):
""" General Home Assistant exception occured. """
pass
class InvalidEntityFormatError(HomeAssistantError):
""" When an invalid formatted entity is encountered. """
pass
class NoEntitySpecifiedError(HomeAssistantError):
""" When no entity is specified. """
pass
def create_timer(hass, interval=TIMER_INTERVAL):
""" Creates a timer. Timer will start on HOMEASSISTANT_START. """
# We want to be able to fire every time a minute starts (seconds=0).
# We want this so other modules can use that to make sure they fire
# every minute.
assert 60 % interval == 0, "60 % TIMER_INTERVAL should be 0!"
def timer():
"""Send an EVENT_TIME_CHANGED on interval."""
stop_event = threading.Event()
def stop_timer(event):
"""Stop the timer."""
stop_event.set()
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_timer)
_LOGGER.info("Timer:starting")
last_fired_on_second = -1
calc_now = date_util.utcnow
while not stop_event.isSet():
now = calc_now()
# First check checks if we are not on a second matching the
# timer interval. Second check checks if we did not already fire
# this interval.
if now.second % interval or \
now.second == last_fired_on_second:
# Sleep till it is the next time that we have to fire an event.
# Aim for halfway through the second that fits TIMER_INTERVAL.
# If TIMER_INTERVAL is 10 fire at .5, 10.5, 20.5, etc seconds.
# This will yield the best results because time.sleep() is not
# 100% accurate because of non-realtime OS's
slp_seconds = interval - now.second % interval + \
.5 - now.microsecond/1000000.0
time.sleep(slp_seconds)
now = calc_now()
last_fired_on_second = now.second
# Event might have been set while sleeping
if not stop_event.isSet():
try:
hass.bus.fire(EVENT_TIME_CHANGED, {ATTR_NOW: now})
except HomeAssistantError:
# HA raises error if firing event after it has shut down
break
def start_timer(event):
"""Start the timer."""
thread = threading.Thread(target=timer)
thread.daemon = True
thread.start()
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_timer)
def create_worker_pool(worker_count=MIN_WORKER_THREAD):
""" Creates a worker pool to be used. """
def job_handler(job):
""" Called whenever a job is available to do. """
try:
func, arg = job
func(arg)
except Exception: # pylint: disable=broad-except
# Catch any exception our service/event_listener might throw
# We do not want to crash our ThreadPool
_LOGGER.exception("BusHandler:Exception doing job")
def busy_callback(worker_count, current_jobs, pending_jobs_count):
""" Callback to be called when the pool queue gets too big. """
_LOGGER.warning(
"WorkerPool:All %d threads are busy and %d jobs pending",
worker_count, pending_jobs_count)
for start, job in current_jobs:
_LOGGER.warning("WorkerPool:Current job from %s: %s",
date_util.datetime_to_local_str(start), job)
return util.ThreadPool(job_handler, worker_count, busy_callback)

13
homeassistant/bootstrap.py
View File

@ -13,7 +13,7 @@ import os
import logging
from collections import defaultdict
import homeassistant
import homeassistant.core as core
import homeassistant.util.dt as date_util
import homeassistant.util.package as pkg_util
import homeassistant.util.location as loc_util
@ -152,9 +152,9 @@ def from_config_dict(config, hass=None):
Dynamically loads required components and its dependencies.
"""
if hass is None:
hass = homeassistant.HomeAssistant()
hass = core.HomeAssistant()
process_ha_core_config(hass, config.get(homeassistant.DOMAIN, {}))
process_ha_core_config(hass, config.get(core.DOMAIN, {}))
enable_logging(hass)
@ -168,7 +168,7 @@ def from_config_dict(config, hass=None):
# Filter out the repeating and common config section [homeassistant]
components = (key for key in config.keys()
if ' ' not in key and key != homeassistant.DOMAIN)
if ' ' not in key and key != core.DOMAIN)
if not core_components.setup(hass, config):
_LOGGER.error('Home Assistant core failed to initialize. '
@ -192,7 +192,7 @@ def from_config_file(config_path, hass=None):
instantiates a new Home Assistant object if 'hass' is not given.
"""
if hass is None:
hass = homeassistant.HomeAssistant()
hass = core.HomeAssistant()
# Set config dir to directory holding config file
hass.config.config_dir = os.path.abspath(os.path.dirname(config_path))
@ -222,7 +222,8 @@ def enable_logging(hass):
}
))
except ImportError:
_LOGGER.warn("Colorlog package not found, console coloring disabled")
_LOGGER.warning(
"Colorlog package not found, console coloring disabled")
# Log errors to a file if we have write access to file or config dir
err_log_path = hass.config.path('home-assistant.log')

2
homeassistant/components/__init__.py
View File

@ -17,7 +17,7 @@ Each component should publish services only under its own domain.
import itertools as it
import logging
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.util as util
from homeassistant.helpers import extract_entity_ids
from homeassistant.loader import get_component

2
homeassistant/components/api.py
View File

@ -9,7 +9,7 @@ import logging
import threading
import json
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.helpers.state import TrackStates
import homeassistant.remote as rem
from homeassistant.const import (

7
homeassistant/components/automation/__init__.py
View File

@ -25,6 +25,7 @@ _LOGGER = logging.getLogger(__name__)
def setup(hass, config):
""" Sets up automation. """
success = False
for p_type, p_config in config_per_platform(config, DOMAIN, _LOGGER):
platform = prepare_setup_platform(hass, config, DOMAIN, p_type)
@ -36,11 +37,12 @@ def setup(hass, config):
if platform.register(hass, p_config, _get_action(hass, p_config)):
_LOGGER.info(
"Initialized %s rule %s", p_type, p_config.get(CONF_ALIAS, ""))
success = True
else:
_LOGGER.error(
"Error setting up rule %s", p_config.get(CONF_ALIAS, ""))
return True
return success
def _get_action(hass, config):
@ -56,8 +58,7 @@ def _get_action(hass, config):
service_data = config.get(CONF_SERVICE_DATA, {})
if not isinstance(service_data, dict):
_LOGGER.error(
"%s should be a serialized JSON object", CONF_SERVICE_DATA)
_LOGGER.error("%s should be a dictionary", CONF_SERVICE_DATA)
service_data = {}
if CONF_SERVICE_ENTITY_ID in config:

16
homeassistant/components/conversation.py
View File

@ -8,7 +8,7 @@ This is more a proof of concept.
import logging
import re
import homeassistant
from homeassistant import core
from homeassistant.const import (
ATTR_FRIENDLY_NAME, ATTR_ENTITY_ID, SERVICE_TURN_ON, SERVICE_TURN_OFF)
@ -52,16 +52,14 @@ def setup(hass, config):
return
if command == 'on':
hass.services.call(
homeassistant.DOMAIN, SERVICE_TURN_ON, {
ATTR_ENTITY_ID: entity_ids,
}, blocking=True)
hass.services.call(core.DOMAIN, SERVICE_TURN_ON, {
ATTR_ENTITY_ID: entity_ids,
}, blocking=True)
elif command == 'off':
hass.services.call(
homeassistant.DOMAIN, SERVICE_TURN_OFF, {
ATTR_ENTITY_ID: entity_ids,
}, blocking=True)
hass.services.call(core.DOMAIN, SERVICE_TURN_OFF, {
ATTR_ENTITY_ID: entity_ids,
}, blocking=True)
else:
logger.error(

2
homeassistant/components/demo.py
View File

@ -6,7 +6,7 @@ Sets up a demo environment that mimics interaction with devices.
"""
import time
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.bootstrap as bootstrap
import homeassistant.loader as loader
from homeassistant.const import (

66
homeassistant/components/device_tracker/tplink.py
View File

@ -31,6 +31,7 @@ password
The password for your given admin account.
"""
import base64
import logging
from datetime import timedelta
import re
@ -55,7 +56,10 @@ def get_scanner(hass, config):
_LOGGER):
return None
scanner = TplinkDeviceScanner(config[DOMAIN])
scanner = Tplink2DeviceScanner(config[DOMAIN])
if not scanner.success_init:
scanner = TplinkDeviceScanner(config[DOMAIN])
return scanner if scanner.success_init else None
@ -115,3 +119,63 @@ class TplinkDeviceScanner(object):
return True
return False
class Tplink2DeviceScanner(TplinkDeviceScanner):
""" This class queries a wireless router running newer version of TP-Link
firmware for connected devices.
"""
def scan_devices(self):
""" Scans for new devices and return a
list containing found device ids. """
self._update_info()
return self.last_results.keys()
# pylint: disable=no-self-use
def get_device_name(self, device):
""" The TP-Link firmware doesn't save the name of the wireless
device. """
return self.last_results.get(device)
@Throttle(MIN_TIME_BETWEEN_SCANS)
def _update_info(self):
""" Ensures the information from the TP-Link router is up to date.
Returns boolean if scanning successful. """
with self.lock:
_LOGGER.info("Loading wireless clients...")
url = 'http://{}/data/map_access_wireless_client_grid.json'\
.format(self.host)
referer = 'http://{}'.format(self.host)
# Router uses Authorization cookie instead of header
# Let's create the cookie
username_password = '{}:{}'.format(self.username, self.password)
b64_encoded_username_password = base64.b64encode(
username_password.encode('ascii')
).decode('ascii')
cookie = 'Authorization=Basic {}'\
.format(b64_encoded_username_password)
response = requests.post(url, headers={'referer': referer,
'cookie': cookie})
try:
result = response.json().get('data')
except ValueError:
_LOGGER.error("Router didn't respond with JSON. "
"Check if credentials are correct.")
return False
if result:
self.last_results = {
device['mac_addr'].replace('-', ':'): device['name']
for device in result
}
return True
return False

2
homeassistant/components/group.py
View File

@ -5,7 +5,7 @@ homeassistant.components.group
Provides functionality to group devices that can be turned on or off.
"""
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.helpers import generate_entity_id
from homeassistant.helpers.event import track_state_change
from homeassistant.helpers.entity import Entity

2
homeassistant/components/http.py
View File

@ -86,7 +86,7 @@ from http import cookies
from socketserver import ThreadingMixIn
from urllib.parse import urlparse, parse_qs
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.const import (
SERVER_PORT, CONTENT_TYPE_JSON,
HTTP_HEADER_HA_AUTH, HTTP_HEADER_CONTENT_TYPE, HTTP_HEADER_ACCEPT_ENCODING,

2
homeassistant/components/light/__init__.py
View File

@ -1,6 +1,6 @@
"""
homeassistant.components.light
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Provides functionality to interact with lights.

6
homeassistant/components/light/hue.py
View File

@ -1,4 +1,8 @@
""" Support for Hue lights. """
"""
homeassistant.components.light.hue
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for Hue lights.
"""
import logging
import socket
from datetime import timedelta

12
homeassistant/components/light/isy994.py
View File

@ -1,8 +1,10 @@
""" Support for ISY994 lights. """
# system imports
"""
homeassistant.components.light.isy994
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for ISY994 lights.
"""
import logging
# homeassistant imports
from homeassistant.components.isy994 import (ISYDeviceABC, ISY, SENSOR_STRING,
HIDDEN_STRING)
from homeassistant.components.light import ATTR_BRIGHTNESS
@ -10,7 +12,7 @@ from homeassistant.const import STATE_ON, STATE_OFF
def setup_platform(hass, config, add_devices, discovery_info=None):
""" Sets up the isy994 platform. """
""" Sets up the ISY994 platform. """
logger = logging.getLogger(__name__)
devs = []
# verify connection
@ -29,7 +31,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
class ISYLightDevice(ISYDeviceABC):
""" represents as isy light within home assistant. """
""" Represents as ISY light. """
_domain = 'light'
_dtype = 'analog'

21
homeassistant/components/light/limitlessled.py
View File

@ -1,16 +1,21 @@
"""
homeassistant.components.light.limitlessled
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for LimitlessLED bulbs, also known as...
EasyBulb
AppLight
AppLamp
MiLight
LEDme
dekolight
iLight
- EasyBulb
- AppLight
- AppLamp
- MiLight
- LEDme
- dekolight
- iLight
Configuration:
To use limitlessled you will need to add the following to your
config/configuration.yaml.
light:
platform: limitlessled

10
homeassistant/components/light/tellstick.py
View File

@ -1,4 +1,8 @@
""" Support for Tellstick lights. """
"""
homeassistant.components.light.tellstick
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for Tellstick lights.
"""
import logging
# pylint: disable=no-name-in-module, import-error
from homeassistant.components.light import Light, ATTR_BRIGHTNESS
@ -9,7 +13,7 @@ REQUIREMENTS = ['tellcore-py>=1.0.4']
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
""" Find and return tellstick lights. """
""" Find and return Tellstick lights. """
try:
import tellcore.telldus as telldus
@ -29,7 +33,7 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
class TellstickLight(Light):
""" Represents a tellstick light """
""" Represents a Tellstick light. """
last_sent_command_mask = (tellcore_constants.TELLSTICK_TURNON |
tellcore_constants.TELLSTICK_TURNOFF |
tellcore_constants.TELLSTICK_DIM |

32
homeassistant/components/light/vera.py
View File

@ -1,13 +1,15 @@
"""
Support for Vera lights.
homeassistant.components.light.vera
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for Vera lights. This component is useful if you wish for switches
connected to your Vera controller to appear as lights in Home Assistant.
All switches will be added as a light unless you exclude them in the config.
Configuration:
This component is useful if you wish for switches connected to your Vera
controller to appear as lights in homeassistant. All switches will be added
as a light unless you exclude them in the config.
To use the Vera lights you will need to add something like the following to
your config/configuration.yaml
your config/configuration.yaml.
light:
platform: vera
@ -19,22 +21,19 @@ light:
13:
name: Another switch
VARIABLES:
Variables:
vera_controller_url
*Required
This is the base URL of your vera controller including the port number if not
running on 80
Example: http://192.168.1.21:3480/
running on 80. Example: http://192.168.1.21:3480/
device_data
*Optional
This contains an array additional device info for your Vera devices. It is not
This contains an array additional device info for your Vera devices. It is not
required and if not specified all lights configured in your Vera controller
will be added with default values. You should use the id of your vera device
as the key for the device within device_data
will be added with default values. You should use the id of your vera device
as the key for the device within device_data.
These are the variables for the device_data array:
@ -42,13 +41,12 @@ name
*Optional
This parameter allows you to override the name of your Vera device in the HA
interface, if not specified the value configured for the device in your Vera
will be used
will be used.
exclude
*Optional
This parameter allows you to exclude the specified device from homeassistant,
it should be set to "true" if you want this device excluded
This parameter allows you to exclude the specified device from Home Assistant,
it should be set to "true" if you want this device excluded.
"""
import logging

8
homeassistant/components/light/wink.py
View File

@ -1,4 +1,8 @@
""" Support for Wink lights. """
"""
homeassistant.components.light.wink
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for Wink lights.
"""
import logging
from homeassistant.components.light import ATTR_BRIGHTNESS
@ -29,7 +33,7 @@ def setup_platform(hass, config, add_devices_callback, discovery_info=None):
class WinkLight(WinkToggleDevice):
""" Represents a Wink light """
""" Represents a Wink light. """
# pylint: disable=too-few-public-methods
def turn_on(self, **kwargs):

2
homeassistant/components/logbook.py
View File

@ -8,7 +8,7 @@ from datetime import timedelta
from itertools import groupby
import re
from homeassistant import State, DOMAIN as HA_DOMAIN
from homeassistant.core import State, DOMAIN as HA_DOMAIN
from homeassistant.const import (
EVENT_STATE_CHANGED, STATE_HOME, STATE_ON, STATE_OFF,
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP, HTTP_BAD_REQUEST)

6
homeassistant/components/mqtt.py
View File

@ -46,7 +46,7 @@ The keep alive in seconds for this client. Default is 60.
import logging
import socket
from homeassistant import HomeAssistantError
from homeassistant.core import HomeAssistantError
import homeassistant.util as util
from homeassistant.helpers import validate_config
from homeassistant.const import (
@ -155,7 +155,7 @@ def setup(hass, config):
# This is based on one of the paho-mqtt examples:
# http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt.python.git/tree/examples/sub-class.py
# pylint: disable=too-many-arguments
class MQTT(object):
class MQTT(object): # pragma: no cover
""" Implements messaging service for MQTT. """
def __init__(self, hass, broker, port, client_id, keepalive, username,
password):
@ -237,7 +237,7 @@ class MQTT(object):
})
def _raise_on_error(result):
def _raise_on_error(result): # pragma: no cover
""" Raise error if error result. """
if result != 0:
raise HomeAssistantError('Error talking to MQTT: {}'.format(result))

2
homeassistant/components/recorder.py
View File

@ -13,7 +13,7 @@ from datetime import datetime, date
import json
import atexit
from homeassistant import Event, EventOrigin, State
from homeassistant.core import Event, EventOrigin, State
import homeassistant.util.dt as date_util
from homeassistant.remote import JSONEncoder
from homeassistant.const import (

2
homeassistant/components/scene.py
View File

@ -18,7 +18,7 @@ old state will not be restored when it is being deactivated.
import logging
from collections import namedtuple
from homeassistant import State
from homeassistant.core import State
from homeassistant.helpers.event import track_state_change
from homeassistant.helpers.entity import ToggleEntity
from homeassistant.helpers.entity_component import EntityComponent

5
homeassistant/components/sensor/__init__.py
View File

@ -6,7 +6,7 @@ Component to interface with various sensors that can be monitored.
import logging
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.components import wink, zwave, isy994
from homeassistant.components import wink, zwave, isy994, verisure
DOMAIN = 'sensor'
DEPENDENCIES = []
@ -18,7 +18,8 @@ ENTITY_ID_FORMAT = DOMAIN + '.{}'
DISCOVERY_PLATFORMS = {
wink.DISCOVER_SENSORS: 'wink',
zwave.DISCOVER_SENSORS: 'zwave',
isy994.DISCOVER_SENSORS: 'isy994'
isy994.DISCOVER_SENSORS: 'isy994',
verisure.DISCOVER_SENSORS: 'verisure'
}

2
homeassistant/components/sensor/mysensors.py
View File

@ -44,7 +44,7 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
""" Setup the mysensors platform. """
import mysensors.mysensors as mysensors
import mysensors.const as const
import mysensors.const_14 as const
devices = {} # keep track of devices added to HA
# Just assume celcius means that the user wants metric for now.

127
homeassistant/components/sensor/verisure.py Normal file
View File

@ -0,0 +1,127 @@
"""
homeassistant.components.sensor.verisure
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Interfaces with Verisure sensors.
"""
import logging
import homeassistant.components.verisure as verisure
from homeassistant.helpers.entity import Entity
from homeassistant.const import TEMP_CELCIUS
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_devices, discovery_info=None):
""" Sets up the Verisure platform. """
if not verisure.MY_PAGES:
_LOGGER.error('A connection has not been made to Verisure mypages.')
return False
sensors = []
sensors.extend([
VerisureThermometer(value)
for value in verisure.get_climate_status().values()
if verisure.SHOW_THERMOMETERS and
hasattr(value, 'temperature') and value.temperature
])
sensors.extend([
VerisureHygrometer(value)
for value in verisure.get_climate_status().values()
if verisure.SHOW_HYGROMETERS and
hasattr(value, 'humidity') and value.humidity
])
sensors.extend([
VerisureAlarm(value)
for value in verisure.get_alarm_status().values()
if verisure.SHOW_ALARM
])
add_devices(sensors)
class VerisureThermometer(Entity):
""" represents a Verisure thermometer within home assistant. """
def __init__(self, climate_status):
self._id = climate_status.id
self._device = verisure.MY_PAGES.DEVICE_CLIMATE
@property
def name(self):
""" Returns the name of the device. """
return '{} {}'.format(
verisure.STATUS[self._device][self._id].location,
"Temperature")
@property
def state(self):
""" Returns the state of the device. """
# remove ° character
return verisure.STATUS[self._device][self._id].temperature[:-1]
@property
def unit_of_measurement(self):
""" Unit of measurement of this entity """
return TEMP_CELCIUS # can verisure report in fahrenheit?
def update(self):
''' update sensor '''
verisure.update()
class VerisureHygrometer(Entity):
""" represents a Verisure hygrometer within home assistant. """
def __init__(self, climate_status):
self._id = climate_status.id
self._device = verisure.MY_PAGES.DEVICE_CLIMATE
@property
def name(self):
""" Returns the name of the device. """
return '{} {}'.format(
verisure.STATUS[self._device][self._id].location,
"Humidity")
@property
def state(self):
""" Returns the state of the device. """
# remove % character
return verisure.STATUS[self._device][self._id].humidity[:-1]
@property
def unit_of_measurement(self):
""" Unit of measurement of this entity """
return "%"
def update(self):
''' update sensor '''
verisure.update()
class VerisureAlarm(Entity):
""" represents a Verisure alarm status within home assistant. """
def __init__(self, alarm_status):
self._id = alarm_status.id
self._device = verisure.MY_PAGES.DEVICE_ALARM
@property
def name(self):
""" Returns the name of the device. """
return 'Alarm {}'.format(self._id)
@property
def state(self):
""" Returns the state of the device. """
return verisure.STATUS[self._device][self._id].label
def update(self):
''' update sensor '''
verisure.update()

3
homeassistant/components/switch/__init__.py
View File

@ -11,7 +11,7 @@ from homeassistant.helpers.entity import ToggleEntity
from homeassistant.const import (
STATE_ON, SERVICE_TURN_ON, SERVICE_TURN_OFF, ATTR_ENTITY_ID)
from homeassistant.components import group, discovery, wink, isy994
from homeassistant.components import group, discovery, wink, isy994, verisure
DOMAIN = 'switch'
DEPENDENCIES = []
@ -32,6 +32,7 @@ DISCOVERY_PLATFORMS = {
discovery.SERVICE_WEMO: 'wemo',
wink.DISCOVER_SWITCHES: 'wink',
isy994.DISCOVER_SWITCHES: 'isy994',
verisure.DISCOVER_SWITCHES: 'verisure'
}
PROP_TO_ATTR = {

63
homeassistant/components/switch/verisure.py Normal file
View File

@ -0,0 +1,63 @@
"""
homeassistant.components.switch.verisure
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Support for Verisure Smartplugs
"""
import logging
import homeassistant.components.verisure as verisure
from homeassistant.components.switch import SwitchDevice
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_devices, discovery_info=None):
""" Sets up the Arduino platform. """
if not verisure.MY_PAGES:
_LOGGER.error('A connection has not been made to Verisure mypages.')
return False
switches = []
switches.extend([
VerisureSmartplug(value)
for value in verisure.get_smartplug_status().values()
if verisure.SHOW_SMARTPLUGS
])
add_devices(switches)
class VerisureSmartplug(SwitchDevice):
""" Represents a Verisure smartplug. """
def __init__(self, smartplug_status):
self._id = smartplug_status.id
self.status_on = verisure.MY_PAGES.SMARTPLUG_ON
self.status_off = verisure.MY_PAGES.SMARTPLUG_OFF
@property
def name(self):
""" Get the name (location) of the smartplug. """
return verisure.get_smartplug_status()[self._id].location
@property
def is_on(self):
""" Returns True if on """
plug_status = verisure.get_smartplug_status()[self._id].status
return plug_status == self.status_on
def turn_on(self):
""" Set smartplug status on """
verisure.MY_PAGES.set_smartplug_status(
self._id,
self.status_on)
def turn_off(self):
""" Set smartplug status off """
verisure.MY_PAGES.set_smartplug_status(
self._id,
self.status_off)
def update(self):
verisure.update()

179
homeassistant/components/verisure.py Normal file
View File

@ -0,0 +1,179 @@
"""
components.verisure
~~~~~~~~~~~~~~~~~~
Provides support for verisure components
Configuration:
verisure:
username: user@example.com
password: password
alarm: 1
hygrometers: 0
smartplugs: 1
thermometers: 0
Variables:
username
*Required
Username to verisure mypages
password
*Required
Password to verisure mypages
alarm
*Opional
Set to 1 to show alarm, 0 to disable. Default 1
hygrometers
*Opional
Set to 1 to show hygrometers, 0 to disable. Default 1
smartplugs
*Opional
Set to 1 to show smartplugs, 0 to disable. Default 1
thermometers
*Opional
Set to 1 to show thermometers, 0 to disable. Default 1
"""
import logging
from datetime import timedelta
from homeassistant import bootstrap
from homeassistant.loader import get_component
from homeassistant.helpers import validate_config
from homeassistant.util import Throttle
from homeassistant.const import (
EVENT_PLATFORM_DISCOVERED,
ATTR_SERVICE, ATTR_DISCOVERED,
CONF_USERNAME, CONF_PASSWORD)
DOMAIN = "verisure"
DISCOVER_SENSORS = 'verisure.sensors'
DISCOVER_SWITCHES = 'verisure.switches'
DEPENDENCIES = []
REQUIREMENTS = [
'https://github.com/persandstrom/python-verisure/archive/master.zip'
]
_LOGGER = logging.getLogger(__name__)
MY_PAGES = None
STATUS = {}
VERISURE_LOGIN_ERROR = None
VERISURE_ERROR = None
SHOW_THERMOMETERS = True
SHOW_HYGROMETERS = True
SHOW_ALARM = True
SHOW_SMARTPLUGS = True
# if wrong password was given don't try again
WRONG_PASSWORD_GIVEN = False
MIN_TIME_BETWEEN_REQUESTS = timedelta(seconds=5)
def setup(hass, config):
""" Setup the Verisure component. """
if not validate_config(config,
{DOMAIN: [CONF_USERNAME, CONF_PASSWORD]},
_LOGGER):
return False
from verisure import MyPages, LoginError, Error
STATUS[MyPages.DEVICE_ALARM] = {}
STATUS[MyPages.DEVICE_CLIMATE] = {}
STATUS[MyPages.DEVICE_SMARTPLUG] = {}
global SHOW_THERMOMETERS, SHOW_HYGROMETERS, SHOW_ALARM, SHOW_SMARTPLUGS
SHOW_THERMOMETERS = int(config[DOMAIN].get('thermometers', '1'))
SHOW_HYGROMETERS = int(config[DOMAIN].get('hygrometers', '1'))
SHOW_ALARM = int(config[DOMAIN].get('alarm', '1'))
SHOW_SMARTPLUGS = int(config[DOMAIN].get('smartplugs', '1'))
global MY_PAGES
MY_PAGES = MyPages(
config[DOMAIN][CONF_USERNAME],
config[DOMAIN][CONF_PASSWORD])
global VERISURE_LOGIN_ERROR, VERISURE_ERROR
VERISURE_LOGIN_ERROR = LoginError
VERISURE_ERROR = Error
try:
MY_PAGES.login()
except (ConnectionError, Error) as ex:
_LOGGER.error('Could not log in to verisure mypages, %s', ex)
return False
update()
# Load components for the devices in the ISY controller that we support
for comp_name, discovery in ((('sensor', DISCOVER_SENSORS),
('switch', DISCOVER_SWITCHES))):
component = get_component(comp_name)
_LOGGER.info(config[DOMAIN])
bootstrap.setup_component(hass, component.DOMAIN, config)
hass.bus.fire(EVENT_PLATFORM_DISCOVERED,
{ATTR_SERVICE: discovery,
ATTR_DISCOVERED: {}})
return True
def get_alarm_status():
''' return a list of status overviews for alarm components '''
return STATUS[MY_PAGES.DEVICE_ALARM]
def get_climate_status():
''' return a list of status overviews for alarm components '''
return STATUS[MY_PAGES.DEVICE_CLIMATE]
def get_smartplug_status():
''' return a list of status overviews for alarm components '''
return STATUS[MY_PAGES.DEVICE_SMARTPLUG]
def reconnect():
''' reconnect to verisure mypages '''
try:
MY_PAGES.login()
except VERISURE_LOGIN_ERROR as ex:
_LOGGER.error("Could not login to Verisure mypages, %s", ex)
global WRONG_PASSWORD_GIVEN
WRONG_PASSWORD_GIVEN = True
except (ConnectionError, VERISURE_ERROR) as ex:
_LOGGER.error("Could not login to Verisure mypages, %s", ex)
@Throttle(MIN_TIME_BETWEEN_REQUESTS)
def update():
''' Updates the status of verisure components '''
if WRONG_PASSWORD_GIVEN:
# Is there any way to inform user?
return
try:
for overview in MY_PAGES.get_overview(MY_PAGES.DEVICE_ALARM):
STATUS[MY_PAGES.DEVICE_ALARM][overview.id] = overview
for overview in MY_PAGES.get_overview(MY_PAGES.DEVICE_CLIMATE):
STATUS[MY_PAGES.DEVICE_CLIMATE][overview.id] = overview
for overview in MY_PAGES.get_overview(MY_PAGES.DEVICE_SMARTPLUG):
STATUS[MY_PAGES.DEVICE_SMARTPLUG][overview.id] = overview
except ConnectionError as ex:
_LOGGER.error('Caught connection error %s, tries to reconnect', ex)
reconnect()

2
homeassistant/config.py
View File

@ -7,7 +7,7 @@ Module to help with parsing and generating configuration files.
import logging
import os
from homeassistant import HomeAssistantError
from homeassistant.core import HomeAssistantError
from homeassistant.const import (
CONF_LATITUDE, CONF_LONGITUDE, CONF_TEMPERATURE_UNIT, CONF_NAME,
CONF_TIME_ZONE)

800
homeassistant/core.py Normal file
View File

@ -0,0 +1,800 @@
"""
homeassistant
~~~~~~~~~~~~~
Home Assistant is a Home Automation framework for observing the state
of entities and react to changes.
"""
import os
import time
import logging
import threading
import enum
import re
import functools as ft
from collections import namedtuple
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP,
SERVICE_HOMEASSISTANT_STOP, EVENT_TIME_CHANGED, EVENT_STATE_CHANGED,
EVENT_CALL_SERVICE, ATTR_NOW, ATTR_DOMAIN, ATTR_SERVICE, MATCH_ALL,
EVENT_SERVICE_EXECUTED, ATTR_SERVICE_CALL_ID, EVENT_SERVICE_REGISTERED,
TEMP_CELCIUS, TEMP_FAHRENHEIT, ATTR_FRIENDLY_NAME)
import homeassistant.util as util
import homeassistant.util.dt as date_util
import homeassistant.helpers.temperature as temp_helper
DOMAIN = "homeassistant"
# How often time_changed event should fire
TIMER_INTERVAL = 1 # seconds
# How long we wait for the result of a service call
SERVICE_CALL_LIMIT = 10 # seconds
# Define number of MINIMUM worker threads.
# During bootstrap of HA (see bootstrap._setup_component()) worker threads
# will be added for each component that polls devices.
MIN_WORKER_THREAD = 2
# Pattern for validating entity IDs (format: <domain>.<entity>)
ENTITY_ID_PATTERN = re.compile(r"^(?P<domain>\w+)\.(?P<entity>\w+)$")
_LOGGER = logging.getLogger(__name__)
# Temporary to support deprecated methods
_MockHA = namedtuple("MockHomeAssistant", ['bus'])
class HomeAssistant(object):
""" Core class to route all communication to right components. """
def __init__(self):
self.pool = pool = create_worker_pool()
self.bus = EventBus(pool)
self.services = ServiceRegistry(self.bus, pool)
self.states = StateMachine(self.bus)
self.config = Config()
def start(self):
""" Start home assistant. """
_LOGGER.info(
"Starting Home Assistant (%d threads)", self.pool.worker_count)
create_timer(self)
self.bus.fire(EVENT_HOMEASSISTANT_START)
def block_till_stopped(self):
""" Will register service homeassistant/stop and
will block until called. """
request_shutdown = threading.Event()
def stop_homeassistant(service):
""" Stops Home Assistant. """
request_shutdown.set()
self.services.register(
DOMAIN, SERVICE_HOMEASSISTANT_STOP, stop_homeassistant)
while not request_shutdown.isSet():
try:
time.sleep(1)
except KeyboardInterrupt:
break
self.stop()
def stop(self):
""" Stops Home Assistant and shuts down all threads. """
_LOGGER.info("Stopping")
self.bus.fire(EVENT_HOMEASSISTANT_STOP)
# Wait till all responses to homeassistant_stop are done
self.pool.block_till_done()
self.pool.stop()
def track_point_in_time(self, action, point_in_time):
"""Deprecated method as of 8/4/2015 to track point in time."""
_LOGGER.warning(
'hass.track_point_in_time is deprecated. '
'Please use homeassistant.helpers.event.track_point_in_time')
import homeassistant.helpers.event as helper
helper.track_point_in_time(self, action, point_in_time)
def track_point_in_utc_time(self, action, point_in_time):
"""Deprecated method as of 8/4/2015 to track point in UTC time."""
_LOGGER.warning(
'hass.track_point_in_utc_time is deprecated. '
'Please use homeassistant.helpers.event.track_point_in_utc_time')
import homeassistant.helpers.event as helper
helper.track_point_in_utc_time(self, action, point_in_time)
def track_utc_time_change(self, action,
year=None, month=None, day=None,
hour=None, minute=None, second=None):
"""Deprecated method as of 8/4/2015 to track UTC time change."""
# pylint: disable=too-many-arguments
_LOGGER.warning(
'hass.track_utc_time_change is deprecated. '
'Please use homeassistant.helpers.event.track_utc_time_change')
import homeassistant.helpers.event as helper
helper.track_utc_time_change(self, action, year, month, day, hour,
minute, second)
def track_time_change(self, action,
year=None, month=None, day=None,
hour=None, minute=None, second=None, utc=False):
"""Deprecated method as of 8/4/2015 to track time change."""
# pylint: disable=too-many-arguments
_LOGGER.warning(
'hass.track_time_change is deprecated. '
'Please use homeassistant.helpers.event.track_time_change')
import homeassistant.helpers.event as helper
helper.track_time_change(self, action, year, month, day, hour,
minute, second)
class JobPriority(util.OrderedEnum):
""" Provides priorities for bus events. """
# pylint: disable=no-init,too-few-public-methods
EVENT_CALLBACK = 0
EVENT_SERVICE = 1
EVENT_STATE = 2
EVENT_TIME = 3
EVENT_DEFAULT = 4
@staticmethod
def from_event_type(event_type):
""" Returns a priority based on event type. """
if event_type == EVENT_TIME_CHANGED:
return JobPriority.EVENT_TIME
elif event_type == EVENT_STATE_CHANGED:
return JobPriority.EVENT_STATE
elif event_type == EVENT_CALL_SERVICE:
return JobPriority.EVENT_SERVICE
elif event_type == EVENT_SERVICE_EXECUTED:
return JobPriority.EVENT_CALLBACK
else:
return JobPriority.EVENT_DEFAULT
class EventOrigin(enum.Enum):
""" Distinguish between origin of event. """
# pylint: disable=no-init,too-few-public-methods
local = "LOCAL"
remote = "REMOTE"
def __str__(self):
return self.value
# pylint: disable=too-few-public-methods
class Event(object):
""" Represents an event within the Bus. """
__slots__ = ['event_type', 'data', 'origin', 'time_fired']
def __init__(self, event_type, data=None, origin=EventOrigin.local,
time_fired=None):
self.event_type = event_type
self.data = data or {}
self.origin = origin
self.time_fired = date_util.strip_microseconds(
time_fired or date_util.utcnow())
def as_dict(self):
""" Returns a dict representation of this Event. """
return {
'event_type': self.event_type,
'data': dict(self.data),
'origin': str(self.origin),
'time_fired': date_util.datetime_to_str(self.time_fired),
}
def __repr__(self):
# pylint: disable=maybe-no-member
if self.data:
return "<Event {}[{}]: {}>".format(
self.event_type, str(self.origin)[0],
util.repr_helper(self.data))
else:
return "<Event {}[{}]>".format(self.event_type,
str(self.origin)[0])
def __eq__(self, other):
return (self.__class__ == other.__class__ and
self.event_type == other.event_type and
self.data == other.data and
self.origin == other.origin and
self.time_fired == other.time_fired)
class EventBus(object):
""" Class that allows different components to communicate via services
and events.
"""
def __init__(self, pool=None):
self._listeners = {}
self._lock = threading.Lock()
self._pool = pool or create_worker_pool()
@property
def listeners(self):
""" Dict with events that is being listened for and the number
of listeners.
"""
with self._lock:
return {key: len(self._listeners[key])
for key in self._listeners}
def fire(self, event_type, event_data=None, origin=EventOrigin.local):
""" Fire an event. """
if not self._pool.running:
raise HomeAssistantError('Home Assistant has shut down.')
with self._lock:
# Copy the list of the current listeners because some listeners
# remove themselves as a listener while being executed which
# causes the iterator to be confused.
get = self._listeners.get
listeners = get(MATCH_ALL, []) + get(event_type, [])
event = Event(event_type, event_data, origin)
if event_type != EVENT_TIME_CHANGED:
_LOGGER.info("Bus:Handling %s", event)
if not listeners:
return
job_priority = JobPriority.from_event_type(event_type)
for func in listeners:
self._pool.add_job(job_priority, (func, event))
def listen(self, event_type, listener):
""" Listen for all events or events of a specific type.
To listen to all events specify the constant ``MATCH_ALL``
as event_type.
"""
with self._lock:
if event_type in self._listeners:
self._listeners[event_type].append(listener)
else:
self._listeners[event_type] = [listener]
def listen_once(self, event_type, listener):
""" Listen once for event of a specific type.
To listen to all events specify the constant ``MATCH_ALL``
as event_type.
Returns registered listener that can be used with remove_listener.
"""
@ft.wraps(listener)
def onetime_listener(event):
""" Removes listener from eventbus and then fires listener. """
if hasattr(onetime_listener, 'run'):
return
# Set variable so that we will never run twice.
# Because the event bus might have to wait till a thread comes
# available to execute this listener it might occur that the
# listener gets lined up twice to be executed.
# This will make sure the second time it does nothing.
onetime_listener.run = True
self.remove_listener(event_type, onetime_listener)
listener(event)
self.listen(event_type, onetime_listener)
return onetime_listener
def remove_listener(self, event_type, listener):
""" Removes a listener of a specific event_type. """
with self._lock:
try:
self._listeners[event_type].remove(listener)
# delete event_type list if empty
if not self._listeners[event_type]:
self._listeners.pop(event_type)
except (KeyError, ValueError):
# KeyError is key event_type listener did not exist
# ValueError if listener did not exist within event_type
pass
class State(object):
"""
Object to represent a state within the state machine.
entity_id: the entity that is represented.
state: the state of the entity
attributes: extra information on entity and state
last_changed: last time the state was changed, not the attributes.
last_updated: last time this object was updated.
"""
__slots__ = ['entity_id', 'state', 'attributes',
'last_changed', 'last_updated']
# pylint: disable=too-many-arguments
def __init__(self, entity_id, state, attributes=None, last_changed=None,
last_updated=None):
if not ENTITY_ID_PATTERN.match(entity_id):
raise InvalidEntityFormatError((
"Invalid entity id encountered: {}. "
"Format should be <domain>.<object_id>").format(entity_id))
self.entity_id = entity_id.lower()
self.state = state
self.attributes = attributes or {}
self.last_updated = date_util.strip_microseconds(
last_updated or date_util.utcnow())
# Strip microsecond from last_changed else we cannot guarantee
# state == State.from_dict(state.as_dict())
# This behavior occurs because to_dict uses datetime_to_str
# which does not preserve microseconds
self.last_changed = date_util.strip_microseconds(
last_changed or self.last_updated)
@property
def domain(self):
""" Returns domain of this state. """
return util.split_entity_id(self.entity_id)[0]
@property
def object_id(self):
""" Returns object_id of this state. """
return util.split_entity_id(self.entity_id)[1]
@property
def name(self):
""" Name to represent this state. """
return (
self.attributes.get(ATTR_FRIENDLY_NAME) or
self.object_id.replace('_', ' '))
def copy(self):
""" Creates a copy of itself. """
return State(self.entity_id, self.state,
dict(self.attributes), self.last_changed)
def as_dict(self):
""" Converts State to a dict to be used within JSON.
Ensures: state == State.from_dict(state.as_dict()) """
return {'entity_id': self.entity_id,
'state': self.state,
'attributes': self.attributes,
'last_changed': date_util.datetime_to_str(self.last_changed),
'last_updated': date_util.datetime_to_str(self.last_updated)}
@classmethod
def from_dict(cls, json_dict):
""" Static method to create a state from a dict.
Ensures: state == State.from_json_dict(state.to_json_dict()) """
if not (json_dict and
'entity_id' in json_dict and
'state' in json_dict):
return None
last_changed = json_dict.get('last_changed')
if last_changed:
last_changed = date_util.str_to_datetime(last_changed)
last_updated = json_dict.get('last_updated')
if last_updated:
last_updated = date_util.str_to_datetime(last_updated)
return cls(json_dict['entity_id'], json_dict['state'],
json_dict.get('attributes'), last_changed, last_updated)
def __eq__(self, other):
return (self.__class__ == other.__class__ and
self.entity_id == other.entity_id and
self.state == other.state and
self.attributes == other.attributes)
def __repr__(self):
attr = "; {}".format(util.repr_helper(self.attributes)) \
if self.attributes else ""
return "<state {}={}{} @ {}>".format(
self.entity_id, self.state, attr,
date_util.datetime_to_local_str(self.last_changed))
class StateMachine(object):
""" Helper class that tracks the state of different entities. """
def __init__(self, bus):
self._states = {}
self._bus = bus
self._lock = threading.Lock()
def entity_ids(self, domain_filter=None):
""" List of entity ids that are being tracked. """
if domain_filter is None:
return list(self._states.keys())
domain_filter = domain_filter.lower()
return [state.entity_id for key, state
in self._states.items()
if util.split_entity_id(key)[0] == domain_filter]
def all(self):
""" Returns a list of all states. """
with self._lock:
return [state.copy() for state in self._states.values()]
def get(self, entity_id):
""" Returns the state of the specified entity. """
state = self._states.get(entity_id.lower())
# Make a copy so people won't mutate the state
return state.copy() if state else None
def is_state(self, entity_id, state):
""" Returns True if entity exists and is specified state. """
entity_id = entity_id.lower()
return (entity_id in self._states and
self._states[entity_id].state == state)
def remove(self, entity_id):
""" Removes an entity from the state machine.
Returns boolean to indicate if an entity was removed. """
entity_id = entity_id.lower()
with self._lock:
return self._states.pop(entity_id, None) is not None
def set(self, entity_id, new_state, attributes=None):
""" Set the state of an entity, add entity if it does not exist.
Attributes is an optional dict to specify attributes of this state.
If you just update the attributes and not the state, last changed will
not be affected.
"""
entity_id = entity_id.lower()
new_state = str(new_state)
attributes = attributes or {}
with self._lock:
old_state = self._states.get(entity_id)
is_existing = old_state is not None
same_state = is_existing and old_state.state == new_state
same_attr = is_existing and old_state.attributes == attributes
if same_state and same_attr:
return
# If state did not exist or is different, set it
last_changed = old_state.last_changed if same_state else None
state = State(entity_id, new_state, attributes, last_changed)
self._states[entity_id] = state
event_data = {'entity_id': entity_id, 'new_state': state}
if old_state:
event_data['old_state'] = old_state
self._bus.fire(EVENT_STATE_CHANGED, event_data)
def track_change(self, entity_ids, action, from_state=None, to_state=None):
"""
DEPRECATED AS OF 8/4/2015
"""
_LOGGER.warning(
'hass.states.track_change is deprecated. '
'Use homeassistant.helpers.event.track_state_change instead.')
import homeassistant.helpers.event as helper
helper.track_state_change(_MockHA(self._bus), entity_ids, action,
from_state, to_state)
# pylint: disable=too-few-public-methods
class ServiceCall(object):
""" Represents a call to a service. """
__slots__ = ['domain', 'service', 'data']
def __init__(self, domain, service, data=None):
self.domain = domain
self.service = service
self.data = data or {}
def __repr__(self):
if self.data:
return "<ServiceCall {}.{}: {}>".format(
self.domain, self.service, util.repr_helper(self.data))
else:
return "<ServiceCall {}.{}>".format(self.domain, self.service)
class ServiceRegistry(object):
""" Offers services over the eventbus. """
def __init__(self, bus, pool=None):
self._services = {}
self._lock = threading.Lock()
self._pool = pool or create_worker_pool()
self._bus = bus
self._cur_id = 0
bus.listen(EVENT_CALL_SERVICE, self._event_to_service_call)
@property
def services(self):
""" Dict with per domain a list of available services. """
with self._lock:
return {domain: list(self._services[domain].keys())
for domain in self._services}
def has_service(self, domain, service):
""" Returns True if specified service exists. """
return service in self._services.get(domain, [])
def register(self, domain, service, service_func):
""" Register a service. """
with self._lock:
if domain in self._services:
self._services[domain][service] = service_func
else:
self._services[domain] = {service: service_func}
self._bus.fire(
EVENT_SERVICE_REGISTERED,
{ATTR_DOMAIN: domain, ATTR_SERVICE: service})
def call(self, domain, service, service_data=None, blocking=False):
"""
Calls specified service.
Specify blocking=True to wait till service is executed.
Waits a maximum of SERVICE_CALL_LIMIT.
If blocking = True, will return boolean if service executed
succesfully within SERVICE_CALL_LIMIT.
This method will fire an event to call the service.
This event will be picked up by this ServiceRegistry and any
other ServiceRegistry that is listening on the EventBus.
Because the service is sent as an event you are not allowed to use
the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.
"""
call_id = self._generate_unique_id()
event_data = service_data or {}
event_data[ATTR_DOMAIN] = domain
event_data[ATTR_SERVICE] = service
event_data[ATTR_SERVICE_CALL_ID] = call_id
if blocking:
executed_event = threading.Event()
def service_executed(call):
"""
Called when a service is executed.
Will set the event if matches our service call.
"""
if call.data[ATTR_SERVICE_CALL_ID] == call_id:
executed_event.set()
self._bus.listen(EVENT_SERVICE_EXECUTED, service_executed)
self._bus.fire(EVENT_CALL_SERVICE, event_data)
if blocking:
success = executed_event.wait(SERVICE_CALL_LIMIT)
self._bus.remove_listener(
EVENT_SERVICE_EXECUTED, service_executed)
return success
def _event_to_service_call(self, event):
""" Calls a service from an event. """
service_data = dict(event.data)
domain = service_data.pop(ATTR_DOMAIN, None)
service = service_data.pop(ATTR_SERVICE, None)
if not self.has_service(domain, service):
return
service_handler = self._services[domain][service]
service_call = ServiceCall(domain, service, service_data)
# Add a job to the pool that calls _execute_service
self._pool.add_job(JobPriority.EVENT_SERVICE,
(self._execute_service,
(service_handler, service_call)))
def _execute_service(self, service_and_call):
""" Executes a service and fires a SERVICE_EXECUTED event. """
service, call = service_and_call
service(call)
if ATTR_SERVICE_CALL_ID in call.data:
self._bus.fire(
EVENT_SERVICE_EXECUTED,
{ATTR_SERVICE_CALL_ID: call.data[ATTR_SERVICE_CALL_ID]})
def _generate_unique_id(self):
""" Generates a unique service call id. """
self._cur_id += 1
return "{}-{}".format(id(self), self._cur_id)
class Config(object):
""" Configuration settings for Home Assistant. """
# pylint: disable=too-many-instance-attributes
def __init__(self):
self.latitude = None
self.longitude = None
self.temperature_unit = None
self.location_name = None
self.time_zone = None
# List of loaded components
self.components = []
# Remote.API object pointing at local API
self.api = None
# Directory that holds the configuration
self.config_dir = os.path.join(os.getcwd(), 'config')
def path(self, *path):
""" Returns path to the file within the config dir. """
return os.path.join(self.config_dir, *path)
def temperature(self, value, unit):
""" Converts temperature to user preferred unit if set. """
if not (unit in (TEMP_CELCIUS, TEMP_FAHRENHEIT) and
self.temperature_unit and unit != self.temperature_unit):
return value, unit
try:
temp = float(value)
except ValueError: # Could not convert value to float
return value, unit
return (
round(temp_helper.convert(temp, unit, self.temperature_unit), 1),
self.temperature_unit)
def as_dict(self):
""" Converts config to a dictionary. """
time_zone = self.time_zone or date_util.UTC
return {
'latitude': self.latitude,
'longitude': self.longitude,
'temperature_unit': self.temperature_unit,
'location_name': self.location_name,
'time_zone': time_zone.zone,
'components': self.components,
}
class HomeAssistantError(Exception):
""" General Home Assistant exception occured. """
pass
class InvalidEntityFormatError(HomeAssistantError):
""" When an invalid formatted entity is encountered. """
pass
class NoEntitySpecifiedError(HomeAssistantError):
""" When no entity is specified. """
pass
def create_timer(hass, interval=TIMER_INTERVAL):
""" Creates a timer. Timer will start on HOMEASSISTANT_START. """
# We want to be able to fire every time a minute starts (seconds=0).
# We want this so other modules can use that to make sure they fire
# every minute.
assert 60 % interval == 0, "60 % TIMER_INTERVAL should be 0!"
def timer():
"""Send an EVENT_TIME_CHANGED on interval."""
stop_event = threading.Event()
def stop_timer(event):
"""Stop the timer."""
stop_event.set()
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_timer)
_LOGGER.info("Timer:starting")
last_fired_on_second = -1
calc_now = date_util.utcnow
while not stop_event.isSet():
now = calc_now()
# First check checks if we are not on a second matching the
# timer interval. Second check checks if we did not already fire
# this interval.
if now.second % interval or \
now.second == last_fired_on_second:
# Sleep till it is the next time that we have to fire an event.
# Aim for halfway through the second that fits TIMER_INTERVAL.
# If TIMER_INTERVAL is 10 fire at .5, 10.5, 20.5, etc seconds.
# This will yield the best results because time.sleep() is not
# 100% accurate because of non-realtime OS's
slp_seconds = interval - now.second % interval + \
.5 - now.microsecond/1000000.0
time.sleep(slp_seconds)
now = calc_now()
last_fired_on_second = now.second
# Event might have been set while sleeping
if not stop_event.isSet():
try:
hass.bus.fire(EVENT_TIME_CHANGED, {ATTR_NOW: now})
except HomeAssistantError:
# HA raises error if firing event after it has shut down
break
def start_timer(event):
"""Start the timer."""
thread = threading.Thread(target=timer)
thread.daemon = True
thread.start()
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_timer)
def create_worker_pool(worker_count=MIN_WORKER_THREAD):
""" Creates a worker pool to be used. """
def job_handler(job):
""" Called whenever a job is available to do. """
try:
func, arg = job
func(arg)
except Exception: # pylint: disable=broad-except
# Catch any exception our service/event_listener might throw
# We do not want to crash our ThreadPool
_LOGGER.exception("BusHandler:Exception doing job")
def busy_callback(worker_count, current_jobs, pending_jobs_count):
""" Callback to be called when the pool queue gets too big. """
_LOGGER.warning(
"WorkerPool:All %d threads are busy and %d jobs pending",
worker_count, pending_jobs_count)
for start, job in current_jobs:
_LOGGER.warning("WorkerPool:Current job from %s: %s",
date_util.datetime_to_local_str(start), job)
return util.ThreadPool(job_handler, worker_count, busy_callback)

2
homeassistant/helpers/entity.py
View File

@ -7,7 +7,7 @@ Provides ABC for entities in HA.
from collections import defaultdict
from homeassistant import NoEntitySpecifiedError
from homeassistant.core import NoEntitySpecifiedError
from homeassistant.const import (
ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT, ATTR_HIDDEN,

2
homeassistant/helpers/state.py
View File

@ -6,7 +6,7 @@ Helpers that help with state related things.
"""
import logging
from homeassistant import State
from homeassistant.core import State
import homeassistant.util.dt as dt_util
from homeassistant.const import (
STATE_ON, STATE_OFF, SERVICE_TURN_ON, SERVICE_TURN_OFF, ATTR_ENTITY_ID)

19
homeassistant/helpers/temperature.py Normal file
View File

@ -0,0 +1,19 @@
"""
homeassistant.helpers.temperature
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Methods to help handle temperature in Home Assistant.
"""
from homeassistant.const import TEMP_CELCIUS
import homeassistant.util.temperature as temp_util
def convert(temperature, unit, to_unit):
""" Converts temperature to correct unit. """
if unit == to_unit:
return temperature
elif unit == TEMP_CELCIUS:
return temp_util.celcius_to_fahrenheit(temperature)
return temp_util.fahrenheit_to_celcius(temperature)

2
homeassistant/remote.py
View File

@ -17,7 +17,7 @@ import urllib.parse
import requests
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.bootstrap as bootstrap
from homeassistant.const import (

16
homeassistant/util/temperature.py Normal file
View File

@ -0,0 +1,16 @@
"""
homeassistant.util.temperature
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Temperature util functions.
"""
def fahrenheit_to_celcius(fahrenheit):
""" Convert a Fahrenheit temperature to Celcius. """
return (fahrenheit - 32.0) / 1.8
def celcius_to_fahrenheit(celcius):
""" Convert a Celcius temperature to Fahrenheit. """
return celcius * 1.8 + 32.0

2
requirements.txt
View File

@ -110,3 +110,5 @@ paho-mqtt>=1.1
# PyModbus (modbus)
https://github.com/bashwork/pymodbus/archive/python3.zip#pymodbus>=1.2.0
# Verisure
https://github.com/persandstrom/python-verisure/archive/master.zip

41
tests/common.py
View File

@ -6,14 +6,16 @@ Helper method for writing tests.
"""
import os
from datetime import timedelta
from unittest import mock
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.util.location as location_util
import homeassistant.util.dt as dt_util
from homeassistant.helpers.entity import ToggleEntity
from homeassistant.const import (
STATE_ON, STATE_OFF, DEVICE_DEFAULT_NAME, EVENT_TIME_CHANGED,
EVENT_STATE_CHANGED)
from homeassistant.components import sun
from homeassistant.components import sun, mqtt
def get_test_config_dir():
@ -39,6 +41,23 @@ def get_test_home_assistant(num_threads=None):
return hass
def mock_detect_location_info():
""" Mock implementation of util.detect_location_info. """
return location_util.LocationInfo(
ip='1.1.1.1',
country_code='US',
country_name='United States',
region_code='CA',
region_name='California',
city='San Diego',
zip_code='92122',
time_zone='America/Los_Angeles',
latitude='2.0',
longitude='1.0',
use_fahrenheit=True,
)
def mock_service(hass, domain, service):
"""
Sets up a fake service.
@ -52,6 +71,14 @@ def mock_service(hass, domain, service):
return calls
def fire_mqtt_message(hass, topic, payload, qos=0):
hass.bus.fire(mqtt.EVENT_MQTT_MESSAGE_RECEIVED, {
mqtt.ATTR_TOPIC: topic,
mqtt.ATTR_PAYLOAD: payload,
mqtt.ATTR_QOS: qos,
})
def fire_time_changed(hass, time):
hass.bus.fire(EVENT_TIME_CHANGED, {'now': time})
@ -93,6 +120,16 @@ def mock_http_component(hass):
hass.config.components.append('http')
def mock_mqtt_component(hass):
with mock.patch('homeassistant.components.mqtt.MQTT'):
mqtt.setup(hass, {
mqtt.DOMAIN: {
mqtt.CONF_BROKER: 'mock-broker',
}
})
hass.config.components.append(mqtt.DOMAIN)
class MockHTTP(object):
""" Mocks the HTTP module. """

0
tests/components/automation/__init__.py Normal file
View File

78
tests/components/automation/test_event.py Normal file
View File

@ -0,0 +1,78 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant.core as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.event as event
from homeassistant.const import CONF_PLATFORM
class TestAutomationEvent(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_fails_setup_if_no_event_type(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_event(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_event_with_data(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
event.CONF_EVENT_DATA: {'some_attr': 'some_value'},
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.bus.fire('test_event', {'some_attr': 'some_value'})
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_if_event_data_not_matches(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
event.CONF_EVENT_DATA: {'some_attr': 'some_value'},
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.bus.fire('test_event', {'some_attr': 'some_other_value'})
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))

80
tests/components/automation/test_init.py Normal file
View File

@ -0,0 +1,80 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant.core as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.event as event
from homeassistant.const import CONF_PLATFORM, ATTR_ENTITY_ID
class TestAutomationEvent(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_unknown_platform(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'i_do_not_exist'
}
}))
def test_service_data_not_a_dict(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_DATA: 100
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_service_specify_data(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_DATA: {'some': 'data'}
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('data', self.calls[0].data['some'])
def test_service_specify_entity_id(self):
automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'event',
event.CONF_EVENT_TYPE: 'test_event',
automation.CONF_SERVICE: 'test.automation',
automation.CONF_SERVICE_ENTITY_ID: 'hello.world'
}
})
self.hass.bus.fire('test_event')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual(['hello.world'], self.calls[0].data[ATTR_ENTITY_ID])

81
tests/components/automation/test_mqtt.py Normal file
View File

@ -0,0 +1,81 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant.core as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.mqtt as mqtt
from homeassistant.const import CONF_PLATFORM
from tests.common import mock_mqtt_component, fire_mqtt_message
class TestAutomationState(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
mock_mqtt_component(self.hass)
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_no_topic(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_topic_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', '')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_topic_and_payload_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
mqtt.CONF_PAYLOAD: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', 'hello')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_on_topic_but_no_payload_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'mqtt',
mqtt.CONF_TOPIC: 'test-topic',
mqtt.CONF_PAYLOAD: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_mqtt_message(self.hass, 'test-topic', 'no-hello')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))

139
tests/components/automation/test_state.py Normal file
View File

@ -0,0 +1,139 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant.core as ha
import homeassistant.components.automation as automation
import homeassistant.components.automation.state as state
from homeassistant.const import CONF_PLATFORM
class TestAutomationState(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
self.hass.states.set('test.entity', 'hello')
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_setup_fails_if_no_entity_id(self):
self.assertFalse(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
automation.CONF_SERVICE: 'test.automation'
}
}))
def test_if_fires_on_entity_change(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_from_filter(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_FROM: 'hello',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_to_filter(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_on_entity_change_with_both_filters(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_FROM: 'hello',
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_not_fires_if_to_filter_not_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_FROM: 'hello',
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'moon')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))
def test_if_not_fires_if_from_filter_not_match(self):
self.hass.states.set('test.entity', 'bye')
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.entity',
state.CONF_FROM: 'hello',
state.CONF_TO: 'world',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))
def test_if_not_fires_if_entity_not_match(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'state',
state.CONF_ENTITY_ID: 'test.another_entity',
automation.CONF_SERVICE: 'test.automation'
}
}))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))

97
tests/components/automation/test_time.py Normal file
View File

@ -0,0 +1,97 @@
"""
tests.test_component_demo
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests demo component.
"""
import unittest
import homeassistant.core as ha
import homeassistant.loader as loader
import homeassistant.util.dt as dt_util
import homeassistant.components.automation as automation
import homeassistant.components.automation.time as time
from homeassistant.const import CONF_PLATFORM
from tests.common import fire_time_changed
class TestAutomationTime(unittest.TestCase):
""" Test the event automation. """
def setUp(self): # pylint: disable=invalid-name
self.hass = ha.HomeAssistant()
loader.prepare(self.hass)
self.calls = []
def record_call(service):
self.calls.append(service)
self.hass.services.register('test', 'automation', record_call)
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def test_if_fires_when_hour_matches(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_HOURS: 0,
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(hour=0))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_when_minute_matches(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_MINUTES: 0,
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(minute=0))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_when_second_matches(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_SECONDS: 0,
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(second=0))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
def test_if_fires_when_all_matches(self):
self.assertTrue(automation.setup(self.hass, {
automation.DOMAIN: {
CONF_PLATFORM: 'time',
time.CONF_HOURS: 0,
time.CONF_MINUTES: 0,
time.CONF_SECONDS: 0,
automation.CONF_SERVICE: 'test.automation'
}
}))
fire_time_changed(self.hass, dt_util.utcnow().replace(
hour=0, minute=0, second=0))
self.hass.states.set('test.entity', 'world')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))

2
tests/components/test_api.py
View File

@ -10,7 +10,7 @@ import json
import requests
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.bootstrap as bootstrap
import homeassistant.remote as remote
import homeassistant.components.http as http

2
tests/components/test_configurator.py
View File

@ -7,7 +7,7 @@ Tests Configurator component.
# pylint: disable=too-many-public-methods,protected-access
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.components.configurator as configurator
from homeassistant.const import EVENT_TIME_CHANGED

2
tests/components/test_demo.py
View File

@ -6,7 +6,7 @@ Tests demo component.
"""
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.components.demo as demo
from tests.common import mock_http_component

2
tests/components/test_device_tracker.py
View File

@ -10,7 +10,7 @@ from datetime import timedelta
import logging
import os
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.loader as loader
import homeassistant.util.dt as dt_util
from homeassistant.const import (

2
tests/components/test_frontend.py
View File

@ -10,7 +10,7 @@ import unittest
import requests
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.bootstrap as bootstrap
import homeassistant.components.http as http
from homeassistant.const import HTTP_HEADER_HA_AUTH

2
tests/components/test_group.py
View File

@ -8,7 +8,7 @@ Tests the group compoments.
import unittest
import logging
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.const import STATE_ON, STATE_OFF, STATE_HOME, STATE_UNKNOWN
import homeassistant.components.group as group

2
tests/components/test_history.py
View File

@ -9,7 +9,7 @@ import time
import os
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.util.dt as dt_util
from homeassistant.components import history, recorder

2
tests/components/test_init.py
View File

@ -7,7 +7,7 @@ Tests core compoments.
# pylint: disable=protected-access,too-many-public-methods
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.loader as loader
from homeassistant.const import (
STATE_ON, STATE_OFF, SERVICE_TURN_ON, SERVICE_TURN_OFF)

2
tests/components/test_logbook.py
View File

@ -8,7 +8,7 @@ Tests the logbook component.
import unittest
from datetime import timedelta
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.const import (
EVENT_STATE_CHANGED, EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
import homeassistant.util.dt as dt_util

2
tests/components/test_media_player.py
View File

@ -8,7 +8,7 @@ Tests media_player component.
import logging
import unittest
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.const import (
STATE_OFF,
SERVICE_TURN_ON, SERVICE_TURN_OFF, SERVICE_VOLUME_UP, SERVICE_VOLUME_DOWN,

138
tests/components/test_mqtt.py Normal file
View File

@ -0,0 +1,138 @@
"""
tests.test_component_mqtt
~~~~~~~~~~~~~~~~~~~~~~~~~
Tests MQTT component.
"""
import unittest
from unittest import mock
import socket
import homeassistant.components.mqtt as mqtt
from homeassistant.const import (
EVENT_CALL_SERVICE, ATTR_DOMAIN, ATTR_SERVICE, EVENT_HOMEASSISTANT_START,
EVENT_HOMEASSISTANT_STOP)
from tests.common import (
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
class TestDemo(unittest.TestCase):
""" Test the demo module. """
def setUp(self): # pylint: disable=invalid-name
self.hass = get_test_home_assistant(1)
mock_mqtt_component(self.hass)
self.calls = []
def tearDown(self): # pylint: disable=invalid-name
""" Stop down stuff we started. """
self.hass.stop()
def record_calls(self, *args):
self.calls.append(args)
def test_client_starts_on_home_assistant_start(self):
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
self.hass.pool.block_till_done()
self.assertTrue(mqtt.MQTT_CLIENT.start.called)
def test_client_stops_on_home_assistant_start(self):
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
self.hass.pool.block_till_done()
self.hass.bus.fire(EVENT_HOMEASSISTANT_STOP)
self.hass.pool.block_till_done()
self.assertTrue(mqtt.MQTT_CLIENT.stop.called)
def test_setup_fails_if_no_broker_config(self):
self.assertFalse(mqtt.setup(self.hass, {mqtt.DOMAIN: {}}))
def test_setup_fails_if_no_connect_broker(self):
with mock.patch('homeassistant.components.mqtt.MQTT',
side_effect=socket.error()):
self.assertFalse(mqtt.setup(self.hass, {mqtt.DOMAIN: {
mqtt.CONF_BROKER: 'test-broker',
}}))
def test_publish_calls_service(self):
self.hass.bus.listen_once(EVENT_CALL_SERVICE, self.record_calls)
mqtt.publish(self.hass, 'test-topic', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('test-topic', self.calls[0][0].data[mqtt.ATTR_TOPIC])
self.assertEqual('test-payload', self.calls[0][0].data[mqtt.ATTR_PAYLOAD])
def test_service_call_without_topic_does_not_publush(self):
self.hass.bus.fire(EVENT_CALL_SERVICE, {
ATTR_DOMAIN: mqtt.DOMAIN,
ATTR_SERVICE: mqtt.SERVICE_PUBLISH
})
self.hass.pool.block_till_done()
self.assertTrue(not mqtt.MQTT_CLIENT.publish.called)
def test_subscribe_topic(self):
mqtt.subscribe(self.hass, 'test-topic', self.record_calls)
fire_mqtt_message(self.hass, 'test-topic', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('test-topic', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_not_match(self):
mqtt.subscribe(self.hass, 'test-topic', self.record_calls)
fire_mqtt_message(self.hass, 'another-test-topic', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))
def test_subscribe_topic_level_wildcard(self):
mqtt.subscribe(self.hass, 'test-topic/+/on', self.record_calls)
fire_mqtt_message(self.hass, 'test-topic/bier/on', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('test-topic/bier/on', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_level_wildcard_no_subtree_match(self):
mqtt.subscribe(self.hass, 'test-topic/+/on', self.record_calls)
fire_mqtt_message(self.hass, 'test-topic/bier', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))
def test_subscribe_topic_subtree_wildcard_subtree_topic(self):
mqtt.subscribe(self.hass, 'test-topic/#', self.record_calls)
fire_mqtt_message(self.hass, 'test-topic/bier/on', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('test-topic/bier/on', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_subtree_wildcard_root_topic(self):
mqtt.subscribe(self.hass, 'test-topic/#', self.record_calls)
fire_mqtt_message(self.hass, 'test-topic', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('test-topic', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_subtree_wildcard_no_match(self):
mqtt.subscribe(self.hass, 'test-topic/#', self.record_calls)
fire_mqtt_message(self.hass, 'another-test-topic', 'test-payload')
self.hass.pool.block_till_done()
self.assertEqual(0, len(self.calls))

2
tests/components/test_sun.py
View File

@ -10,7 +10,7 @@ from datetime import timedelta
from astral import Astral
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.util.dt as dt_util
import homeassistant.components.sun as sun

2
tests/helpers/test_entity.py
View File

@ -7,7 +7,7 @@ Tests the entity helper.
# pylint: disable=protected-access,too-many-public-methods
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.helpers.entity as entity
from homeassistant.const import ATTR_HIDDEN

2
tests/helpers/test_event.py
View File

@ -9,7 +9,7 @@ Tests event helpers.
import unittest
from datetime import datetime
import homeassistant as ha
import homeassistant.core as ha
from homeassistant.helpers.event import *

2
tests/helpers/test_init.py
View File

@ -9,7 +9,7 @@ import unittest
from common import get_test_home_assistant
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.loader as loader
from homeassistant.const import STATE_ON, STATE_OFF, ATTR_ENTITY_ID
from homeassistant.helpers import extract_entity_ids

41
tests/test_bootstrap.py Normal file
View File

@ -0,0 +1,41 @@
"""
tests.test_bootstrap
~~~~~~~~~~~~~~~~~~~~
Tests bootstrap.
"""
# pylint: disable=too-many-public-methods,protected-access
import tempfile
import unittest
from unittest import mock
from homeassistant import bootstrap
import homeassistant.util.dt as dt_util
from tests.common import mock_detect_location_info
class TestBootstrap(unittest.TestCase):
""" Test the bootstrap utils. """
def setUp(self):
self.orig_timezone = dt_util.DEFAULT_TIME_ZONE
def tearDown(self):
dt_util.DEFAULT_TIME_ZONE = self.orig_timezone
def test_from_config_file(self):
components = ['browser', 'conversation', 'script']
with tempfile.NamedTemporaryFile() as fp:
for comp in components:
fp.write('{}:\n'.format(comp).encode('utf-8'))
fp.flush()
with mock.patch('homeassistant.util.location.detect_location_info',
mock_detect_location_info):
hass = bootstrap.from_config_file(fp.name)
components.append('group')
self.assertEqual(sorted(components),
sorted(hass.config.components))

22
tests/test_config.py
View File

@ -9,14 +9,13 @@ import unittest
import unittest.mock as mock
import os
from homeassistant import DOMAIN, HomeAssistantError
import homeassistant.util.location as location_util
from homeassistant.core import DOMAIN, HomeAssistantError
import homeassistant.config as config_util
from homeassistant.const import (
CONF_LATITUDE, CONF_LONGITUDE, CONF_TEMPERATURE_UNIT, CONF_NAME,
CONF_TIME_ZONE)
from common import get_test_config_dir
from common import get_test_config_dir, mock_detect_location_info
CONFIG_DIR = get_test_config_dir()
YAML_PATH = os.path.join(CONFIG_DIR, config_util.YAML_CONFIG_FILE)
@ -28,23 +27,6 @@ def create_file(path):
pass
def mock_detect_location_info():
""" Mock implementation of util.detect_location_info. """
return location_util.LocationInfo(
ip='1.1.1.1',
country_code='US',
country_name='United States',
region_code='CA',
region_name='California',
city='San Diego',
zip_code='92122',
time_zone='America/Los_Angeles',
latitude='2.0',
longitude='1.0',
use_fahrenheit=True,
)
class TestConfig(unittest.TestCase):
""" Test the config utils. """

2
tests/test_init.py → tests/test_core.py
View File

@ -15,7 +15,7 @@ from datetime import datetime
import pytz
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.util.dt as dt_util
from homeassistant.helpers.event import track_state_change
from homeassistant.const import (

2
tests/test_remote.py
View File

@ -9,7 +9,7 @@ Uses port 8125 as a port that nothing runs on
# pylint: disable=protected-access,too-many-public-methods
import unittest
import homeassistant as ha
import homeassistant.core as ha
import homeassistant.bootstrap as bootstrap
import homeassistant.remote as remote
import homeassistant.components.http as http