Python SDK for GE Kitchen Appliances
Project description
gekitchen
Python SDK for GE WiFi-enabled kitchen appliances based on slixmpp.
The primary goal is to use this to power integrations for Home Assistant, though that
will probably need to wait on some new entity types.
Installation
pip install gekitchen
Usage
Simple Operation
from gekitchen import do_full_login_flow, GeClient
USERNAME = 'me@email.com'
PASSWORD = '$7r0nkP@s$w0rD'
xmpp_credentials = do_full_login_flow(USERNAME, PASSWORD)
client = GeClient(xmpp_credentials)
client.connect()
client.process(timeout=120)
Longer example
Here we're going to run the client in a pre-existing event loop. We're also going to register some event callbacks to update appliances every five minutes and to turn on our oven the first time we see it. Because that is safe!
import aiohttp
import asyncio
import logging
from datetime import timedelta
from typing import Any, Dict, Tuple
from gekitchen import (
ErdApplianceType,
ErdCode,
ErdCodeType,
ErdOvenCookMode,
GeAppliance,
GeClient,
OvenCookSetting,
async_do_full_login_flow,
)
_LOGGER = logging.getLogger(__name__)
USERNAME = 'me@email.com'
PASSWORD = '$7r0nkP@s$w0rD'
async def detect_appliance_type(data: Tuple[GeAppliance, Dict[ErdCodeType, Any]]):
"""
Detect the appliance type.
This should only be triggered once since the appliance type should never change.
Also, let's turn on ovens!
"""
appliance, state_changes = data
_LOGGER.debug(f'Appliance state change detected in {appliance:s}')
try:
_LOGGER.info(f'Setting appliance type to {state_changes[ErdCode.APPLIANCE_TYPE]:s}')
except KeyError:
# Not an APPLIANCE_TYPE update
return
if appliance.appliance_type == ErdApplianceType.OVEN:
_LOGGER.info('Turning on the oven!')
appliance.set_erd_value(
ErdCode.UPPER_OVEN_COOK_MODE,
OvenCookSetting(ErdOvenCookMode.BAKE_NOOPTION, 350)
)
_LOGGER.info('Set the timer!')
appliance.set_erd_value(ErdCode.UPPER_OVEN_KITCHEN_TIMER, timedelta(minutes=45))
async def do_periodic_update(appliance: GeAppliance):
"""Request a full state update every five minutes forever"""
_LOGGER.debug(f'Registering update callback for {appliance:s}')
while True:
await asyncio.sleep(5 * 60)
_LOGGER.debug(f'Requesting update for {appliance:s}')
if appliance.available:
await appliance.request_update()
async def start_client(evt_loop: asyncio.AbstractEventLoop, username: str, password: str):
"""Authenticate and launch the client."""
session = aiohttp.ClientSession()
_LOGGER.debug('Logging in')
xmpp_credentials = await async_do_full_login_flow(session, username, password)
client = GeClient(xmpp_credentials, event_loop=evt_loop)
client.add_event_handler('appliance_state_change', detect_appliance_type)
client.add_event_handler('add_appliance', do_periodic_update)
_LOGGER.debug('Connecting')
client.connect()
_LOGGER.debug('Processing')
asyncio.ensure_future(client.process_in_running_loop(), loop=evt_loop)
logging.basicConfig(level=logging.DEBUG, format='%(levelname)-8s %(message)s')
loop = asyncio.get_event_loop()
asyncio.ensure_future(start_client(loop, USERNAME, PASSWORD), loop=loop)
loop.run_until_complete(asyncio.sleep(300))
API Overview
The GE app communicates with devices via XMPP (Jabbber), sending pseudo-HTTP chat messages back and forth. Device
properties are represented by hex codes (represented by ErdCode objects in gekitchen), and values are sent as
hexadecimal strings without leading "0x", then json encoded as a dictionary. The device informs the client of a
state change by sending a PUBLISH message like this, informing us that the value of property 0x5205
(ErdCode.LOWER_OVEN_KITCHEN_TIMER in gekitchen) is now "002d" (45 minutes):
<body>
<publish>
<method>PUBLISH</method>
<uri>/UUID/erd/0x5205</uri>
<json>{"0x5205":"002d"}</json>
</publish>
</body>
Similarly, we can set the timer to 45 minutes by POSTing to the same "endpoint":
<body>
<request>
<method>POST</method>
<uri>/UUID/erd/0x5205</uri>
<json>{"0x5205":"002d"}</json>
</request>
</body>
In gekitchen, that would handled by the GeAppliance.set_erd_value method:
appliance.set_erd_value(ErdCode.LOWER_OVEN_KITCHEN_TIMER, timedelta(minutes=45))
We can also get a specific property, or, more commonly, request a full cache refresh by GETing the /UUID/cache
endpoint:
<body>
<request>
<id>0</id>
<method>GET</method>
<uri>/UUID/cache</uri>
</request>
</body>
The device will then respond to the GET with a response having a json payload:
<body>
<response>
<id>0</id>
<method>GET</method>
<uri>/UUID/cache</uri>
<json>{
"0x0006":"00",
"0x0007":"00",
"0x0008":"07",
"0x0009":"00",
"0x000a":"03",
"0x0089":"",
...
}</json>
</response>
</body>
Authentication
The authentication process has a few steps. First, we use Oauth2 to authenticate to the HTTPS API. After
authenticating, we can get a mobile device token, which in turn be used to get a new Bearer token, which, finally,
is used to get XMPP credentials to authenticate to the Jabber server. In gekitchen, going from username/password
to XMPP credentials is handled by do_full_login_flow(username, password).
Useful functions
do_full_login_flow(username, password)
Function to authenticate to the web API and get XMPP credentials. Returns a dict of XMPP credentials
Objects
GeClient(xmpp_credentials, event_loop=None, **kwargs)
Main XMPP client, and a subclass of slixmpp.ClientXMPP.
xmpp_credentials: dictA dictionary of XMPP credentials, usually obtained from eitherdo_full_login_flowor, in a more manual process,get_xmpp_credentialsevent_loop: asyncio.AbstractEventLoopOptional event loop. IfNone, the client will useasyncio.get_event_loop()**kwargsPassed toslixmpp.ClientXMPP
Useful Methods
connect()Connect to the XMPP serverprocess_in_running_loop(timeout: Optional[int] = None)Run in an existing event loop. Iftimeoutis given, stop running aftertimeoutsecondsadd_event_handler(name: str, func: Callable)Add an event handler. In addition to the events supported byslixmpp.ClientXMPP, we've added some more event types detailed below.
Properties
appliancesADict[str, GeAppliance]of all known appliances keyed on the appliances' JIDs.
Events
In addition to the standard slixmpp events, the GeClient object has support for the following:
'add_appliance'- Triggered after a new appliance is added. TheGeApplianceobject is passed to the callback'appliance_state_change'- Triggered when an appliance message with a new state, different from the existing, cached state is received. A tuple(appliance, state_changes)is passed to the callback, whereapplianceis theGeApplianceobject with the updated state andstate_changesis a dictionary{erd_key: new_value}of the changed state.
GeAppliance(jid, xmpp_client)
Representation of a single appliance
jid: Union[str, slixmpp.JID]The appliance's Jabber IDxmpp_client: GeClientThe client used to communicate with the device
Useful Methods
decode_erd_value(erd_code: ErdCodeType, erd_value: str)Decode a raw ERD property value.encode_erd_value(erd_code: ErdCodeType, erd_value: str)Decode a raw ERD property value.get_erd_value(erd_code: ErdCodeType)Get the cached value of ERD codeerd_code. Iferd_codeis a string, this function will attempt to convert it to anErdCodeobject first.request_update()Request the appliance send an update of all propertiessend_raw_message(mto, mbody, mtype='chat', msg_id=None)Send a message tomtowithmbodyin the body. No processing or sanity checking will be applied.send_request(method: str, uri: str, key=None, value=None, message_id=None)Send a pseudo-HTTP request to the applianceset_available()Mark the appliance as availableset_erd_value(erd_code: ErdType, value)Tell the device to set the property represented byerd_codetovalueset_unavailable()Mark the appliance as unavailableupdate_erd_value(erd_code: ErdType, value)Update the local property cache value forerd_codetovalue, where value is the not yet decoded hex string sent from the API. ReturnsTrueif that is a change in state,Falseotherwise.update_erd_values(self, erd_values: Dict[ErdCodeType, str])Update multiple values in the local property cache. Returns a dictionary of changed states or an emptydictif nothing actually changed.
Properties
appliance_type: Optional[ErdApplianceType]The type of appliance,Noneif unknownavailable: boolTrueif the appliance is available, otherwiseFalsejid: slixmpp.JIDThe appliance's Jabber ID
Useful Enum types
ErdCodeEnumof known ERD property codesErdApplianceTypeValues forErdCode.APPLIANCE_TYPEErdMeasurementUnitsValues forErdCode.TEMPERATURE_UNITErdOvenCookModePossible oven cook modes, used forOvenCookSettingamong other thingsErdOvenStateValues forErdCode.LOWER_OVEN_CURRENT_STATEandErdCode.UPPER_OVEN_CURRENT_STATE
Other types
OvenCookSettingAnamedtupleof anErdOvenCookModeand aninttemperatureOvenConfigurationAnamedtupleof boolean properties representing an oven's physical configuration
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for gekitchen-0.1.1-py3-none-any.whl
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 80d973209cbbfcba6ea0c6fccc6787d60fc2a664d469dfcd492c17fffb17ffc3 |
|
| MD5 | ded5e8ddbe286070ad069b126193278d |
|
| BLAKE2b-256 | f2928b66ba641cbf1429cf3601cdd0d1c58ccb32095c149b9ed137be0bf30ed4 |
