First off, a quick shout to @mingaldrichgan and @MHedish @mingaldrichgan had a first version of this that worked, but it would only work for a week or two before tokens expired. With the help of @MHedish we got updated python and driver code working so that it seems you don’t need to update the tokens….ever? I have been running it for a few months now and it has not needed any updating.
I have tested this only with my own Kwikset Halo Touch. It is the WiFi model. This could/probably does work with other Kwikset locks….but likely DOES need the lock to be accessible via the Kwikset app.
There are 3 main sections of steps….first you have to make the device in Hubitat, second you have to use python/command prompt to generate tokens and lastly you have to enter those tokens into habitat. The entire process is quite easy…but is slightly more work than a normal device integration. If unfamiliar with python…..I wasn’t either before this project. If you get stuck, ask Gemini AI or hit me up with a DM.
Start
Copy the below code.
/*
Kwikset Cloud Lock for Hubitat
Uses tokens initially generated by aiokwikset, then refreshes them directly
through AWS Cognito inside Hubitat.
Based on aiokwikset 0.7.3 protocol details.
*/
import groovy.json.JsonOutput
import groovy.json.JsonSlurper
import groovy.transform.Field
metadata {
definition(
name: "Kwikset Cloud Lock",
namespace: "community",
author: "Marc and Drew"
) {
capability "Actuator"
capability "Battery"
capability "Lock"
capability "Refresh"
capability "Initialize"
capability "Polling"
attribute "connectionStatus", "enum", ["online", "offline", "authenticationRequired"]
attribute "lastUpdated", "string"
attribute "door", "enum", ["open", "closed", "unknown"]
attribute "serialNumber", "string"
attribute "firmwareVersion", "string"
command "renewTokens"
command "discoverDevice"
command "setRefreshToken", [
[name: "Refresh Token*", type: "STRING", description: "Paste full Kwikset Refresh Token"]
]
command "setIdToken", [
[name: "ID Token*", type: "STRING", description: "Paste full Kwikset ID Token"]
]
command "setAccessToken", [
[name: "Access Token*", type: "STRING", description: "Paste full Kwikset Access Token"]
]
}
preferences {
input name: "kwiksetDeviceId", type: "text",
title: "Kwikset Device ID", required: true
input name: "kwiksetSerialNumber", type: "text",
title: "Serial number (optional; discovered automatically)", required: false
input name: "sourceFirstName", type: "text",
title: "First name for lock history (optional)", required: false
input name: "sourceLastName", type: "text",
title: "Last name for lock history (optional)", required: false
input name: "pollInterval", type: "enum",
title: "Polling interval",
options: ["1":"1 minute", "5":"5 minutes", "10":"10 minutes",
"15":"15 minutes", "30":"30 minutes", "60":"60 minutes"],
defaultValue: "5", required: true
input name: "logEnable", type: "bool",
title: "Enable debug logging", defaultValue: true
input name: "txtEnable", type: "bool",
title: "Enable description-text logging", defaultValue: true
}
}
@Field static final String COGNITO_ENDPOINT =
"https://cognito-idp.us-east-1.amazonaws.com/"
@Field static final String COGNITO_CLIENT_ID =
"5eu1cdkjp1itd1fi7b91m6g79s"
@Field static final String API_BASE =
"https://ynk95r1v52.execute-api.us-east-1.amazonaws.com"
@Field static final String API_USER_AGENT = "okhttp/5.0.0-alpha.14"
@Field static final Integer TOKEN_SAFETY_SECONDS = 300
def installed() {
logInfo("Installed")
updated()
}
def updated() {
unschedule()
if (settings.kwiksetSerialNumber) {
state.serialNumber = (settings.kwiksetSerialNumber as String).trim()
}
if (state.refreshToken && settings.kwiksetDeviceId) {
schedulePolling()
runIn(2, "initialize")
} else {
log.warn "Set your tokens using the Device Commands section, and save your Kwikset Device ID in Preferences."
}
if (logEnable) {
runIn(1800, "disableDebugLogging")
}
}
// Custom Token Commands
def setRefreshToken(String token) {
if (!token) {
log.error "No Refresh Token provided."
return
}
state.refreshToken = token.trim()
logInfo("Refresh Token updated successfully.")
checkAndInitialize()
}
def setIdToken(String token) {
if (!token) {
log.error "No ID Token provided."
return
}
state.idToken = token.trim()
logInfo("ID Token updated successfully.")
checkAndInitialize()
}
def setAccessToken(String token) {
if (!token) {
log.error "No Access Token provided."
return
}
state.accessToken = token.trim()
logInfo("Access Token updated successfully.")
checkAndInitialize()
}
private void checkAndInitialize() {
if (state.refreshToken && settings.kwiksetDeviceId) {
runIn(1, "initialize")
}
}
def initialize() {
try {
ensureAuthenticated()
discoverDevice()
refresh()
} catch (Exception e) {
handleFailure("Initialization", e)
}
}
def poll() {
refresh()
}
def refresh() {
logInfo("Executing refresh...")
try {
ensureAuthenticated()
Map response = apiRequest(
"GET",
"/prod_v1/devices_v2/${urlEncode(settings.kwiksetDeviceId as String)}"
)
logDebug("Raw API Response: ${response}")
Map deviceData = firstDataObject(response)
if (!deviceData) {
throw new RuntimeException("Kwikset returned no device data")
}
updateFromDevice(deviceData)
sendEvent(name: "connectionStatus", value: "online")
sendEvent(name: "lastUpdated", value: formatTimestamp(new Date()))
} catch (Exception e) {
handleFailure("Refresh", e)
}
}
def discoverDevice() {
try {
ensureAuthenticated()
Map userResponse = apiRequest("GET", "/prod_v1/users/me")
Map userData = firstDataObject(userResponse)
if (userData) {
state.firstName = settings.sourceFirstName ?: userData.firstname ?: ""
state.lastName = settings.sourceLastName ?: userData.lastname ?: ""
}
Map deviceResponse = apiRequest(
"GET",
"/prod_v1/devices_v2/${urlEncode(settings.kwiksetDeviceId as String)}"
)
Map deviceData = firstDataObject(deviceResponse)
if (!deviceData) {
throw new RuntimeException(
"Device ID ${settings.kwiksetDeviceId} was not found"
)
}
state.serialNumber =
settings.kwiksetSerialNumber ?: deviceData.serialnumber
if (!state.serialNumber) {
throw new RuntimeException(
"Device response did not contain serialnumber"
)
}
sendEvent(name: "serialNumber", value: state.serialNumber)
if (deviceData.devicename) {
device.updateLabel(deviceData.devicename as String)
}
updateFromDevice(deviceData)
logInfo("Discovered ${deviceData.devicename ?: settings.kwiksetDeviceId}")
} catch (Exception e) {
handleFailure("Device discovery", e)
}
}
def lock() {
executeLockAction("lock")
}
def unlock() {
executeLockAction("unlock")
}
private void executeLockAction(String action) {
try {
ensureAuthenticated()
if (!state.serialNumber) {
discoverDevice()
}
if (!state.serialNumber) {
throw new RuntimeException("No serial number is available")
}
String personName = (
(settings.sourceFirstName ?: state.firstName ?: "") +
(settings.sourceLastName ?: state.lastName ?: "")
).take(20)
if (!personName) {
personName = "Hubitat"
}
Map payload = [
action: action,
source: JsonOutput.toJson([
name: personName,
device: "apikwikset"
])
]
apiRequest(
"PATCH",
"/prod_v1/devices/${urlEncode(state.serialNumber as String)}/status",
payload
)
sendEvent(
name: "lock",
value: action == "lock" ? "locking" : "unlocking",
descriptionText: "${device.displayName} is ${action}ing"
)
sendEvent(name: "connectionStatus", value: "online")
// Dual follow-up to capture delayed cloud updates
runIn(4, "refresh")
runIn(10, "refresh")
} catch (Exception e) {
handleFailure("${action.capitalize()} command", e)
}
}
def renewTokens() {
try {
refreshTokens()
logInfo("Kwikset tokens renewed")
} catch (Exception e) {
handleFailure("Token renewal", e)
}
}
private void ensureAuthenticated() {
if (!state.refreshToken) {
sendEvent(name: "connectionStatus", value: "authenticationRequired")
throw new RuntimeException("A refresh token is required")
}
Long expiresAt = (state.tokenExpiresAt ?: 0L) as Long
Boolean missingIdToken = !state.idToken
Boolean expiringSoon =
expiresAt == 0L ||
now() >= expiresAt - (TOKEN_SAFETY_SECONDS * 1000L)
if (missingIdToken || expiringSoon) {
refreshTokens()
}
}
private void refreshTokens() {
String refreshToken = state.refreshToken as String
if (!refreshToken) {
throw new RuntimeException("No refresh token is available")
}
Map body = [
AuthFlow: "REFRESH_TOKEN_AUTH",
ClientId: COGNITO_CLIENT_ID,
AuthParameters: [
REFRESH_TOKEN: refreshToken
]
]
Map params = [
uri: COGNITO_ENDPOINT,
headers: [
"Content-Type": "application/x-amz-json-1.1",
"X-Amz-Target":
"AWSCognitoIdentityProviderService.InitiateAuth",
"User-Agent":
"aws-sdk-kotlin/1.3.81 ua/2.1 api/cognito-identity-provider#1.3.81"
],
contentType: "application/json",
requestContentType: "application/json",
body: JsonOutput.toJson(body),
timeout: 30
]
Map authenticationResult
httpPost(params) { response ->
authenticationResult =
response?.data?.AuthenticationResult as Map
}
if (!authenticationResult?.IdToken ||
!authenticationResult?.AccessToken) {
throw new RuntimeException(
"Cognito refresh returned no ID/access token"
)
}
state.idToken = authenticationResult.IdToken
state.accessToken = authenticationResult.AccessToken
if (authenticationResult.RefreshToken) {
state.refreshToken = authenticationResult.RefreshToken
}
Integer expiresIn =
(authenticationResult.ExpiresIn ?: 3600) as Integer
state.tokenExpiresAt = now() + (expiresIn * 1000L)
sendEvent(name: "connectionStatus", value: "online")
logDebug("Token refresh succeeded; expires in ${expiresIn} seconds")
}
private Map apiRequest(
String method,
String path,
Map body = null,
Boolean retryAuthentication = true
) {
ensureIdTokenPresent()
Map params = [
uri: "${API_BASE}${path}",
headers: [
"Authorization": "Bearer ${state.idToken}",
"User-Agent": API_USER_AGENT,
"Accept-Encoding": "gzip",
"Accept": "application/json"
],
contentType: "application/json",
timeout: 30
]
if (body != null) {
params.requestContentType = "application/json"
params.body = JsonOutput.toJson(body)
}
try {
Map result
switch (method) {
case "GET":
httpGet(params) { response ->
result = normalizeResponseData(response?.data)
}
break
case "PATCH":
httpPatch(params) { response ->
result = normalizeResponseData(response?.data)
}
break
case "POST":
httpPost(params) { response ->
result = normalizeResponseData(response?.data)
}
break
default:
throw new IllegalArgumentException(
"Unsupported HTTP method ${method}"
)
}
return result ?: [:]
} catch (groovyx.net.http.HttpResponseException e) {
Integer status = e?.statusCode as Integer
if (retryAuthentication && (status == 401 || status == 403)) {
logDebug("API rejected token; renewing and retrying once")
refreshTokens()
return apiRequest(method, path, body, false)
}
throw e
}
}
private void ensureIdTokenPresent() {
if (!state.idToken) {
refreshTokens()
}
}
private Map normalizeResponseData(Object data) {
if (data == null) {
return [:]
}
if (data instanceof Map) {
return data as Map
}
if (data instanceof String) {
Object parsed = new JsonSlurper().parseText(data as String)
return parsed instanceof Map ? parsed as Map : [data: parsed]
}
return [data: data]
}
private Map firstDataObject(Map response) {
Object data = response?.data
if (data instanceof List && !data.isEmpty() && data[0] instanceof Map) {
return data[0] as Map
}
if (data instanceof Map) {
return data as Map
}
return null
}
private void updateFromDevice(Map data) {
logDebug("Processing device payload: ${data}")
String lockValue = mapLockState(data)
logInfo("Lock status mapped to: ${lockValue}")
if (lockValue) {
sendEvent(
name: "lock",
value: lockValue,
descriptionText: "${device.displayName} is ${lockValue}"
)
}
Integer battery = parseInteger(
firstNonNull(
data.batterypercentage,
data.batteryPercentage,
data.battery
)
)
if (battery != null) {
battery = Math.max(0, Math.min(100, battery))
sendEvent(name: "battery", value: battery, unit: "%")
}
String doorValue = mapDoorState(data)
if (doorValue) {
sendEvent(name: "door", value: doorValue)
}
Object serial = firstNonNull(data.serialnumber, data.serialNumber)
if (serial) {
state.serialNumber = serial as String
sendEvent(name: "serialNumber", value: serial as String)
}
Object firmware =
firstNonNull(data.firmwareversion, data.firmwareVersion)
if (firmware) {
sendEvent(name: "firmwareVersion", value: firmware as String)
}
}
private String mapLockState(Map data) {
Object raw = firstNonNull(
data.doorstatus,
data.doorStatus,
data.devicestatus,
data.deviceStatus,
data.lockstatus,
data.lockStatus,
data.alexahomelockstatus,
data.googlehomelockstatus
)
if (raw == null) {
return null
}
String value = raw.toString().trim().toLowerCase()
if (value in ["locked", "lock", "1", "true"]) {
return "locked"
}
if (value in ["unlocked", "unlock", "0", "false"]) {
return "unlocked"
}
if (value.contains("jam")) {
return "unknown"
}
if (value.contains("unlock")) {
return "unlocked"
}
if (value.contains("lock")) {
return "locked"
}
return "unknown"
}
private String mapDoorState(Map data) {
// Halo model payloads map lock state to doorstatus;
// explicit open/close door contact sensors use doorposition
Object raw = firstNonNull(
data.doorposition,
data.doorPosition
)
if (raw == null) {
return null
}
String value = raw.toString().trim().toLowerCase()
if (value in ["open", "opened", "1", "true"]) {
return "open"
}
if (value in ["closed", "close", "0", "false"]) {
return "closed"
}
return "unknown"
}
private Object firstNonNull(Object... values) {
for (Object value : values) {
if (value != null) {
return value
}
}
return null
}
private Integer parseInteger(Object value) {
if (value == null) {
return null
}
try {
if (value instanceof Number) {
return Math.round((value as Number).doubleValue()) as Integer
}
return Math.round(Double.parseDouble(value.toString())) as Integer
} catch (Exception ignored) {
return null
}
}
private void schedulePolling() {
Integer minutes = (settings.pollInterval ?: "5") as Integer
switch (minutes) {
case 1:
runEvery1Minute("poll")
break
case 5:
runEvery5Minutes("poll")
break
case 10:
runEvery10Minutes("poll")
break
case 15:
runEvery15Minutes("poll")
break
case 30:
runEvery30Minutes("poll")
break
case 60:
runEvery1Hour("poll")
break
default:
runEvery5Minutes("poll")
}
}
private String urlEncode(String value) {
return java.net.URLEncoder.encode(value, "UTF-8")
.replace("+", "%20")
}
private String formatTimestamp(Date date) {
return date.format(
"yyyy-MM-dd'T'HH:mm:ssXXX",
location?.timeZone ?: TimeZone.getTimeZone("UTC")
)
}
private void handleFailure(String operation, Exception e) {
String message = e?.message ?: e?.toString() ?: "Unknown error"
log.error "${operation} failed: ${message}"
if (message.toLowerCase().contains("notauthorized") ||
message.toLowerCase().contains("refresh token") ||
message.toLowerCase().contains("authentication")) {
sendEvent(
name: "connectionStatus",
value: "authenticationRequired"
)
} else {
sendEvent(name: "connectionStatus", value: "offline")
}
}
def disableDebugLogging() {
log.warn "Debug logging disabled automatically"
device.updateSetting("logEnable", [value: "false", type: "bool"])
}
private void logDebug(String message) {
if (settings.logEnable) {
log.debug message
}
}
private void logInfo(String message) {
if (settings.txtEnable) {
log.info message
}
}
- In Hubitat, go to Drives Code on the left side, then click the “add driver” button.Once there, paste the code you copied above into the blank space and click save in top right.
- Go to Devices > Add Device > Virtual > Kwikset Cloud Lock > Next
- Finish the device add steps as you normally would.Then go to the device details>commands tab.
Now we move to capture the Tokens…….Before you continue, you need to have Python installed. If you don’t, just go to Download Python | Python.org and downloand and install the latest version of Python.
Once Python is installed follow these steps.
-
Create a folder directly in your C drive called “KwiksetLock” so that it’s address is “C:\KwiksetLock”
-
Copy the below code into a notepad and save it as “python kwikset_token_seed.py” and save that file in the previous created folder in C:\KwiksetLock
import asyncio
from getpass import getpassfrom aiokwikset import APIasync def main() -> None:
email = input("Enter your Kwikset Email: ").strip()
password = getpass("Enter your Kwikset Password: ")async with API() as api:
print("Authenticating with Kwikset...")
await api.async_login(email, password)print("\n--- TOKENS FOR HUBITAT ---")
print(f"ID Token: {api.id_token}")
print(f"Access Token: {api.access_token}")
print(f"Refresh Token: {api.refresh_token}")user = await api.user.get_info()
if user:
print(
f"User: {user.get('firstname', '')} "
f"{user.get('lastname', '')}".strip()
)homes = await api.user.get_homes()
for home in homes:
print(
f"\nHome Name: {home.get('homename')} | "
f"Home ID: {home.get('homeid')}"
)
devices = await api.device.get_devices(home["homeid"])
for device in devices:
print(
" -> "
f"Device Name: {device.get('devicename')} | "
f"Device ID: {device.get('deviceid')} | "
f"Serial Number: {device.get('serialnumber')}"
)print("--------------------------")if __name__ == "__main__":
asyncio.run(main()) -
Open Command prompt and type “cd C:\KwiksetLock”
-
Type “pip install aiokwikset==0.7.3” and let it run.
-
Type “python kwikset_token_seed.py”
-
You will be prompted to enter your Kwikset email and password.This is the same info you use to log into the Kiwkset app.Note…when typing your password…it is a “blind” fill.It doesn’t show what you are typing.
If you did this correctly, you will get a message that starts with:
-----TOKENS FOR HUBITAT----
From there you will see: ID Token, Access Token and Refresh Token…and then at the bottom is some home info.
- Copy the ID Token (no extra data….and only copy the ID Token) and past that in the Hubitat device “Set ID Token” spot and click “set”.
- Repeat with the Access Token and the Refresh Token.
- In the habitat device now go to the Preferences Tab.
- Copy the “Device ID” from the end of the command prompt and paste that in the “Kwikset Device ID” spot in habitat.
- click the save button.
- Go back to the command tab and the information on the right side should be filled out and you can now test the commands.