dependencies module

Action Engine

ActionEngine for HomeControl

class homecontrol.dependencies.action_engine.ActionEngine(item: homecontrol.dependencies.entity_types.Item, core)

Bases: object

Holds available actions for an item

coroutine execute(name: str, *args, **kwargs) → bool

Executes an action, optionally with parameters

homecontrol.dependencies.action_engine.action(arg: Union[Callable, str]) → Callable

Decorator to mark a coroutine as an action

>>> @action
>>> async def foo(): ...

>>> @action(name)
>>> async def foo(): ...

Configuration Manager

config_manager module

class homecontrol.dependencies.config_manager.ConfigManager(cfg: dict, cfg_path: str)

Bases: object

Manages the configuration with configuration domains

coroutine approve_domain_config(domain: str, config: dict, initial: bool = True) → dict

Returns an approved and validated version of config for a domain

get(key, default=None)

getter for self.cfg

coroutine register_domain(domain: str, handler: object = None, schema: voluptuous.schema_builder.Schema = None, allow_reload: bool = False, default: dict = None) → object

Registers a configuration domain

Objects can register themselves to their own configuration domain and subscribe to changes in the configuration

Parameters:

• domain (str) – The configuration domain (A top-level key in config.yaml)
• handler (object) – The object subscribing to this domain. Methods it can implement are: - approve_configuration(domain, config) -> bool: - apply_new_configuration(domain, config) -> None: If not specified then your configuration domain will not be reloadable

allow_reload (bool):

Allow reloading this configuration domain

schema (voluptuous.Schema):

Schema to validate the configuration and to fill in defaults

default (dict):

A default configuration

Returns:A validated and approved version of the configuration

coroutine reload_config() → None

Reloads the configuration and updates where it can

Data Types

data types for HomeControl that can be translated to JSON for API usage

class homecontrol.dependencies.data_types.Color(h: int, s: int, l: int)

Bases: object

Representation for a color

dump() -> (<class 'int'>, <class 'int'>, <class 'int'>)

Dumps the Color into a JSON serialisable format

staticmethod from_data(hsl: tuple)

Constructor from the data received through the API or configuration

staticmethod from_hsl(hsl: tuple)

HSL constructor

staticmethod from_rgb(rgb: tuple)

RGB constructor

h

Hue

l

Lightness

rgb

RGB

s

Saturation

class homecontrol.dependencies.data_types.DateTime

Bases: datetime.datetime

date time format

dump()

Dump to JSON serialisable data

staticmethod from_data(data)

Construct from JSON serialisable data

Entity Types

Module containing the entity types Every new Item or Module will get one of these classes as a base class

class homecontrol.dependencies.entity_types.Item

Bases: object

A dummy Item

config_schema = <Schema(<class 'object'>, extra=PREVENT_EXTRA, required=False) object>

coroutine constructor(identifier: str, name: str, cfg: dict, state_defaults: dict, core: Core, unique_identifier: str = None) → Item

Constructs an item

coroutine init() → None

Default init method

status = 'offline'

coroutine stop() → None

Default stop method

update_status(status: homecontrol.const.ItemStatus) → None

Updates the item status and broadcasts an event

class homecontrol.dependencies.entity_types.Module

Bases: object

A dummy Module

coroutine init() → None

Default init method

coroutine stop() → None

Default stop method

homecontrol.dependencies.entity_types.ModuleDef

alias of homecontrol.dependencies.entity_types.Module

Event Engine

EventEngine for HomeControl

class homecontrol.dependencies.event_engine.Event(event_type: str, data: dict = None, timestamp: datetime.datetime = None, kwargs: dict = None)

Bases: object

Representation for an Event

data

event_type

kwargs

timestamp

class homecontrol.dependencies.event_engine.EventEngine(core)

Bases: object

Dispatcher for events

broadcast(event_type: str, data: dict = None, **kwargs) → List[_asyncio.Future]

Broadcast an event and return the futures

