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

Reorganized some core methods

Browse Source
This commit is contained in:
Paulus Schoutsen 2014-11-28 23:19:59 -08:00
parent 6f05548ec8
commit c08676aa81
11 changed files with 106 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
homeassistant/__init__.py
View File

@ -100,14 +100,6 @@ class HomeAssistant(object):
self.bus.fire(EVENT_CALL_SERVICE, event_data)
def get_entity_ids(self, domain_filter=None):
""" Returns known entity ids. """
if domain_filter:
return [entity_id for entity_id in self.states.entity_ids
if entity_id.startswith(domain_filter)]
else:
return self.states.entity_ids
def track_state_change(self, entity_ids, action,
from_state=None, to_state=None):
"""
@ -202,31 +194,6 @@ class HomeAssistant(object):
self.bus.listen(EVENT_TIME_CHANGED, time_listener)
def listen_once_event(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.
Note: at the moment it is impossible to remove a one time listener.
"""
@ft.wraps(listener)
def onetime_listener(event):
""" Removes listener from eventbus and then fires listener. """
if not hasattr(onetime_listener, 'run'):
# 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.bus.remove_listener(event_type, onetime_listener)
listener(event)
self.bus.listen(event_type, onetime_listener)
def stop(self):
""" Stops Home Assistant and shuts down all threads. """
_LOGGER.info("Stopping")
@ -238,6 +205,32 @@ class HomeAssistant(object):
self._pool.stop()
def get_entity_ids(self, domain_filter=None):
"""
Returns known entity ids.
THIS METHOD IS DEPRECATED. Use hass.states.entity_ids
"""
_LOGGER.warning(
"hass.get_entiy_ids is deprecated. Use hass.states.entity_ids")
return self.states.entity_ids(domain_filter)
def listen_once_event(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.
Note: at the moment it is impossible to remove a one time listener.
THIS METHOD IS DEPRECATED. Please use hass.events.listen_once.
"""
_LOGGER.warning(
"hass.listen_once_event is deprecated. Use hass.bus.listen_once")
self.bus.listen_once(event_type, listener)
def _process_match_param(parameter):
""" Wraps parameter in a list if it is not one and returns it. """
@ -390,6 +383,31 @@ class EventBus(object):
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.
Note: at the moment it is impossible to remove a one time listener.
"""
@ft.wraps(listener)
def onetime_listener(event):
""" Removes listener from eventbus and then fires listener. """
if not hasattr(onetime_listener, 'run'):
# 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)
def remove_listener(self, event_type, listener):
""" Removes a listener of a specific event_type. """
with self._lock:
@ -487,10 +505,13 @@ class StateMachine(object):
self._bus = bus
self._lock = threading.Lock()
@property
def entity_ids(self):
def entity_ids(self, domain_filter=None):
""" List of entity ids that are being tracked. """
return list(self._states.keys())
if domain_filter is not None:
return [entity_id for entity_id in self._states.keys()
if util.split_entity_id(entity_id)[0] == domain_filter]
else:
return list(self._states.keys())
def all(self):
""" Returns a list of all states. """
@ -619,14 +640,14 @@ class Timer(threading.Thread):
# every minute.
assert 60 % self.interval == 0, "60 % TIMER_INTERVAL should be 0!"
hass.listen_once_event(EVENT_HOMEASSISTANT_START,
lambda event: self.start())
hass.bus.listen_once(EVENT_HOMEASSISTANT_START,
lambda event: self.start())
def run(self):
""" Start the timer. """
self.hass.listen_once_event(EVENT_HOMEASSISTANT_STOP,
lambda event: self._stop.set())
self.hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP,
lambda event: self._stop.set())
_LOGGER.info("Timer:starting")

2
homeassistant/components/__init__.py
View File

@ -61,7 +61,7 @@ def is_on(hass, entity_id=None):
entity_ids = group.expand_entity_ids(hass, [entity_id])
else:
entity_ids = hass.states.entity_ids
entity_ids = hass.states.entity_ids()
for entity_id in entity_ids:
domain = util.split_entity_id(entity_id)[0]

2
homeassistant/components/chromecast.py
View File

@ -38,7 +38,7 @@ def is_on(hass, entity_id=None):
""" Returns true if specified ChromeCast entity_id is on.
Will check all chromecasts if no entity_id specified. """
entity_ids = [entity_id] if entity_id else hass.get_entity_ids(DOMAIN)
entity_ids = [entity_id] if entity_id else hass.states.entity_ids(DOMAIN)
return any(not hass.states.is_state(entity_id, STATE_NO_APP)
for entity_id in entity_ids)

