import asyncio
import aiohttp
from threading import RLock
from typing import Optional, Dict, List, Union
from platypush.plugins import action
from platypush.plugins.switch import SwitchPlugin
class SmartthingsPlugin(SwitchPlugin):
"""
Plugin to interact with devices and locations registered to a Samsung SmartThings account.
Requires:
* **pysmartthings** (``pip install pysmartthings``)
"""
_timeout = aiohttp.ClientTimeout(total=20.)
def __init__(self, access_token: str, **kwargs):
"""
:param access_token: SmartThings API access token - you can get one at https://account.smartthings.com/tokens.
"""
super().__init__(**kwargs)
self._access_token = access_token
self._refresh_lock = RLock()
self._execute_lock = RLock()
self._locations = []
self._devices = []
self._rooms_by_location = {}
self._locations_by_id = {}
self._locations_by_name = {}
self._devices_by_id = {}
self._devices_by_name = {}
self._rooms_by_id = {}
self._rooms_by_location_and_id = {}
self._rooms_by_location_and_name = {}
async def _refresh_locations(self, api):
self._locations = await api.locations()
self._locations_by_id = {
loc.location_id: loc
for loc in self._locations
}
self._locations_by_name = {
loc.name: loc
for loc in self._locations
}
async def _refresh_devices(self, api):
self._devices = await api.devices()
self._devices_by_id = {
dev.device_id: dev
for dev in self._devices
}
self._devices_by_name = {
dev.label: dev
for dev in self._devices
}
async def _refresh_rooms(self, api, location_id: str):
self._rooms_by_location[location_id] = await api.rooms(location_id=location_id)
self._rooms_by_id.update(**{
room.room_id: room
for room in self._rooms_by_location[location_id]
})
self._rooms_by_location_and_id[location_id] = {
room.room_id: room
for room in self._rooms_by_location[location_id]
}
self._rooms_by_location_and_name[location_id] = {
room.name: room
for room in self._rooms_by_location[location_id]
}
async def _refresh_info(self):
import pysmartthings
async with aiohttp.ClientSession(timeout=self._timeout) as session:
api = pysmartthings.SmartThings(session, self._access_token)
tasks = [
asyncio.ensure_future(self._refresh_locations(api)),
asyncio.ensure_future(self._refresh_devices(api)),
]
await asyncio.gather(*tasks)
room_tasks = [
asyncio.ensure_future(self._refresh_rooms(api, location.location_id))
for location in self._locations
]
await asyncio.gather(*room_tasks)
def refresh_info(self):
with self._refresh_lock:
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
loop.run_until_complete(self._refresh_info())
finally:
loop.stop()
def _location_to_dict(self, location) -> Dict:
return {
'name': location.name,
'location_id': location.location_id,
'country_code': location.country_code,
'locale': location.locale,
'latitude': location.latitude,
'longitude': location.longitude,
'temperature_scale': location.temperature_scale,
'region_radius': location.region_radius,
'timezone_id': location.timezone_id,
'rooms': {
room.room_id: self._room_to_dict(room)
for room in self._rooms_by_location.get(location.location_id, {})
}
}
@staticmethod
def _device_to_dict(device) -> Dict:
return {
'capabilities': device.capabilities,
'name': device.label,
'device_id': device.device_id,
'location_id': device.location_id,
'room_id': device.room_id,
'device_type_id': device.device_type_id,
'device_type_name': device.device_type_name,
'device_type_network': device.device_type_network,
}
@staticmethod
def _room_to_dict(room) -> Dict:
return {
'name': room.name,
'background_image': room.background_image,
'room_id': room.room_id,
'location_id': room.location_id,
}
@action
def info(self) -> Dict[str, Dict[str, dict]]:
"""
Return the objects registered to the account, including locations and devices.
.. code-block:: json
{
"devices": {
"smart-tv-id": {
"capabilities": [
"ocf",
"switch",
"audioVolume",
"audioMute",
"tvChannel",
"mediaInputSource",
"mediaPlayback",
"mediaTrackControl",
"custom.error",
"custom.picturemode",
"custom.soundmode",
"custom.accessibility",
"custom.launchapp",
"custom.recording",
"custom.tvsearch",
"custom.disabledCapabilities",
"samsungvd.ambient",
"samsungvd.ambientContent",
"samsungvd.ambient18",
"samsungvd.mediaInputSource",
"refresh",
"execute",
"samsungvd.firmwareVersion",
"samsungvd.supportsPowerOnByOcf"
],
"device_id": "smart-tv-id",
"device_type_id": null,
"device_type_name": null,
"device_type_network": null,
"location_id": "location-id",
"name": "Samsung Smart TV",
"room_id": "room-1"
},
"tv-switch-id": {
"capabilities": [
"switch",
"refresh",
"healthCheck"
],
"device_id": "tv-switch-id",
"device_type_id": null,
"device_type_name": null,
"device_type_network": null,
"location_id": "location-id",
"name": "TV Smart Switch",
"room_id": "room-1"
},
"lights-switch-id": {
"capabilities": [
"switch",
"refresh",
"healthCheck"
],
"device_id": "lights-switch-id",
"device_type_id": null,
"device_type_name": null,
"device_type_network": null,
"location_id": "location-id",
"name": "Lights Switch",
"room_id": "room-2"
}
},
"locations": {
"location-id": {
"name": "My home",
"location_id": "location-id",
"country_code": "us",
"locale": "en-US",
"latitude": "latitude",
"longitude": "longitude",
"temperature_scale": null,
"region_radius": null,
"timezone_id": null,
"rooms": {
"room-1": {
"background_image": null,
"location_id": "location-1",
"name": "Living Room",
"room_id": "room-1"
},
"room-2": {
"background_image": null,
"location_id": "location-1",
"name": "Bedroom",
"room_id": "room-2"
}
}
}
}
}
"""
self.refresh_info()
return {
'locations': {loc.location_id: self._location_to_dict(loc) for loc in self._locations},
'devices': {dev.device_id: self._device_to_dict(dev) for dev in self._devices},
}
@action
def get_location(self, location_id: Optional[str] = None, name: Optional[str] = None) -> dict:
"""
Get the info of a location by ID or name.
.. code-block:: json
{
"name": "My home",
"location_id": "location-id",
"country_code": "us",
"locale": "en-US",
"latitude": "latitude",
"longitude": "longitude",
"temperature_scale": null,
"region_radius": null,
"timezone_id": null,
"rooms": {
"room-1": {
"background_image": null,
"location_id": "location-1",
"name": "Living Room",
"room_id": "room-1"
},
"room-2": {
"background_image": null,
"location_id": "location-1",
"name": "Bedroom",
"room_id": "room-2"
}
}
}
"""
assert location_id or name, 'Specify either location_id or name'
if location_id not in self._locations_by_id or name not in self._locations_by_name:
self.refresh_info()
location = self._locations_by_id.get(location_id, self._locations_by_name.get(name))
assert location, 'Location {} not found'.format(location_id or name)
return self._location_to_dict(location)
def _get_device(self, device: str):
if device not in self._devices_by_id or device not in self._devices_by_name:
self.refresh_info()
device = self._devices_by_id.get(device, self._devices_by_name.get(device))
assert device, 'Device {} not found'.format(device)
return device
@action
def get_device(self, device: str) -> dict:
"""
Get a device info by ID or name.
:param device: Device ID or name.
:return:
.. code-block:: json
"tv-switch-id": {
"capabilities": [
"switch",
"refresh",
"healthCheck"
],
"device_id": "tv-switch-id",
"device_type_id": null,
"device_type_name": null,
"device_type_network": null,
"location_id": "location-id",
"name": "TV Smart Switch",
"room_id": "room-1"
}
"""
device = self._get_device(device)
return self._device_to_dict(device)
async def _execute(self, device_id: str, capability: str, command, component_id: str, args: Optional[list]):
import pysmartthings
async with aiohttp.ClientSession(timeout=self._timeout) as session:
api = pysmartthings.SmartThings(session, self._access_token)
device = await api.device(device_id)
ret = await device.command(component_id=component_id, capability=capability, command=command, args=args)
assert ret, 'The command {capability}={command} failed on device {device}'.format(
capability=capability, command=command, device=device_id)
@action
def execute(self,
device: str,
capability: str,
command,
component_id: str = 'main',
args: Optional[list] = None):
"""
Execute a command on a device.
Example request to turn on a device with ``switch`` capability:
.. code-block:: json
{
"type": "request",
"action": "smartthings.execute",
"args": {
"device": "My Switch",
"capability": "switch",
"command": "on"
}
}
:param device: Device ID or name.
:param capability: Property to be read/written (see device ``capabilities`` returned from :meth:`.get_device`).
:param command: Command to execute on the ``capability``
(see https://smartthings.developer.samsung.com/docs/api-ref/capabilities.html).
:param component_id: ID of the component to execute the command on (default: ``main``, i.e. the device itself).
:param args: Command extra arguments, as a list.
"""
device = self._get_device(device)
with self._execute_lock:
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
loop.run_until_complete(self._execute(
device_id=device.device_id, capability=capability, command=command,
component_id=component_id, args=args))
finally:
loop.stop()
@staticmethod
async def _get_device_status(api, device_id: str) -> dict:
device = await api.device(device_id)
await device.status.refresh()
return {
'device_id': device_id,
'name': device.label,
**{
cap: getattr(device.status, cap)
for cap in device.capabilities
if hasattr(device.status, cap)
and not callable(getattr(device.status, cap))
}
}
async def _refresh_status(self, devices: List[str]) -> List[dict]:
import pysmartthings
device_ids = []
missing_device_ids = set()
def parse_device_id(device):
device_id = None
if device in self._devices_by_id:
device_id = device
device_ids.append(device_id)
elif device in self._devices_by_name:
device_id = self._devices_by_name[device].device_id
device_ids.append(device_id)
else:
missing_device_ids.add(device)
if device_id and device in missing_device_ids:
missing_device_ids.remove(device)
for dev in devices:
parse_device_id(dev)
# Fail if some devices haven't been found after refreshing
assert not missing_device_ids, 'Could not find the following devices: {}'.format(list(missing_device_ids))
async with aiohttp.ClientSession(timeout=self._timeout) as session:
api = pysmartthings.SmartThings(session, self._access_token)
status_tasks = [
asyncio.ensure_future(self._get_device_status(api, device_id))
for device_id in device_ids
]
# noinspection PyTypeChecker
return await asyncio.gather(*status_tasks)
@action
def status(self, device: Optional[Union[str, List[str]]] = None) -> List[dict]:
"""
Refresh and return the status of one or more devices.
:param device: Device or list of devices to refresh (default: all)
:return: A list containing on entry per device, and each entry containing the current device state. Example:
.. code-block:: json
[
{
"device_id": "switch-1",
"name": "Fan",
"switch": false
},
{
"device_id": "tv-1",
"name": "Samsung Smart TV",
"switch": true
}
]
"""
self.refresh_info()
if not device:
self.refresh_info()
devices = self._devices_by_id.keys()
elif isinstance(device, str):
devices = [device]
else:
devices = device
with self._refresh_lock:
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
return loop.run_until_complete(self._refresh_status(devices))
finally:
loop.stop()
@action
def on(self, device: str, *args, **kwargs) -> dict:
"""
Turn on a device with ``switch`` capability.
:param device: Device name or ID.
:return: Device status
"""
self.execute(device, 'switch', 'on')
# noinspection PyUnresolvedReferences
return self.status(device).output[0]
@action
def off(self, device: str, *args, **kwargs) -> dict:
"""
Turn off a device with ``switch`` capability.
:param device: Device name or ID.
:return: Device status
"""
self.execute(device, 'switch', 'off')
# noinspection PyUnresolvedReferences
return self.status(device).output[0]
@action
def toggle(self, device: str, *args, **kwargs) -> dict:
"""
Toggle a device with ``switch`` capability.
:param device: Device name or ID.
:return: Device status
"""
import pysmartthings
device = self._get_device(device)
device_id = device.device_id
async def _toggle() -> bool:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
api = pysmartthings.SmartThings(session, self._access_token)
dev = await api.device(device_id)
assert 'switch' in dev.capabilities, 'The device {} has no switch capability'.format(dev.label)
await dev.status.refresh()
state = 'off' if dev.status.switch else 'on'
ret = await dev.command(component_id='main', capability='switch', command=state, args=args)
assert ret, 'The command switch={state} failed on device {device}'.format(state=state, device=dev.label)
return not dev.status.switch
with self._refresh_lock:
loop = asyncio.new_event_loop()
state = loop.run_until_complete(_toggle())
return {
'id': device_id,
'name': device.label,
'on': state,
}
@property
def switches(self) -> List[dict]:
"""
:return: List of switch devices statuses in :class:`platypush.plugins.switch.SwitchPlugin` compatible format.
Example:
.. code-block:: json
[
{
"id": "switch-1",
"name": "Fan",
"on": false
},
{
"id": "tv-1",
"name": "Samsung Smart TV",
"on": true
}
]
"""
# noinspection PyUnresolvedReferences
devices = self.status().output
return [
{
'name': device['name'],
'id': device['device_id'],
'on': device['switch'],
}
for device in devices
if 'switch' in device
]
# vim:sw=4:ts=4:et: