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

Moved actor helper functions to observer package

Browse Source
This commit is contained in:
Paulus Schoutsen 2013-12-10 00:16:51 -08:00
parent d68f8ee309
commit 8a0ca534a2
3 changed files with 108 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
homeassistant/actors.py
View File

@ -18,13 +18,12 @@ import homeassistant as ha
import homeassistant.util as util
from homeassistant.observers import (
STATE_CATEGORY_SUN, SUN_STATE_BELOW_HORIZON, SUN_STATE_ABOVE_HORIZON,
STATE_ATTRIBUTE_NEXT_SUN_SETTING, STATE_ATTRIBUTE_NEXT_SUN_RISING,
is_sun_up, next_sun_setting,
STATE_CATEGORY_ALL_DEVICES, DEVICE_STATE_HOME, DEVICE_STATE_NOT_HOME,
STATE_CATEGORY_DEVICE_FORMAT,
STATE_CATEGORY_DEVICE_FORMAT, get_device_ids, is_device_home,
DOMAIN_LIGHT_CONTROL, SERVICE_TURN_LIGHT_ON, SERVICE_TURN_LIGHT_OFF,
STATE_CATEGORY_ALL_LIGHTS, STATE_CATEGORY_LIGHT_FORMAT, LIGHT_STATE_ON)
is_light_on, turn_light_on, turn_light_off, get_light_ids)
LIGHT_TRANSITION_TIME = timedelta(minutes=15)
@ -42,90 +41,15 @@ SERVICE_KEYBOARD_MEDIA_NEXT_TRACK = "media_next_track"
SERVICE_KEYBOARD_MEDIA_PREV_TRACK = "media_prev_track"
def is_sun_up(statemachine):
""" Returns if the sun is currently up based on the statemachine. """
return statemachine.is_state(STATE_CATEGORY_SUN, SUN_STATE_ABOVE_HORIZON)
def next_sun_setting(statemachine):
""" Returns the datetime object representing the next sun setting. """
state = statemachine.get_state(STATE_CATEGORY_SUN)
return None if not state else ha.str_to_datetime(
state['attributes'][STATE_ATTRIBUTE_NEXT_SUN_SETTING])
def next_sun_rising(statemachine):
""" Returns the datetime object representing the next sun setting. """
state = statemachine.get_state(STATE_CATEGORY_SUN)
return None if not state else ha.str_to_datetime(
state['attributes'][STATE_ATTRIBUTE_NEXT_SUN_RISING])
def is_device_home(statemachine, device_id=None):
""" Returns if any or specified device is home. """
category = STATE_CATEGORY_DEVICE_FORMAT.format(device_id) if device_id \
else STATE_CATEGORY_ALL_DEVICES
return statemachine.is_state(category, DEVICE_STATE_HOME)
def is_light_on(statemachine, light_id=None):
""" Returns if the lights are on based on the statemachine. """
category = STATE_CATEGORY_LIGHT_FORMAT.format(light_id) if light_id \
else STATE_CATEGORY_ALL_LIGHTS
return statemachine.is_state(category, LIGHT_STATE_ON)
def turn_light_on(bus, light_id=None, transition_seconds=None):
""" Turns all or specified light on. """
data = {}
if light_id:
data["light_id"] = light_id
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN_LIGHT_CONTROL, SERVICE_TURN_LIGHT_ON, data)
def turn_light_off(bus, light_id=None, transition_seconds=None):
""" Turns all or specified light off. """
data = {}
if light_id:
data["light_id"] = light_id
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN_LIGHT_CONTROL, SERVICE_TURN_LIGHT_OFF, data)
def get_light_count(statemachine):
""" Get the number of lights being tracked in the statemachine. """
return len(get_light_ids(statemachine))
def get_light_ids(statemachine):
""" Get the light IDs that are being tracked in the statemachine. """
lights_prefix = STATE_CATEGORY_LIGHT_FORMAT.format("")
light_id_part = slice(len(lights_prefix), None)
return [cat[light_id_part] for cat in statemachine.categories
if cat.startswith(lights_prefix)]
# pylint: disable=too-many-branches
def setup_device_light_triggers(bus, statemachine, device_state_categories):
def setup_device_light_triggers(bus, statemachine):
""" Triggers to turn lights on or off based on device precense. """
logger = logging.getLogger(__name__)
device_state_categories = [STATE_CATEGORY_DEVICE_FORMAT.format(device_id)
for device_id in get_device_ids(statemachine)]
if len(device_state_categories) == 0:
logger.error("LightTrigger:No devices given to track")
@ -141,7 +65,7 @@ def setup_device_light_triggers(bus, statemachine, device_state_categories):
# Calculates the time when to start fading lights in when sun sets
time_for_light_before_sun_set = lambda: \
(next_sun_setting(statemachine) - LIGHT_TRANSITION_TIME *
get_light_count(statemachine))
len(statemachine))
# pylint: disable=unused-argument
def handle_sun_rising(category, old_state, new_state):

14
homeassistant/bootstrap.py
View File

