In the Evening, Turn Light Temperature to Warm

Hi, I’ve built myself a feature which I’d like to contribute upstream to the HA codebase so others can benefit from it.

Scenario: Much like my computer is set to automatically adjust the color temperature of my screen, I like to have the same happen with the lamps at my home, so that the reduced exposure to blue light will make me fall asleep faster.

Solution: I’ve tried using HA Automations but it quickly got out of hand and unreliable. So I’ve devised a python script to do this.

#!/usr/bin/env python3

import paho.mqtt.client as mqtt
import requests
import json
from datetime import datetime

WARM_COLOR_TEMP = 500
COLD_COLOR_TEMP = 143

import time
from datetime import timedelta
cache = {}
def cache_with_expiry(expiry_duration):
    """Cache decorator with expiry duration."""
    def decorator(func):
        def wrapper(*args, **kwargs):
            current_time = time.time()
            expiry_time = expiry_duration.total_seconds()
            if func not in cache or current_time - cache[func]['time'] > expiry_time:
                result = func(*args, **kwargs)
                cache[func] = {'result': result, 'time': current_time}
            return cache[func]['result']
        return wrapper
    return decorator

@cache_with_expiry(timedelta(hours=6))
def get_sun_times():
    """Get the sunrise and sunset times for today."""
    offset = datetime.now().astimezone().utcoffset().total_seconds() / 3600
    offset = f"{offset:+03.0f}:00"
    date = datetime.utcnow().isoformat()[:10]
    url = f"https://api.met.no/weatherapi/sunrise/3.0/sun?lat=${latitude}&lon=${longitude}&date={date}&offset={offset}"
    print('url', url)
    try:
      response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/115.0'})
      response.raise_for_status()
    except requests.exceptions.HTTPError as err:
      print(f'HTTP error occurred: {err}') 
      print(f'Status Code: {response.status_code}') 
      print(f'Response Text: {response.text}') 
      raise err
    print('response', response)
    data = response.json()
    print('data', data)
    sunrise = data['properties']['sunrise']['time']
    sunrise = datetime.strptime(sunrise, "%Y-%m-%dT%H:%M%z")
    sunset = data['properties']['sunset']['time']
    sunset = datetime.strptime(sunset, "%Y-%m-%dT%H:%M%z")
    return sunrise, sunset

def is_sun_above_horizon():
    """Check if the sun is above the horizon."""
    sunrise, sunset = get_sun_times()
    now = datetime.now(sunrise.tzinfo)
    return sunrise < now < sunset

# MQTT callback function
def on_message(client, userdata, msg):
    try:
      payload = json.loads(msg.payload.decode('utf-8'))
    except:
      return
    if not isinstance(payload, dict):
      return
      print(f"Received message on {msg.topic}: {payload}")
    if (payload.get('state') == 'ON' and '/set' not in msg.topic and 'brightness' in payload):
      print('<--', msg.topic, payload)
      if (is_sun_above_horizon()):
        if (payload.get('color_temp') != COLD_COLOR_TEMP or 'color_temp' not in payload):
          print("Setting color_temp to cold")
          new_payload = json.dumps({"state": "ON", "brightness": 254, "color_temp": COLD_COLOR_TEMP})
          print('-->', f"{msg.topic}/set", new_payload)
          client.publish(f"{msg.topic}/set", new_payload, qos=1, retain=False)
      else:
        if (payload.get('color_temp') != WARM_COLOR_TEMP or 'color_temp' not in payload):
          print("Setting color_temp to warm")
          new_payload = json.dumps({"state": "ON", "brightness": 254, "color_temp": WARM_COLOR_TEMP})
          print('-->', f"{msg.topic}/set", new_payload)
          client.publish(f"{msg.topic}/set", new_payload, qos=1, retain=False)

print("Populating cache...")
print(get_sun_times())
print("Cache populated.")

print("Setting up MQTT...")
client = mqtt.Client()
client.username_pw_set("${mqttUser}", "${mqttPassword}")
client.on_message = on_message
client.connect("${mqttHost}", ${mqttPort}, 60)
client.subscribe("zigbee2mqtt/#")
client.loop_forever()

I have this set up in a systemd service.

So my question is, how can I contribute this script to the HA codebase and community? What would that look like, as an Integration perhaps?

Note that this script works for all and any lamps that you have added to your zigbee network. This is actually my desired behavior. I suppose if we add it to HA we could have a ‘denylist’ selection or ‘allowlist’ to select only a few devices.

I which I had more context on why this got hard to maintain as a simple HA automation but honestly it’s been a while and I forgot the edge cases.

One thing to note is that my lamps are turned on and off by fliping the switch, which cuts power to the lamp. Luckly, when lamps are turned on they broadcast their current state, so that means it works. When the light first is turned on, it’ll have the previous state, but a couple of seconds later the color temperature is adjusted.

If you tell me this is not something the HA community is intereted in, that’s useful feedback too. Just thought I’d share the script and ask how to merge it upstream so that I could give back a little bit of all that I’ve received from this project for free.

Thanks,
ooo

Not to sound negative, but this is usually solved by an automation. Where did you get stuck?

This is bad practice in a Zigbee network. Router devices needs their power on at all times to keep a healthy network. If you found the automation unreliable this might very well be the reason for it.

1 Like

There are a couple of existing integrations that do this you could look at to see how they implemented it.