Every listener is a coroutine that will simply receive event and kwargs

Example: >>> async def on_event(event: Event, …): >>> return

broadcast_threaded(event_type: str, data: dict = None, **kwargs) → List[_asyncio.Task]

Same as broadcast BUT - It returns Futures and not Tasks - It uses threads

staticmethod create_event(event_type: str, data: dict = None, **kwargs) → homecontrol.dependencies.event_engine.Event

Creates an Event to be broadcasted

coroutine gather(event_type: str, data: dict = None, timeout: Union[float, int, None] = None, **kwargs) → List[Any]

Broadcast an event and return the results

get_event_handlers(event: homecontrol.dependencies.event_engine.Event) → List[T]

Returns a list of handlers for an Event

register(event: str) → Callable

Decorator to register event handlers

remove_handler(event: str, handler: Callable) → None

Removes an event handler

Item Manager

ItemManager for HomeControl

class homecontrol.dependencies.item_manager.ItemManager(core)

Bases: object

ItemManager manages all your stateful items

coroutine add_from_module(mod_obj: homecontrol.dependencies.entity_types.Module) → None

Adds the item specifications of a module to the dict of available ones

mod_obj: homecontrol.entity_types.Module

coroutine create_from_storage_entry(storage_entry: homecontrol.dependencies.item_manager.StorageEntry) → homecontrol.dependencies.entity_types.Item

Creates an Item from a storage entry

coroutine create_item(identifier: str, item_type: str, cfg: dict = None, state_defaults: dict = None, name: str = None, unique_identifier: str = None) → homecontrol.dependencies.entity_types.Item

Creates a HomeControl item

get_by_unique_identifier(unique_identifier: str) → homecontrol.dependencies.entity_types.Item

Returns an item by its unique identifier

get_item(identifier: str) → homecontrol.dependencies.entity_types.Item

Returns an item by identifier or unique_identifier

get_storage_entry(unique_identifier: str) → homecontrol.dependencies.item_manager.StorageEntry

Returns the StorageEntry for a unique_identifier

coroutine init() → None

Initialise the items from configuration

coroutine init_item(item: homecontrol.dependencies.entity_types.Item) → None

Initialises an item

for ... in iter_items_by_id(iterable) → [<class 'homecontrol.dependencies.entity_types.Item'>]

Translates item identifiers into item instances

load_yaml_config() → None

Loads the YAML configuration

coroutine register_entry(storage_entry: homecontrol.dependencies.item_manager.StorageEntry, override: bool = False) → Optional[homecontrol.dependencies.entity_types.Item]

Registers a storage entry

coroutine remove_item(identifier: str) → None

Removes a HomeControl item

identifier: str

The item’s identifier

coroutine stop() → None

Stops every item

coroutine stop_item(item: homecontrol.dependencies.entity_types.Item, status: homecontrol.const.ItemStatus = <ItemStatus.STOPPED: 'stopped'>) → None

Stops an item

update_storage_entry(entry: homecontrol.dependencies.item_manager.StorageEntry) → None

Updates a config storage entry

class homecontrol.dependencies.item_manager.StorageEntry(unique_identifier: str, type: str, provider: str, enabled: bool = True, identifier: str = None, state_defaults: dict = NOTHING, cfg: dict = NOTHING, name: str = None, hidden: bool = False)

Bases: object

The storage representation of an item

cfg

enabled

hidden

identifier

name

provider

state_defaults

type

unique_identifier

homecontrol.dependencies.item_manager.yaml_entry_to_storage_entry(yaml_entry: dict) → homecontrol.dependencies.item_manager.StorageEntry

Converts a yaml entry to a JSON entry

JSON Response

JSONResponse module

class homecontrol.dependencies.json_response.JSONResponse(data: Any = None, error: str = None, status_code: int = 200, core=None, headers: dict = None)

Bases: aiohttp.web_response.Response

A HTTP response for JSON data

JSON

JSON Encoder and Decoder

