# pyscript/daikin_ml.py
# Online-RLS (oppiva) ohjain Daikin demand-selectille.
# Ominaisuudet:
# - min 30 %, max 100 %
# - 6h ulkoennusteen keskiarvo feed-forwardina
# - pakotettu suunta-askel (aina vähintään step_limit kohti setpointtia deadbandin ulkopuolella)
# - deadband, step-limit, monotonisuus ja selectin pykälöinti
# - oppiminen (RLS) jäädytetään defrostin aikana (sensor.faikin_liquid < 20)
# - parametrisäilö input_text.daikin_rls_params (vain theta tallennetaan, ei P)
#
# VAIHDA NÄMÄ ENTITYT:
# INDOOR = "sensor.home_temp" # sisälämpötila
# OUTDOOR = "sensor.outdoor_temp" # ulkolämpötila
# WEATHER = "weather.home" # weather.* jossa 'forecast'
# SELECT = "select.faikin_demand_control" # Daikin demand-select
# LIQUID = "sensor.faikin_liquid" # defrost-indikaattori (alle 20 = defrost)
# INDOOR_RATE, SP_HELPER, STEP_LIMIT_HELPER, DEADBAND_HELPER, PARAMS_TXT: ks. alta
import json
import time
from math import isfinite
# --------- ENTITYT (VAIHDA OMIKSI) ----------
INDOOR = "sensor.living_room_lampotila" # <-- sisälämpötila
INDOOR_RATE = "sensor.sisalampotila_aula" # derivative-sensori (°C/h)
OUTDOOR = "sensor.iv_tulo_lampotila" # <-- ulkolämpötila
WEATHER = "weather.koti" # <-- weather.* josta forecast-attribuutti
SELECT = "select.faikin_demand_control" # <-- Daikinin demand-select
LIQUID = "sensor.faikin_liquid" # <-- defrost-indikaattori
SP_HELPER = "input_number.daikin_setpoint"
STEP_LIMIT_HELPER = "input_number.daikin_step_limit"
DEADBAND_HELPER = "input_number.daikin_deadband"
# TÄHÄN TALLENNETAAN OPPIMINEN (pysyväksi)
STORE_ENTITY = "pyscript.daikin_ml_params"
# ---------- SÄÄTÖNUPIT ----------
HORIZON_H = 1.0 # tuntia; kuinka nopeasti pyritään korjaamaan virhettä
FORECAST_H = 6 # montako forecast-askelta mukaan
LAMBDA = 0.995 # RLS-unkohduskerroin
P0 = 1e4
MIN_DEM = 30.0
MAX_DEM = 100.0
KAPPA = 1.0 # kuinka aggressiivisesti virhettä korjataan (feed-forward taso)
# Defrost-cooldown
COOLDOWN_MINUTES = 15 # defrostin jälkeen hillitään nousua
COOLDOWN_STEP_UP = 3.0 # max ylös-muutos (%-yks/sykli) cooldownissa
# Icing band (staattinen raja) – -4…+4, cap 80 %
ICING_BAND_MIN = -4.0
ICING_BAND_MAX = 4.0
ICING_BAND_CAP = 80.0
# AUTO-TUNING rajat
AUTO_STEP_MIN = 3.0 # %-yks / sykli
AUTO_STEP_MAX = 20.0
AUTO_STEP_BASE = 10.0
AUTO_DB_MIN = 0.05 # °C
AUTO_DB_MAX = 0.50
AUTO_DB_BASE = 0.10
# ---------- Konteksti per ulkolämpöaste ----------
def _context_key_for_outdoor(Tout: float) -> str:
"""Pyöristetään ulkolämpö 1°C tarkkuudelle, esim. -6.7 -> '-7'."""
if not isfinite(Tout):
return "nan"
return str(int(round(Tout)))
# Moduulitason tila: mallit per aste
_theta_by_ctx = {} # esim. {"-7":[...], "0":[...], "3":[...]}
_P_by_ctx = {} # ei tallenneta pysyvästi, elää muistissa
_params_loaded = False
_last_defrosting = None
_cooldown_until = 0.0
# ---------- PYSYVÄ STORE pyscriptin state.persistillä ----------
state.persist(STORE_ENTITY)
def _load_params_from_store():
"""Lue per-aste thetat pysyvän entityn attribuuteista."""
global _theta_by_ctx
_theta_by_ctx = {}
attrs = state.getattr(STORE_ENTITY) or {}
tb = attrs.get("theta_by_ctx")
if isinstance(tb, dict):
for key, val in tb.items():
if isinstance(val, (list, tuple)) and len(val) == 4:
try:
th = [float(val[0]), float(val[1]), float(val[2]), float(val[3])]
_theta_by_ctx[str(key)] = th
except Exception:
continue
if _theta_by_ctx:
log.info("Daikin ML: loaded %d contexts from %s", len(_theta_by_ctx), STORE_ENTITY)
else:
log.info("Daikin ML: no stored thetas in %s, starting fresh", STORE_ENTITY)
def _save_params_to_store():
"""Tallenna kaikkien kontekstien thetat pysyvän entityn attribuutiksi."""
global _theta_by_ctx
clean = {}
for key, th in _theta_by_ctx.items():
if not isinstance(th, (list, tuple)) or len(th) != 4:
continue
clean[str(key)] = [
round(float(th[0]), 4),
round(float(th[1]), 4),
round(float(th[2]), 4),
round(float(th[3]), 4),
]
try:
state.set(
STORE_ENTITY,
value=time.time(),
theta_by_ctx=clean,
)
log.debug(
"Daikin ML: stored %d contexts into %s (attr size ~%d chars)",
len(clean),
STORE_ENTITY,
len(str(clean)),
)
except Exception as e:
log.error("Daikin ML: error saving thetas to %s: %s", STORE_ENTITY, e)
def _init_context_params_if_needed():
"""Lataa per-aste-mallit muistissa ja pysyvästä storagesta tarvittaessa."""
global _theta_by_ctx, _P_by_ctx, _params_loaded
if _params_loaded:
return
_theta_by_ctx = {}
_P_by_ctx = {}
_load_params_from_store()
# Alusta P jokaiselle olemassa olevalle kontekstille
for key in _theta_by_ctx.keys():
_P_by_ctx[str(key)] = [[P0, 0, 0, 0],
[0, P0, 0, 0],
[0, 0, P0, 0],
[0, 0, 0, P0]]
_params_loaded = True
# ---------- Matriisiapuja ----------
def _dot(a, b):
total = 0.0
for i in range(len(a)):
total += a[i] * b[i]
return total
def _matvec(M, v):
out = [0.0, 0.0, 0.0, 0.0]
for i in range(4):
s = 0.0
for j in range(4):
s += M[i][j] * v[j]
out[i] = s
return out
def _matsub(A, B):
C = [[0.0] * 4 for _ in range(4)]
for i in range(4):
for j in range(4):
C[i][j] = A[i][j] - B[i][j]
return C
def _matscale(A, s):
C = [[0.0] * 4 for _ in range(4)]
for i in range(4):
for j in range(4):
C[i][j] = s * A[i][j]
return C
def _rls_update(theta, P, x, y, lam=LAMBDA):
Px = _matvec(P, x)
denom = lam + _dot(x, Px)
if denom == 0:
denom = 1e-6
K = [v / denom for v in Px]
err_est = y - _dot(x, theta)
theta = [theta[0] + K[0] * err_est,
theta[1] + K[1] * err_est,
theta[2] + K[2] * err_est,
theta[3] + K[3] * err_est]
xTP = [[0.0] * 4 for _ in range(4)]
for i in range(4):
for j in range(4):
xTP[i][j] = x[i] * Px[j]
P = _matscale(_matsub(P, xTP), 1.0 / lam)
return theta, P
# ---------- Sääennuste ----------
def _avg_future_outdoor():
attrs = state.getattr(WEATHER) or {}
fc = attrs.get("forecast") or []
temps = []
idx = 0
for f in fc:
if idx >= FORECAST_H:
break
t = f.get("temperature")
if t is not None:
try:
temps.append(float(t))
idx += 1
except Exception:
pass
if len(temps) > 0:
s = 0.0
for v in temps:
s += v
return s / float(len(temps))
try:
return float(state.get(OUTDOOR) or 0.0)
except Exception:
return 0.0
# ---------- Selectin optiot & pykälöinti ----------
def _with_pct():
attrs = state.getattr(SELECT) or {}
opts = attrs.get("options") or []
return ('%' in (opts[0] if opts else ''))
def _select_options_nums():
attrs = state.getattr(SELECT) or {}
opts = attrs.get("options") or []
nums = []
for o in opts:
try:
s = str(o).replace('%', '').strip()
nums.append(float(s))
except Exception:
pass
nums.sort()
has_pct = False
if len(opts) > 0 and ('%' in str(opts[0])):
has_pct = True
return nums, has_pct
def _snap_to_select(value, direction):
nums, has_pct = _select_options_nums()
picked = None
if len(nums) == 0:
picked = round(value)
out = str(int(picked)) + ('%' if (has_pct or _with_pct()) else '')
return out
if direction > 0:
chosen = None
for n in nums:
if n >= value:
chosen = n
break
if chosen is None:
chosen = nums[-1]
picked = chosen
elif direction < 0:
chosen = None
for i in range(len(nums) - 1, -1, -1):
if nums[i] <= value:
chosen = nums[i]
break
if chosen is None:
chosen = nums[0]
picked = chosen
else:
best = nums[0]
bestd = abs(nums[0] - value)
for n in nums:
d = abs(n - value)
if d < bestd:
bestd = d
best = n
picked = best
if float(picked).is_integer():
picked = int(picked)
return str(picked) + ('%' if has_pct else '')
def _clip(v, lo, hi):
if v < lo:
return lo
if v > hi:
return hi
return v
# ---------- AUTO-TUNING (vain sisäiseen käyttöön + log.info) ----------
def _auto_tune_helpers(theta, ctx, step_limit_current, deadband_current):
"""
Laske automaattinen step_limit ja deadband thetan perusteella.
Käytetään erityisesti theta[2]: demandin vaikutus lämpötilan muutokseen.
EI kutsu HA-palveluja; käyttää arvoja sisäisesti ja loggaa ehdotuksen.
"""
try:
gain = abs(float(theta[2]))
except Exception:
gain = 1.0
if gain < 0.1:
gain = 0.1
if gain > 10.0:
gain = 10.0
step_raw = AUTO_STEP_BASE / gain
step_auto = _clip(step_raw, AUTO_STEP_MIN, AUTO_STEP_MAX)
db_raw = AUTO_DB_BASE * (1.0 + 0.2 * gain)
db_auto = _clip(db_raw, AUTO_DB_MIN, AUTO_DB_MAX)
step_auto_rounded = round(step_auto, 1)
db_auto_rounded = round(db_auto, 2)
if (abs(step_auto_rounded - step_limit_current) > 0.1 or
abs(db_auto_rounded - deadband_current) > 0.01):
log.info(
"Daikin ML AUTO-TUNE suggestion (ctx=%s): gain=%.3f -> "
"step_limit=%.1f (was %.1f), deadband=%.2f (was %.2f)",
ctx, gain,
step_auto_rounded, step_limit_current,
db_auto_rounded, deadband_current,
)
# Palautetaan suoraan autotunetut arvot sisäiseen käyttöön
return step_auto_rounded, db_auto_rounded
# ---------- Varsinainen ohjain ----------
@state_trigger(f"{INDOOR}", f"{OUTDOOR}", f"{INDOOR_RATE}")
@time_trigger("cron(*/6 * * * *)")
def daikin_ml_controller(**kwargs):
global _last_defrosting, _cooldown_until, _theta_by_ctx, _P_by_ctx
_init_context_params_if_needed()
try:
Tin = float(state.get(INDOOR))
except Exception:
Tin = float("nan")
try:
Tout = float(state.get(OUTDOOR))
except Exception:
Tout = float("nan")
try:
rate = float(state.get(INDOOR_RATE) or 0.0)
except Exception:
rate = 0.0
sp = float(state.get(SP_HELPER) or 22.5)
step_limit_current = float(state.get(STEP_LIMIT_HELPER) or 10.0)
deadband_current = float(state.get(DEADBAND_HELPER) or 0.1)
prev_str = state.get(SELECT) or ""
try:
prev = float(prev_str.replace('%', ''))
except Exception:
prev = 60.0
try:
liquid = float(state.get(LIQUID) or 100.0)
except Exception:
liquid = 100.0
defrosting = (liquid < 20.0)
now = time.time()
if _last_defrosting is None:
_last_defrosting = defrosting
if (_last_defrosting is True) and (defrosting is False):
_cooldown_until = now + COOLDOWN_MINUTES * 60.0
log.info("Daikin ML: defrost ended -> cooldown active for %d min", COOLDOWN_MINUTES)
_last_defrosting = defrosting
in_cooldown = (now < _cooldown_until)
if not (isfinite(Tin) and isfinite(Tout)):
log.info("Daikin ML: sensors not ready; Tin=%s Tout=%s", state.get(INDOOR), state.get(OUTDOOR))
return
ctx = _context_key_for_outdoor(Tout)
if ctx not in _theta_by_ctx:
_theta_by_ctx[ctx] = [0.0, 0.0, 5.0, 0.0]
if ctx not in _P_by_ctx:
_P_by_ctx[ctx] = [[P0, 0, 0, 0],
[0, P0, 0, 0],
[0, 0, P0, 0],
[0, 0, 0, P0]]
theta = _theta_by_ctx[ctx]
P = _P_by_ctx[ctx]
# Auto-tune käyttöön sisäisesti
step_limit, deadband = _auto_tune_helpers(theta, ctx, step_limit_current, deadband_current)
in_icing_band = (Tout >= ICING_BAND_MIN and Tout <= ICING_BAND_MAX)
band_upper = min(MAX_DEM, (ICING_BAND_CAP if in_icing_band else 100.0))
if prev > band_upper:
option_cap = _snap_to_select(band_upper, -1)
if option_cap and option_cap != prev_str:
select.select_option(entity_id=SELECT, option=option_cap)
log.info(
"Daikin ML: CAP ENFORCED: prev=%s -> %s (upper=%.0f%%, ctx=%s)",
prev_str, option_cap, band_upper, ctx,
)
return
prev_eff = prev if prev <= band_upper else band_upper
err = sp - Tin
demand_norm = _clip(prev_eff / 100.0, 0.0, 1.0)
x = [1.0, err, demand_norm, (Tout - Tin)]
y = rate
if not defrosting:
theta, P = _rls_update(theta, P, x, y)
_theta_by_ctx[ctx] = theta
_P_by_ctx[ctx] = P
_save_params_to_store()
else:
_theta_by_ctx[ctx] = theta
_P_by_ctx[ctx] = P
Tout_future = _avg_future_outdoor()
Tout_eff = 0.5 * Tout + 0.5 * Tout_future
dTin_target = _clip(KAPPA * err / HORIZON_H, -2.0, 2.0)
num = dTin_target - (theta[0] + theta[1] * err + theta[3] * (Tout_eff - Tin))
denom_raw = theta[2]
denom_sign = 1.0 if denom_raw >= 0 else -1.0
denom_mag = abs(denom_raw)
if denom_mag < 0.5:
denom_mag = 0.5
dem_opt = _clip((num / (denom_mag * denom_sign)) * 100.0, 0.0, 100.0)
if abs(err) <= deadband:
dem_target = prev_eff
else:
dem_target = dem_opt
delta = dem_target - prev_eff
if delta > step_limit:
dem_target = prev_eff + step_limit
elif delta < -step_limit:
dem_target = prev_eff - step_limit
dem_clip = _clip(dem_target, MIN_DEM, band_upper)
direction = 0
if Tin < sp - deadband:
direction = 1
elif Tin > sp + deadband:
direction = -1
if direction == 1:
up_limit = COOLDOWN_STEP_UP if in_cooldown else step_limit
forced = _clip(prev_eff + up_limit, MIN_DEM, band_upper)
dem_mono = dem_clip if dem_clip > forced else forced
elif direction == -1:
forced = _clip(prev_eff - step_limit, MIN_DEM, band_upper)
dem_mono = dem_clip if dem_clip < forced else forced
else:
forced = prev_eff
dem_mono = prev_eff
option = _snap_to_select(dem_mono, direction)
try:
opt_num = float(str(option).replace('%', ''))
except Exception:
opt_num = 999.0
if opt_num > band_upper:
option = _snap_to_select(band_upper, -1)
log.info("Daikin ML: post-cap clamp -> %s (upper=%.0f%%, ctx=%s)", option, band_upper, ctx)
if option and option != prev_str:
select.select_option(entity_id=SELECT, option=option)
theta_str = "[" + ", ".join([str(round(float(v), 4)) for v in theta]) + "]"
cool_str = "ACTIVE" if in_cooldown else "off"
icing_str = "ON" if in_icing_band else "off"
log.info(
"Daikin ML: ctx=%s | Tin=%.2f°C, Tout=%.1f°C (→%.1f°C), DEFROST=%s, cooldown=%s, "
"icing_band=%s(cap=%s), theta=%s | "
"SP=%.2f, step_limit=%.1f, deadband=%.2f | "
"prev=%s → opt≈%.0f%% → "
"clip=%.0f%% → forced=%.0f%% → select=%s",
ctx, Tin, Tout, Tout_future, str(defrosting), cool_str,
icing_str, (str(ICING_BAND_CAP) if in_icing_band else "off"),
theta_str, sp, step_limit, deadband,
prev_str, dem_opt, dem_clip, forced, option,
)
# ---------- CAP-VARTIJA ----------
@state_trigger(f"{SELECT}")
def daikin_ml_cap_guard(value=None, **kwargs):
try:
Tout = float(state.get(OUTDOOR))
except Exception:
return
in_icing_band = (Tout >= ICING_BAND_MIN and Tout <= ICING_BAND_MAX)
band_upper = min(MAX_DEM, (ICING_BAND_CAP if in_icing_band else 100.0))
curr_str = state.get(SELECT) or ""
try:
curr = float(str(curr_str).replace('%', ''))
except Exception:
return
if curr > band_upper:
option_cap = _snap_to_select(band_upper, -1)
if option_cap and option_cap != curr_str:
select.select_option(entity_id=SELECT, option=option_cap)
log.info(
"Daikin ML: CAP GUARD: %s -> %s (upper=%.0f%%)",
curr_str, option_cap, band_upper,
)
# ---------- Käsin ajettava askel ----------
@service
def daikin_ml_step():
"""Aja ohjain kerran (voi kutsua HA-automaatioista)."""
try:
daikin_ml_controller()
except Exception as e:
log.error("Daikin ML step error: %s", e)