heatpump_weather.py

from appdaemon.plugins.hass import Hass
import datetime as dt
import requests

HA_URL = "http://xxx.xxx.xxx.xxx:8123"
ACCESS_TOKEN = "<Your access token>"
HEADERS = {
  "Authorization": f"Bearer {ACCESS_TOKEN}",
  "Content-Type": "application/json",
}
START_TIME = "08:00:00"
WEATHER_ENTITY = "weather.xxx"
SCHEDULE_ID = "xxxxxx"

class HeatPumpWeather(Hass):

    def initialize(self):
        run_time = self.parse_time(START_TIME, aware=True)
        self.run_daily(self.timer_cb, run_time)
        self.listen_event(self.event_cb, "UPDATE_HEATPUMP_SCHEDULE")

    def timer_cb(self, kwargs):
        self.log("Timer-Trigger")
        self.run_main_logic()

    def event_cb(self, event_name, data, kwargs):
        self.log(f"Event-Trigger: {event_name}")
        self.run_main_logic()

    def run_main_logic(self):
        weather_data = self.call_service(
            "weather/get_forecasts",
            entity_id=WEATHER_ENTITY,
            type="hourly"
        )
        self.process_data(weather_data)

    def process_data(self, weather_data):
        forecast = weather_data["result"]["response"][WEATHER_ID]["forecast"]
        tomorrow = (dt.datetime.now() + dt.timedelta(days=1)).astimezone().date()

        def utc_to_local(ts):
            return dt.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ") \
                .replace(tzinfo=dt.timezone.utc) \
                .astimezone()

        processed_weather = []
        for f in forecast:
            local_dt = utc_to_local(f["datetime"])
            if local_dt.date() == tomorrow:
                temp = f["temperature"]
                # if local_dt.hour in range(0, 7):
                #     temp += 2

                processed_weather.append({
                    "datetime": local_dt,
                    "temperature": temp
                })

        if not processed_weather:
            self.log("No weather data for today.")
            return

        max_temp = max(h["temperature"] for h in processed_weather)
        max_entries = [h for h in processed_weather if h["temperature"] == max_temp]

        local_times = [e["datetime"] for e in max_entries]

        if len(local_times) > 1:
            avg_timestamp = sum(t.timestamp() for t in local_times) / len(local_times)
            avg_time = dt.datetime.fromtimestamp(avg_timestamp).astimezone()
        else:
            avg_time = local_times[0]

        result = {"temperature": round(max_temp, 1), "time": avg_time.strftime("%H:%M")}
        self.log(result)
        self.edit_schedule(avg_time)

    def edit_schedule(self, time):
        start = (time + dt.timedelta(hours=-1)).strftime('%H:%M')
        stop = (time + dt.timedelta(hours=1)).strftime('%H:%M')

        data = {
            "schedule_id": SCHEDULE_ID,
            "timeslots": [
                {
                    "start": start,
                    "stop": stop,
                    "actions": [
                        {
                            "service": "select.select_option",
                            "entity_id": "select.daikin_ekrhh_dhw_reheat_on_off",
                            "service_data": {"option": "On"}
                        }
                    ]
                },
                {
                    "start": stop,
                    "stop": "00:00",
                    "actions": [
                        {
                            "service": "select.select_option",
                            "entity_id": "select.daikin_ekrhh_dhw_reheat_on_off",
                            "service_data": {"option": "Off"}
                        }
                    ]
                }
            ]
        }

        headers = HEADERS
        response = requests.post(f"{HA_URL}/api/scheduler/edit", json=data, headers=headers)

        if response.ok:
            self.log("Scheduler erfolgreich aktualisiert!")
            self.log(response.json())
        else:
            self.log(f"Fehler: {response.status_code}")
            self.log(response.text)