@ -52,14 +52,10 @@ def from_config_file(config_path):
# Device Tracker
if device_scanner:
device_tracker = observers.DeviceTracker(
bus, statemachine, device_scanner)
observers.DeviceTracker(bus, statemachine, device_scanner)
statusses.append(("Device Tracker", True))
else:
device_tracker = None
# Sun tracker
if config.has_option("common", "latitude") and \
config.has_option("common", "longitude"):
@ -95,13 +91,11 @@ def from_config_file(config_path):
observers.setup_light_control(bus, statemachine, light_control)
statusses.append(("Light Trigger", actors.setup_device_light_triggers(
bus, statemachine, device_tracker.device_state_categories)))
bus, statemachine)))
if config.has_option("downloader", "download_dir"):
result = actors.setup_file_downloader(
bus, config.get("downloader", "download_dir"))
statusses.append(("Downloader", result))
statusses.append(("Downloader", actors.setup_file_downloader(
bus, config.get("downloader", "download_dir"))))
statusses.append(("Webbrowser", actors.setup_webbrowser(bus)))

96
homeassistant/observers.py
View File

@ -63,6 +63,43 @@ LIGHTS_MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
KNOWN_DEVICES_FILE = "known_devices.csv"
def _get_grouped_states(statemachine, category_format_string):
""" Get states that are part of a group of states.
Example category_format_string can be devices.{}
If input states are devices, devices.paulus and devices.paulus.charging
then the output will be paulus.
"""
group_prefix = category_format_string.format("")
id_part = slice(len(group_prefix), None)
return [cat[id_part] for cat in statemachine.categories
if cat.startswith(group_prefix) and cat.count(".") == 1]
def is_sun_up(statemachine):
""" Returns if the sun is currently up based on the statemachine. """
return statemachine.is_state(STATE_CATEGORY_SUN, SUN_STATE_ABOVE_HORIZON)
def next_sun_setting(statemachine):
""" Returns the datetime object representing the next sun setting. """
state = statemachine.get_state(STATE_CATEGORY_SUN)
return None if not state else ha.str_to_datetime(
state['attributes'][STATE_ATTRIBUTE_NEXT_SUN_SETTING])
def next_sun_rising(statemachine):
""" Returns the datetime object representing the next sun setting. """
state = statemachine.get_state(STATE_CATEGORY_SUN)
return None if not state else ha.str_to_datetime(
state['attributes'][STATE_ATTRIBUTE_NEXT_SUN_RISING])
def track_sun(bus, statemachine, latitude, longitude):
""" Tracks the state of the sun. """
logger = logging.getLogger(__name__)
@ -113,6 +150,11 @@ def track_sun(bus, statemachine, latitude, longitude):
return True
def get_chromecast_ids(statemachine):
""" Gets the IDs of the different Chromecasts that are being tracked. """
return _get_grouped_states(statemachine, STATE_CATEGORY_CHROMECAST_FORMAT)
def setup_chromecast(bus, statemachine, host):
""" Listen for chromecast events. """
from homeassistant.packages import pychromecast
@ -152,6 +194,45 @@ def setup_chromecast(bus, statemachine, host):
return True
def is_light_on(statemachine, light_id=None):
""" Returns if the lights are on based on the statemachine. """
category = STATE_CATEGORY_LIGHT_FORMAT.format(light_id) if light_id \
else STATE_CATEGORY_ALL_LIGHTS
return statemachine.is_state(category, LIGHT_STATE_ON)
def turn_light_on(bus, light_id=None, transition_seconds=None):
""" Turns all or specified light on. """
data = {}
if light_id:
data["light_id"] = light_id
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN_LIGHT_CONTROL, SERVICE_TURN_LIGHT_ON, data)
def turn_light_off(bus, light_id=None, transition_seconds=None):
""" Turns all or specified light off. """
data = {}
if light_id:
data["light_id"] = light_id
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN_LIGHT_CONTROL, SERVICE_TURN_LIGHT_OFF, data)
def get_light_ids(statemachine):
""" Get the light IDs that are being tracked in the statemachine. """
return _get_grouped_states(statemachine, STATE_CATEGORY_LIGHT_FORMAT)
def setup_light_control(bus, statemachine, light_control):
""" Exposes light control via statemachine and services. """
@ -207,6 +288,19 @@ def setup_light_control(bus, statemachine, light_control):
return True
def get_device_ids(statemachine):
""" Returns the devices that are being tracked in the statemachine. """
return _get_grouped_states(statemachine, STATE_CATEGORY_DEVICE_FORMAT)
def is_device_home(statemachine, device_id=None):
""" Returns if any or specified device is home. """
category = STATE_CATEGORY_DEVICE_FORMAT.format(device_id) if device_id \
else STATE_CATEGORY_ALL_DEVICES
return statemachine.is_state(category, DEVICE_STATE_HOME)
class DeviceTracker(object):
""" Class that tracks which devices are home and which are not. """
@ -235,6 +329,8 @@ class DeviceTracker(object):
SERVICE_DEVICE_TRACKER_RELOAD,
lambda service: self._read_known_devices_file())
self.update_devices(device_scanner.scan_devices())
@property
def device_state_categories(self):
""" Returns a set containing all categories