Skip to content

nice interface -- I tested it with Homeworks QS #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open

nice interface -- I tested it with Homeworks QS #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
52731f9
Update __init__.py
dulitz May 9, 2021
df80c9f
Update __init__.py
dulitz May 9, 2021
fd58d0a
C # not returned by Caseta or Radio Ra 2 Selectlarify the parser f…
dulitz May 27, 2021
48ec8db
use f-strings consistently since we are already defined as python 3.6+
dulitz May 27, 2021
ad8f997
add error detection during login. also add a constant used by Homewor…
dulitz May 27, 2021
46ad6bd
updated to reflect testing with Homeworks QS
dulitz May 27, 2021
65244e5
clarify that 'telnet' must be turned on in the app only for Caseta PR…
dulitz May 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,23 @@
# liplib

Interface module for Lutron Integration Protocol (LIP) over Telnet.
Interface module for Lutron Integration Protocol (LIP) over TCP port 23, or "telnet."

This module connects to a Lutron hub through the Telnet interface which must be enabled through the integration menu in the Lutron mobile app.
This module connects to a Lutron hub through the telnet interface which for Caseta PRO and Radio Ra2 Select must be enabled through the integration menu in the Lutron mobile app.

Supported bridges / main repeaters:
Supported bridges / main repeaters / controllers:
- [Lutron Caseta](http://www.casetawireless.com) Smart Bridge **PRO** (L-BDGPRO2-WH)
- [Ra2 Select](http://www.lutron.com/en-US/Products/Pages/WholeHomeSystems/RA2Select/Overview.aspx) Main Repeater (RR-SEL-REP-BL or RR-SEL-REP2S-BL)
- [Radio Ra2 Select](http://www.lutron.com/en-US/Products/Pages/WholeHomeSystems/RA2Select/Overview.aspx) Main Repeater (RR-SEL-REP-BL or RR-SEL-REP2S-BL)
- Radio Ra2
- Homeworks QS

Other bridges / main repeaters that use the Lutron Integration Protocol (e.g. Radio Ra2, HomeWorks QS) may also work with this library, but are untested.
Other bridges / main repeaters that use the Lutron Integration Protocol (e.g. Quantum, Athena, myRoom) should also work with this library, but are untested.

This module is designed to use selected commands from the [Lutron Integration Protocol](http://www.lutron.com/TechnicalDocumentLibrary/040249.pdf). The command set most closely resembles RadioRa 2, but not all features listed for RadioRa 2 are supported.
This module is designed to use selected commands from the [Lutron Integration Protocol](http://www.lutron.com/TechnicalDocumentLibrary/040249.pdf). Not all features documented in the protocol are supported by this module. If you implement an extension, please submit a pull request.

In addition to sending and receiving commands, a function is provided to process a JSON Integration Report obtained by a user from the Lutron mobile app.

An interface to the obsolete Lutron Homeworks Illumination system is [available here](https://github.com/dulitz/porter/blob/main/illiplib.py). It is drop-in compatible with this interface.

Authors:
upsert (https://github.com/upsert)

Expand Down
171 changes: 81 additions & 90 deletions liplib/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
"""
Interface module for Lutron Integration Protocol (LIP) over Telnet.
Interface module for Lutron Integration Protocol (LIP).

This module connects to a Lutron hub through the Telnet interface which must be
enabled through the integration menu in the Lutron mobile app.
This module connects to a Lutron hub through the tcp/23 ("telnet") interface which
must be enabled through the integration menu in the Lutron mobile app.

Authors:
upsert (https://github.com/upsert)
Expand All @@ -26,99 +26,85 @@

_LOGGER = logging.getLogger(__name__)


async def async_load_integration_report(fname: str) -> list:
def load_integration_report(integration_report) -> list:
"""Process a JSON integration report and return a list of devices.

Each returned device will have an 'id', 'name', 'type' and optionally
a list of button IDs under 'buttons' for remotes
and an 'area_name' attribute if the device is assigned
to an area.
and an 'area_name' attribute if the device is assigned to an area.

To generate an integration report in the Lutron (Radio Ra2 Select) app,
click the gear, then Advanced, then "Send Integration Report."
"""
devices = []
with open(fname, encoding='utf-8') as conf_file:
integration_report = json.load(conf_file)
# _LOGGER.debug(integration)
if "LIPIdList" in integration_report:
# lights and switches are in Zones
if "Zones" in integration_report["LIPIdList"]:
_process_zones(devices, integration_report)
# remotes are in Devices, except ID 1 which is the bridge itself
if "Devices" in integration_report["LIPIdList"]:
for device in integration_report["LIPIdList"]["Devices"]:
# extract scenes from integration ID 1 - the smart bridge
if device["ID"] == 1 and "Buttons" in device:
_process_scenes(devices, device)
elif device["ID"] != 1 and "Buttons" in device:
device_obj = {CONF_ID: device["ID"],
CONF_NAME: device["Name"],
CONF_TYPE: "sensor",
CONF_BUTTONS:
[b["Number"]
for b in device["Buttons"]]}
if "Area" in device and "Name" in device["Area"]:
device_obj[CONF_AREA_NAME] = device["Area"]["Name"]
devices.append(device_obj)
else:
_LOGGER.warning("'LIPIdList' not found in the Integration Report."
" No devices will be loaded.")
return devices
lipidlist = integration_report.get("LIPIdList")
assert lipidlist, integration_report


def _process_zones(devices, integration_report):
"""Process zones and append devices."""
for zone in integration_report["LIPIdList"]["Zones"]:
# _LOGGER.debug(zone)
# lights and switches are in Zones
for zone in lipidlist.get("Zones", []):
device_obj = {CONF_ID: zone["ID"],
CONF_NAME: zone["Name"],
CONF_TYPE: "light"}
if "Area" in zone and "Name" in zone["Area"]:
device_obj[CONF_AREA_NAME] = zone["Area"]["Name"]
name = zone.get("Area", {}).get("Name", "")
if name:
device_obj[CONF_AREA_NAME] = name
devices.append(device_obj)

# remotes are in Devices, except ID 1 which is the bridge itself
for device in lipidlist.get("Devices", []):
# extract scenes from integration ID 1 - the smart bridge
if device["ID"] == 1:
for button in device.get("Buttons", []):
if not button["Name"].startswith("Button "):
_LOGGER.info("Found scene %d, %s", button["Number"], button["Name"])
devices.append({CONF_ID: device["ID"],
CONF_NAME: button["Name"],
CONF_SCENE_ID: button["Number"],
CONF_TYPE: "scene"})
else:
device_obj = {CONF_ID: device["ID"],
CONF_NAME: device["Name"],
CONF_TYPE: "sensor",
CONF_BUTTONS: [b["Number"] for b in device.get("Buttons", [])]}
name = device.get("Area", {}).get("Name", "")
device_obj[CONF_AREA_NAME] = name
devices.append(device_obj)

def _process_scenes(devices, device):
"""Process scenes and append devices."""
for button in device["Buttons"]:
if not button["Name"].startswith("Button "):
_LOGGER.info(
"Found scene %d, %s", button["Number"],
button["Name"])
devices.append({CONF_ID: device["ID"],
CONF_NAME: button["Name"],
CONF_SCENE_ID: button["Number"],
CONF_TYPE: "scene"})
return devices


# pylint: disable=too-many-instance-attributes
class LipServer:
"""Async class to communicate with a the bridge."""
"""Communicate with a Lutron bridge, repeater, or controller."""

READ_SIZE = 1024
DEFAULT_USER = b"lutron"
DEFAULT_PASSWORD = b"integration"
DEFAULT_PROMPT = b"GNET> "
LOGIN_PROMPT = b"login: "
RESPONSE_RE = re.compile(b"~([A-Z]+),([0-9.]+),([0-9.]+),([0-9.]+)\r\n")
OUTPUT = "OUTPUT"
DEVICE = "DEVICE"

class Action(IntEnum):
"""Action values."""
"""Action numbers for the OUTPUT command in the Lutron Integration Protocol."""

# Get or Set Zone Level
SET = 1
# Start Raising
RAISING = 2
# Start Lowering
LOWERING = 3
# Stop Raising/Lowering
STOP = 4
SET = 1 # Get or Set Zone Level
RAISING = 2 # Start Raising
LOWERING = 3 # Start Lowering
STOP = 4 # Stop Raising/Lowering

PRESET = 6 # SHADEGRP for Homeworks QS

class Button(IntEnum):
"""Button values."""
"""Action numbers for the DEVICE command in the Lutron Integration Protocol."""

PRESS = 3
RELEASE = 4
HOLD = 5 # not returned by Caseta or Radio Ra 2 Select
DOUBLETAP = 6 # not returned by Caseta or Radio Ra 2 Select

PRESS = 3
RELEASE = 4
LEDSTATE = 9 # "Button" is a misnomer; this queries LED state

class State(IntEnum):
"""Connection state values."""
Expand Down Expand Up @@ -146,7 +132,7 @@ def is_connected(self) -> bool:

async def open(self, host, port=23, username=DEFAULT_USER,
password=DEFAULT_PASSWORD):
"""Open a Telnet connection to the bridge."""
"""Open a telnet connection to the bridge."""
async with self._read_lock:
async with self._write_lock:
if self._state != LipServer.State.Closed:
Expand All @@ -158,31 +144,35 @@ async def open(self, host, port=23, username=DEFAULT_USER,
self._username = username
self._password = password

def cleanup(err):
_LOGGER.warning(f"error opening connection to Lutron {host}:{port}: {err}")
self._state = LipServer.State.Closed

# open connection
try:
connection = await asyncio.open_connection(host, port)
except OSError as err:
_LOGGER.warning("Error opening connection"
" to the bridge: %s", err)
self._state = LipServer.State.Closed
return
return cleanup(err)

self.reader = connection[0]
self.writer = connection[1]

# do login
await self._read_until(b"login: ")
if await self._read_until(self.LOGIN_PROMPT) is False:
return cleanup('no login prompt')
self.writer.write(username + b"\r\n")
await self.writer.drain()
await self._read_until(b"password: ")
if await self._read_until(b"password: ") is False:
return cleanup('no password prompt')
self.writer.write(password + b"\r\n")
await self.writer.drain()
await self._read_until(self.prompt)
if await self._read_until(self.prompt) is False:
return cleanup('login failed')

self._state = LipServer.State.Opened

async def _read_until(self, value):
"""Read until a given value is reached."""
"""Read until a given value is reached. Value may be regex or bytes."""
while True:
if hasattr(value, "search"):
# detected regular expression
Expand All @@ -191,18 +181,20 @@ async def _read_until(self, value):
self._read_buffer = self._read_buffer[match.end():]
return match
else:
assert isinstance(value, bytes), value
where = self._read_buffer.find(value)
if where != -1:
until = self._read_buffer[:where+len(value)]
self._read_buffer = self._read_buffer[where + len(value):]
return True
return until
try:
read_data = await self.reader.read(LipServer.READ_SIZE)
if not len(read_data):
_LOGGER.warning("Empty read from the bridge (clean disconnect)")
_LOGGER.warning("bridge disconnected")
return False
self._read_buffer += read_data
except OSError as err:
_LOGGER.warning("Error reading from the bridge: %s", err)
_LOGGER.warning(f"error reading from the bridge: {err}")
return False

async def read(self):
Expand All @@ -219,45 +211,44 @@ async def read(self):
int(match.group(2)), int(match.group(3)), \
float(match.group(4))
except ValueError:
print("Exception in ", match.group(0))
_LOGGER.warning(f"could not parse {match.group(0)}")
if match is False:
# attempt to reconnect
_LOGGER.info("Reconnecting to the bridge %s", self._host)
_LOGGER.info(f"Reconnecting to the bridge {self._host}")
self._state = LipServer.State.Closed
await self.open(self._host, self._port, self._username,
self._password)
return None, None, None, None

async def write(self, mode, integration, action, *args, value=None):
"""Write a list of values out to the Telnet interface."""
"""Write a list of values to the bridge."""
if hasattr(action, "value"):
action = action.value
async with self._write_lock:
if self._state != LipServer.State.Opened:
return
data = "#{},{},{}".format(mode, integration, action)
data = f"#{mode},{integration},{action}"
if value is not None:
data += ",{}".format(value)
data += f",{value}"
for arg in args:
if arg is not None:
data += ",{}".format(arg)
data += f",{arg}"
try:
self.writer.write((data + "\r\n").encode("ascii"))
await self.writer.drain()
except OSError as err:
_LOGGER.warning("Error writing out to the bridge: %s", err)
_LOGGER.warning(f"Error writing to the bridge: {err}")


async def query(self, mode, integration, action):
"""Query a device to get its current state."""
"""Query a device to get its current state. Does not handle LED queries."""
if hasattr(action, "value"):
action = action.value
_LOGGER.debug("Sending query %s, integration %s, action %s",
mode, integration, action)
_LOGGER.debug(f"Sending query {mode}, integration {integration}, action {action}")
async with self._write_lock:
if self._state != LipServer.State.Opened:
return
self.writer.write("?{},{},{}\r\n".format(mode, integration,
action).encode())
self.writer.write(f"?{mode},{integration},{action}\r\n".encode())
await self.writer.drain()

async def ping(self):
Expand All @@ -269,7 +260,7 @@ async def ping(self):
await self.writer.drain()

async def logout(self):
"""Logout and severe the connect to the bridge."""
"""Close the connection to the bridge."""
async with self._write_lock:
if self._state != LipServer.State.Opened:
return
Expand Down