Permalink
Browse files

Device, Area and event support

Why:

 * HA is exposing more and more functionallity tied to Device and
   Area registries. Nice to be able to edit areas especially without
   requiring use of browser UI.
 * Monitoring events can be tedious. Lets have a way to monitor.

This change addreses the need by:

 * add initial basic support of access websocket api.
   Currently no sharing of connection. One connection
   made per request, similar to rest api.
 * Add `device list` and `assign` command for devices.
   - list takes a match/filter similar to `entity list`
     i.e. `hass-cli device list 'Kitchen.*Light'
   - assign takes an area name or id together with
     one or more devices specificed by id/name or match.
     i.e. `device assign "Kitchen" --match "Kitchen" 'Ceiling Lamp' 32342334
 * Autocompletion for area names.
 * fixes for setuptools bug that cause build failures
   when dist folder present.
 * Finally a crude but working `entity watch` that simply just
   listen to event bus and print them out. Takes optional argument
   for event types.
  • Loading branch information...
maxandersen committed Feb 9, 2019
1 parent b7f8764 commit 237ade81372d25bfb3655c6a9f10d4aa697cad2e
@@ -170,3 +170,29 @@ def api_methods(
completions.sort()

return [c for c in completions if incomplete in c[0]]


def _quoteifneeded(val: str) -> str:
if val and ' ' in val:
return '"{}"'.format(val)
return val


def areas(
ctx: Configuration, args: List, incomplete: str
) -> List[Tuple[str, str]]:
"""Areas."""
_init_ctx(ctx)
allareas = api.get_areas(ctx)

completions = [] # type List[Tuple[str, str]]

if allareas:
for area in allareas:
completions.append((_quoteifneeded(area['name']), area['area_id']))

completions.sort()

return [c for c in completions if incomplete in c[0]]

return completions
@@ -453,3 +453,12 @@
# Static list of entities that will never be exposed to
# cloud, alexa, or google_home components
CLOUD_NEVER_EXPOSED_ENTITIES = ['group.all_locks']


# Websocket API
WS_TYPE_DEVICE_REGISTRY_LIST = 'config/device_registry/list'
WS_TYPE_AREA_REGISTRY_LIST = 'config/area_registry/list'
WS_TYPE_AREA_REGISTRY_CREATE = 'config/area_registry/create'
WS_TYPE_AREA_REGISTRY_DELETE = 'config/area_registry/delete'
WS_TYPE_AREA_REGISTRY_UPDATE = 'config/area_registry/update'
WS_TYPE_DEVICE_REGISTRY_UPDATE = 'config/device_registry/update'
@@ -0,0 +1,132 @@
"""Area (registry) plugin for Home Assistant CLI (hass-cli)."""
import logging
import re
import sys
from typing import Any, Dict, List, Pattern # noqa

import click
import homeassistant_cli.autocompletion as autocompletion
from homeassistant_cli.cli import pass_context
from homeassistant_cli.config import Configuration
import homeassistant_cli.const as const
import homeassistant_cli.helper as helper
import homeassistant_cli.remote as api

_LOGGING = logging.getLogger(__name__)


@click.group('area')
@pass_context
def cli(ctx):
"""Get info and operate on areas from Home Assistant (EXPERIMENTAL)."""


@cli.command('list')
@click.argument('areafilter', default=".*", required=False)
@pass_context
def listcmd(ctx: Configuration, areafilter: str):
"""List all areas from Home Assistant."""
ctx.auto_output("table")

areas = api.get_areas(ctx)

result = [] # type: List[Dict]
if areafilter == ".*":
result = areas
else:
areafilterre = re.compile(areafilter) # type: Pattern

for device in areas:
if areafilterre.search(device['name']):
result.append(device)

cols = [('ID', 'area_id'), ('NAME', 'name')]

ctx.echo(
helper.format_output(
ctx, result, columns=ctx.columns if ctx.columns else cols
)
)


@cli.command('create')
@click.argument('names', nargs=-1, required=True)
@pass_context
def create(ctx, names):
"""Create an area.
NAMES - one or more area names to create
"""
ctx.auto_output("data")

for name in names:
result = api.create_area(ctx, name)

ctx.echo(
helper.format_output(
ctx,
[result],
columns=ctx.columns if ctx.columns else const.COLUMNS_DEFAULT,
)
)


@cli.command('delete')
@click.argument( # type: ignore
'names', nargs=-1, required=True, autocompletion=autocompletion.areas
)
@pass_context
def delete(ctx, names):
"""Delete an area.
NAMES - one or more area names or id to delete
"""
ctx.auto_output("data")
excode = 0

for name in names:
area = api.find_area(ctx, name)
if not area:
_LOGGING.error("Could not find area with id or name: %s", name)
excode = 1
else:
result = api.delete_area(ctx, area['area_id'])

ctx.echo(
helper.format_output(
ctx,
[result],
columns=ctx.columns
if ctx.columns
else const.COLUMNS_DEFAULT,
)
)

if excode != 0:
sys.exit(excode)


@cli.command('rename')
@click.argument( # type: ignore
'oldname', required=True, autocompletion=autocompletion.areas
)
@click.argument('newname', required=True)
@pass_context
def rename(ctx, oldname, newname):
"""Rename an area."""
ctx.auto_output("data")

area = api.find_area(ctx, oldname)
if not area:
_LOGGING.error("Could not find area with id or name: %s", oldname)
sys.exit(1)

result = api.rename_area(ctx, area['area_id'], newname)

ctx.echo(
helper.format_output(
ctx,
[result],
columns=ctx.columns if ctx.columns else const.COLUMNS_DEFAULT,
)
)
@@ -0,0 +1,127 @@
"""Device (registry) plugin for Home Assistant CLI (hass-cli)."""
import logging
import re
import sys
from typing import Any, Dict, List, Optional, Pattern # noqa

import click
import homeassistant_cli.autocompletion as autocompletion
from homeassistant_cli.cli import pass_context
from homeassistant_cli.config import Configuration
import homeassistant_cli.helper as helper
import homeassistant_cli.remote as api

_LOGGING = logging.getLogger(__name__)


@click.group('device')
@pass_context
def cli(ctx):
"""Get info and operate on devices from Home Assistant (EXPERIMENTAL)."""


@cli.command('list')
@click.argument('devicefilter', default=".*", required=False)
@pass_context
def listcmd(ctx: Configuration, devicefilter: str):
"""List all devices from Home Assistant."""
ctx.auto_output("table")

devices = api.get_devices(ctx)

result = [] # type: List[Dict]
if devicefilter == ".*":
result = devices
else:
devicefilterre = re.compile(devicefilter) # type: Pattern

for device in devices:
if devicefilterre.search(device['name']):
result.append(device)

cols = [
('ID', 'id'),
('NAME', 'name'),
('MODEL', 'model'),
('MANUFACTURER', 'manufacturer'),
('AREA', 'area_id'),
]

ctx.echo(
helper.format_output(
ctx, result, columns=ctx.columns if ctx.columns else cols
)
)


@cli.command('assign')
@click.argument( # type: ignore
'area_id_or_name', required=True, autocompletion=autocompletion.areas
)
@click.argument('names', nargs=-1, required=False)
@click.option(
'--match', help="Expression used to find devices matching that name"
)
@pass_context
def assign(
ctx: Configuration,
area_id_or_name,
names: List[str],
match: Optional[str] = None,
):
"""Update area on one or more devices.
NAMES - one or more name or id (Optional)
"""
ctx.auto_output("data")

devices = api.get_devices(ctx)

result = [] # type: List[Dict]

area = api.find_area(ctx, area_id_or_name)
if not area:
_LOGGING.error(
"Could not find area with id or name: %s", area_id_or_name
)
sys.exit(1)

if match:
if match == ".*":
result = devices
else:
devicefilterre = re.compile(match) # type: Pattern

for device in devices:
if devicefilterre.search(device['name']):
result.append(device)

for id_or_name in names:
device = next( # type: ignore
(x for x in devices if x['id'] == id_or_name), None
)
if not device:
device = next( # type: ignore
(x for x in devices if x['name'] == id_or_name), None
)
if not device:
_LOGGING.error(
"Could not find device with id or name: %s", id_or_name
)
sys.exit(1)
result.append(device)

for device in result:
output = api.assign_area(ctx, device['id'], area['area_id'])
if output['success']:
ctx.echo(
"Successfully assigned '{}' to '{}'".format(
area['name'], device['name']
)
)
else:
_LOGGING.error(
"Failed to assign '%s' to '%s'", area['name'], device['name']
)

ctx.echo(str(output))
@@ -51,3 +51,19 @@ def fire(ctx: Configuration, event, json):

if response:
ctx.echo(raw_format_output(ctx.output, [response], ctx.yaml()))


@cli.command()
@click.argument('event_type', required=False)
@pass_context
def watch(ctx: Configuration, event_type):
"""Subscribe and print events.
EVENT-TYPE even type to subscribe to. if empty subscribe to all.
"""
frame = {'type': 'subscribe_events'}

if event_type:
frame['event_type'] = event_type

api.wsapi(ctx, frame, True)
Oops, something went wrong.

0 comments on commit 237ade8

Please sign in to comment.