publish_smartthings_state.py

#!/usr/local/bin/python3.10
from ha_mqtt_discoverable import Settings, DeviceInfo
from ha_mqtt_discoverable.sensors import BinarySensor, BinarySensorInfo, Sensor, SensorInfo
import aiohttp
import pysmartthings
import asyncio
from time import sleep
import pprint 
 
target_attributes = [
#  [Attribute Name, Attribute Type, Friendly Name (Optional)]
   ["name","string"],
   ["switch","binary","Power"],
   ["operatingState","string"],
   ["machineState","string", "Machine State"],
   ["remainingTimeStr","string", "Time Remaining String"],
   ["remainingTime","string", "Time Remaining"],
   ["washerJobState","string", "Current Task"],
   ["dryerJobState","string", "Current Task"],
   ["availableTypes", "string"],
   ["completionTime", "string"],
   ["density", "string"],
   ["dryerDryLevel", "string"],
   ["dryerWrinklePrevent", "string"],
   ["dryingTemperature", "string"],
   ["error", "string"],
   ["inputSource", "string"],
   ["lockState", "string"],
   ["mute", "string"],
   ["n", "string", "Device Name"],
   ["pictureMode", "string"],
   ["playbackStatus", "string"],
   ["presets", "string"],
   ["progress", "string"],
   ["remoteControlEnabled", "string"],
   ["soundMode", "string"],
   ["supportedPictureModes", "string"],
   ["supportedPictureModesMap", "string"],
   ["supportedPlaybackCommands", "string"],
   ["supportedSoundModes", "string"],
   ["supportedSoundModesMap", "string"],
   ["supportedTrackControlCommands", "string"],
   ["supportsPowerOnByOcf", "string"],
   ["tvChannel", "string"],
   ["tvChannelName", "string"],
   ["type", "string"],
   ["volume", "string"],
   ["washerAutoDetergent", "string"],
   ["washerAutoSoftener", "string"],
   ["washerCycle", "string"],
   ["washerRinseCycles", "string"],
   ["washerSoilLevel", "string"],
   ["washerSpinLevel", "string"],
   ["washerWaterTemperature", "string"],
   ["washingTime", "string"],
   ["waterConsumption", "string"],
   ["waterLevel", "string"],
   ["waterValve", "string"],
]

POLL_INTERVAL = 60  # Poll rate in seconds
 
mqtt_settings = Settings.MQTT(host="192.168.0.164")


with open("/root/.smartthings_token") as f:
   token = f.readline().strip()

async def poll_devices():
   async with aiohttp.ClientSession() as session:
       api = pysmartthings.SmartThings(session, token)
       devices = await api.devices()
 
       for device in devices:
           device_info = DeviceInfo(name=device.label, identifiers=f"device_id_{device.label}")
           print(f"Processing device {device.label}")
           try:
               await device.status.refresh()
               pprint.PrettyPrinter(indent=4).pprint(device.status.values)
               for attribute in target_attributes:
                   value = device.status.values.get(attribute[0], None)
                   type = attribute[1]
                   attribute_name = attribute[0] if len(attribute) < 3 else attribute[2]
                   unique_id = f"{device.label}_{attribute_name}"
                   if value != None:
                       if type == "string":
                           print(f"Processing string {attribute_name} value {value}")
                           sensor_info = SensorInfo(name=f"{attribute_name}", unique_id=f"{unique_id}", device=device_info)
                           settings = Settings(mqtt=mqtt_settings, entity=sensor_info)
                           thesensor = Sensor(settings)
                           thesensor.set_state(value)
                       elif type == "binary":
                           print(f"Processing binary {attribute_name} value {value}")
                           sensor_info = BinarySensorInfo(name=f"{attribute_name}", unique_id=f"{unique_id}", device=device_info)
                           settings = Settings(mqtt=mqtt_settings, entity=sensor_info)
                           thesensor = BinarySensor(settings)
                           if value == "on":
                             thesensor.on()
                           else:
                             thesensor.off()
                       else:
                           raise Exception(f"Bad type {type}")
           except Exception as e:
               print(f"{device.label} not available {e}")
 
 
def main():
    loop = asyncio.get_event_loop()
    while True:
        loop.run_until_complete(poll_devices())
        sleep(POLL_INTERVAL)
    loop.close()
 
 
if __name__ == "__main__":
    main()