class homecontrol.dependencies.json.JSONEncoder(core: Core, *args, **kwargs)

Bases: json.encoder.JSONEncoder

Custom JSONEncoder that also parses HomeControl types

default(o)

Encode custom types

homecontrol.dependencies.json.dump(obj, fp, *, indent=None, sort_keys=False, core: Core = None, **kw)

Dumps an object into a Writer with support for HomeControl’s data types

homecontrol.dependencies.json.dumps(obj, *, indent=None, sort_keys=False, core: Core = None, **kw)

Dumps an object into a JSON string with support for HomeControl’s data types

Module Manager

ModuleManager module

class homecontrol.dependencies.module_manager.ModuleAccessor(module_manager: homecontrol.dependencies.module_manager.ModuleManager)

Bases: object

Wrapper for ModuleManager.loaded_modules

class homecontrol.dependencies.module_manager.ModuleFolder(name: str)

Bases: object

module folder representation to create a dummy package in sys.modules

class homecontrol.dependencies.module_manager.ModuleManager(core: Core)

Bases: object

Manages your modules

coroutine init() → None

Initialise the modules

coroutine load_file_module(mod_path: str, name: str) -> (<class 'homecontrol.dependencies.entity_types.Module'>, <class 'Exception'>)

Loads a module from a file and initialises it

Returns a Module object

coroutine load_folder(path: str) → [<class 'object'>]

Load every module in a folder

coroutine load_folder_module(path: str, name: str) -> (<class 'homecontrol.dependencies.entity_types.Module'>, <class 'Exception'>)

Loads a module from a folder and initialises it

It also takes care of pip requirements

Returns a Module object

resource_path(module: homecontrol.dependencies.entity_types.Module, path: str = '') → str

Returns the path for a module’s resource folder Note that only folder modules can have a resource path

coroutine stop() → None

Unloads all modules to prepare for a shutdown

State Engine

Throttle Function

throttle_function

class homecontrol.dependencies.throttle_function.throttle(s: float = 1)

Bases: object

Throttles a function so it only gets called at a fixed frequency

Tick Engine

Validators

Config validators

class homecontrol.dependencies.validators.IsItem(core: Core, msg: str = None)

Bases: object

A validator to get an item

YAML Loader

Provides a YAML loader

class homecontrol.dependencies.yaml_loader.Constructor

Bases: yaml.constructor.SafeConstructor

Constructor for yaml

env_var_constructor(node: yaml.nodes.Node) → str

Embeds an environment variable !env_var <name> [default]

format_string_constructor(node: yaml.nodes.Node = None) → str

Renders a format string .. rubric:: Example

!format { template: “Hello {who}”, who: You }

include_dir_file_mapped_constructor(node: yaml.nodes.Node = None) → dict

!include_dir_file_mapped <folder>

Loads multiple files from a folder and maps their contents to their filenames

include_file_constructor(node: yaml.nodes.Node = None) → object

!include <path> ~/ for paths relative to your home directory / for absolute paths anything else for paths relative to your config folder

include_merge_constructor(node: yaml.nodes.Node = None) -> (<class 'list'>, <class 'dict'>)

!include <file|folder> …

Merges file or folder contents

This constructor only works if all the files’ contents are of same type and if this type is either list or dict.

listdir_constructor(node: yaml.nodes.Node) → list

!listdir <path>

Returns the contents of a directory

path_constructor(node: yaml.nodes.Node) → str

!path <path> ~/ for paths relative to your home directory / for absolute paths anything else for paths relative to your config folder

class homecontrol.dependencies.yaml_loader.YAMLLoader(stream, cfg_folder: str = None)

Bases: yaml.reader.Reader, yaml.scanner.Scanner, yaml.parser.Parser, yaml.composer.Composer, homecontrol.dependencies.yaml_loader.Constructor, yaml.resolver.Resolver

Loads YAML with custom constructors

classmethod load(data, cfg_folder: str = None)

Loads data