4
homeassistant/components/demo.py
View File

@ -41,7 +41,7 @@ def setup(hass, config):
if service.data and ATTR_ENTITY_ID in service.data:
entity_ids = extract_entity_ids(hass, service)
else:
entity_ids = hass.get_entity_ids(service.domain)
entity_ids = hass.states.entity_ids(service.domain)
for entity_id in entity_ids:
domain, _ = split_entity_id(entity_id)
@ -59,7 +59,7 @@ def setup(hass, config):
if service.data and ATTR_ENTITY_ID in service.data:
entity_ids = extract_entity_ids(hass, service)
else:
entity_ids = hass.get_entity_ids(service.domain)
entity_ids = hass.states.entity_ids(service.domain)
for entity_id in entity_ids:
hass.states.set(entity_id, STATE_OFF)

2
homeassistant/components/device_sun_light_trigger.py
View File

@ -36,7 +36,7 @@ def setup(hass, config):
logger = logging.getLogger(__name__)
device_entity_ids = hass.get_entity_ids(device_tracker.DOMAIN)
device_entity_ids = hass.states.entity_ids(device_tracker.DOMAIN)
if not device_entity_ids:
logger.error("No devices found to track")

4
homeassistant/components/http/__init__.py
View File

@ -133,7 +133,7 @@ def setup(hass, config):
RequestHandler, hass, api_password,
development)
hass.listen_once_event(
hass.bus.listen_once(
ha.EVENT_HOMEASSISTANT_START,
lambda event:
threading.Thread(target=server.start, daemon=True).start())
@ -171,7 +171,7 @@ class HomeAssistantHTTPServer(ThreadingMixIn, HTTPServer):
def start(self):
""" Starts the server. """
self.hass.listen_once_event(
self.hass.bus.listen_once(
ha.EVENT_HOMEASSISTANT_STOP,
lambda event: self.shutdown())

9
test/test_component_demo.py
View File

@ -6,9 +6,6 @@ Tests demo component.
"""
# pylint: disable=too-many-public-methods,protected-access
import unittest
import datetime as dt
import ephem
import homeassistant as ha
import homeassistant.components.demo as demo
@ -33,7 +30,7 @@ class TestDemo(unittest.TestCase):
for domain in ('light', 'switch'):
# Focus on 1 entity
entity_id = self.hass.get_entity_ids(domain)[0]
entity_id = self.hass.states.entity_ids(domain)[0]
self.hass.call_service(
domain, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id})
@ -54,7 +51,7 @@ class TestDemo(unittest.TestCase):
self.hass._pool.block_till_done()
for entity_id in self.hass.get_entity_ids(domain):
for entity_id in self.hass.states.entity_ids(domain):
self.assertEqual(
STATE_ON, self.hass.states.get(entity_id).state)
@ -62,7 +59,7 @@ class TestDemo(unittest.TestCase):
self.hass._pool.block_till_done()
for entity_id in self.hass.get_entity_ids(domain):
for entity_id in self.hass.states.entity_ids(domain):
self.assertEqual(
STATE_OFF, self.hass.states.get(entity_id).state)

4
test/test_component_group.py
View File

@ -44,7 +44,7 @@ class TestComponentsGroup(unittest.TestCase):
""" Test setup_group method. """
# Test if group setup in our init mode is ok
self.assertIn(self.group_name, self.hass.states.entity_ids)
self.assertIn(self.group_name, self.hass.states.entity_ids())
group_state = self.hass.states.get(self.group_name)
self.assertEqual(comps.STATE_ON, group_state.state)
@ -73,7 +73,7 @@ class TestComponentsGroup(unittest.TestCase):
['light.Bowl', 'device_tracker.Paulus']))
# Try to setup a group with a non existing state
self.assertNotIn('non.existing', self.hass.states.entity_ids)
self.assertNotIn('non.existing', self.hass.states.entity_ids())
self.assertFalse(group.setup_group(
self.hass, 'light_and_nothing',
['light.Bowl', 'non.existing']))

8
test/test_component_http.py
View File

@ -176,8 +176,6 @@ class TestHTTP(unittest.TestCase):
def test_api_state_change_with_bad_data(self):
""" Test if API sends appropriate error if we omit state. """
new_state = "debug_state_change"
req = requests.post(
_url(remote.URL_API_STATES_ENTITY.format(
"test_entity.that_does_not_exist")),
@ -195,7 +193,7 @@ class TestHTTP(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
hass.listen_once_event("test.event_no_data", listener)
hass.bus.listen_once("test.event_no_data", listener)
requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test.event_no_data")),
@ -216,7 +214,7 @@ class TestHTTP(unittest.TestCase):
if "test" in event.data:
test_value.append(1)
hass.listen_once_event("test_event_with_data", listener)
hass.bus.listen_once("test_event_with_data", listener)
requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test_event_with_data")),
@ -236,7 +234,7 @@ class TestHTTP(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
hass.listen_once_event("test_event_bad_data", listener)
hass.bus.listen_once("test_event_bad_data", listener)
req = requests.post(
_url(remote.URL_API_EVENTS_EVENT.format("test_event_bad_data")),

56
test/test_core.py
View File

@ -64,17 +64,6 @@ class TestHomeAssistant(unittest.TestCase):
self.assertFalse(blocking_thread.is_alive())
def test_get_entity_ids(self):
""" Test get_entity_ids method. """
ent_ids = self.hass.get_entity_ids()
self.assertEqual(2, len(ent_ids))
self.assertTrue('light.Bowl' in ent_ids)
self.assertTrue('switch.AC' in ent_ids)
ent_ids = self.hass.get_entity_ids('light')
self.assertEqual(1, len(ent_ids))
self.assertTrue('light.Bowl' in ent_ids)
def test_track_state_change(self):
""" Test track_state_change. """
# 2 lists to track how often our callbacks got called
@ -112,21 +101,6 @@ class TestHomeAssistant(unittest.TestCase):
self.assertEqual(1, len(specific_runs))
self.assertEqual(3, len(wildcard_runs))
def test_listen_once_event(self):
""" Test listen_once_event method. """
runs = []
self.hass.listen_once_event('test_event', lambda x: runs.append(1))
self.hass.bus.fire('test_event')
self.hass._pool.block_till_done()
self.assertEqual(1, len(runs))
# Second time it should not increase runs
self.hass.bus.fire('test_event')
self.hass._pool.block_till_done()
self.assertEqual(1, len(runs))
def test_track_point_in_time(self):
""" Test track point in time. """
before_birthday = datetime(1985, 7, 9, 12, 0, 0)
@ -234,6 +208,21 @@ class TestEventBus(unittest.TestCase):
# Try deleting listener while category doesn't exist either
self.bus.remove_listener('test', listener)
def test_listen_once_event(self):
""" Test listen_once_event method. """
runs = []
self.bus.listen_once('test_event', lambda x: runs.append(1))
self.bus.fire('test_event')
self.bus._pool.block_till_done()
self.assertEqual(1, len(runs))
# Second time it should not increase runs
self.bus.fire('test_event')
self.bus._pool.block_till_done()
self.assertEqual(1, len(runs))
class TestState(unittest.TestCase):
""" Test EventBus methods. """
@ -276,11 +265,22 @@ class TestStateMachine(unittest.TestCase):
self.assertFalse(self.states.is_state('light.Bowl', 'off'))
self.assertFalse(self.states.is_state('light.Non_existing', 'on'))
def test_entity_ids(self):
""" Test get_entity_ids method. """
ent_ids = self.states.entity_ids()
self.assertEqual(2, len(ent_ids))
self.assertTrue('light.Bowl' in ent_ids)
self.assertTrue('switch.AC' in ent_ids)
ent_ids = self.states.entity_ids('light')
self.assertEqual(1, len(ent_ids))
self.assertTrue('light.Bowl' in ent_ids)
def test_remove(self):
""" Test remove method. """
self.assertTrue('light.Bowl' in self.states.entity_ids)
self.assertTrue('light.Bowl' in self.states.entity_ids())
self.assertTrue(self.states.remove('light.Bowl'))
self.assertFalse('light.Bowl' in self.states.entity_ids)
self.assertFalse('light.Bowl' in self.states.entity_ids())
# If it does not exist, we should get False
self.assertFalse(self.states.remove('light.Bowl'))

4
test/test_remote.py
View File

@ -96,7 +96,7 @@ class TestRemoteMethods(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
hass.listen_once_event("test.event_no_data", listener)
hass.bus.listen_once("test.event_no_data", listener)
remote.fire_event(master_api, "test.event_no_data")
@ -217,7 +217,7 @@ class TestRemoteClasses(unittest.TestCase):
""" Helper method that will verify our event got called. """
test_value.append(1)
slave.listen_once_event("test.event_no_data", listener)
slave.bus.listen_once("test.event_no_data", listener)
slave.bus.fire("test.event_no_data")