|
@@ -5,7 +5,9 @@ Kolm voogu:
|
|
|
|
|
|
|
|
1. Temperatuur/niiskus (``MQTT_TOPIC``): mõõtmised koondatakse mälus täistunni
|
|
1. Temperatuur/niiskus (``MQTT_TOPIC``): mõõtmised koondatakse mälus täistunni
|
|
|
kaupa ja kirjutatakse iga täistunni järel üks agregeeritud rida
|
|
kaupa ja kirjutatakse iga täistunni järel üks agregeeritud rida
|
|
|
- (``CH_TABLE``, vaikimisi ``uus67.temperatuur``).
|
|
|
|
|
|
|
+ (``CH_TABLE``, vaikimisi ``uus67.temperatuur``). Kui tunnis mõõtmisi ei ole
|
|
|
|
|
+ (andur kättesaamatu), kirjutatakse ikkagi rida eelmise teadaoleva väärtusega
|
|
|
|
|
+ ja märgitakse ``taidetud = true`` — nii ei teki aegreas auke.
|
|
|
2. Kontaktandurid (``DOOR_TOPICS``): salvestatakse **ainult olekumuutused**
|
|
2. Kontaktandurid (``DOOR_TOPICS``): salvestatakse **ainult olekumuutused**
|
|
|
(``CH_DOOR_TABLE``, vaikimisi ``uus67.ukseandur``).
|
|
(``CH_DOOR_TABLE``, vaikimisi ``uus67.ukseandur``).
|
|
|
3. Häired (``CH_FAULT_TABLE``, vaikimisi ``uus67.vead``): **ainult muutused**
|
|
3. Häired (``CH_FAULT_TABLE``, vaikimisi ``uus67.vead``): **ainult muutused**
|
|
@@ -138,15 +140,29 @@ class HourBucket:
|
|
|
def empty(self) -> bool:
|
|
def empty(self) -> bool:
|
|
|
return not self.temps and not self.hums
|
|
return not self.temps and not self.hums
|
|
|
|
|
|
|
|
- def row(self) -> dict[str, Any]:
|
|
|
|
|
|
|
+ def row(self, fallback: dict[str, float | None] | None = None) -> dict[str, Any]:
|
|
|
|
|
+ """Ühe tunni rida.
|
|
|
|
|
+
|
|
|
|
|
+ Kui tunnis mõõtmisi ei ole ja ``fallback`` on antud, kasutatakse
|
|
|
|
|
+ eelmise teadaoleva väärtuse andmeid ja märgitakse ``taidetud = true``.
|
|
|
|
|
+ """
|
|
|
def avg(values: list[float]) -> float | None:
|
|
def avg(values: list[float]) -> float | None:
|
|
|
return round(sum(values) / len(values), 2) if values else None
|
|
return round(sum(values) / len(values), 2) if values else None
|
|
|
|
|
|
|
|
|
|
+ if self.empty and fallback is not None:
|
|
|
|
|
+ return {
|
|
|
|
|
+ "hour": self.hour.strftime("%Y-%m-%d %H:%M:%S"),
|
|
|
|
|
+ "device": DEVICE,
|
|
|
|
|
+ "temperature": fallback.get("temperature"),
|
|
|
|
|
+ "humidity": fallback.get("humidity"),
|
|
|
|
|
+ "taidetud": True,
|
|
|
|
|
+ }
|
|
|
return {
|
|
return {
|
|
|
"hour": self.hour.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"hour": self.hour.strftime("%Y-%m-%d %H:%M:%S"),
|
|
|
"device": DEVICE,
|
|
"device": DEVICE,
|
|
|
"temperature": avg(self.temps),
|
|
"temperature": avg(self.temps),
|
|
|
"humidity": avg(self.hums),
|
|
"humidity": avg(self.hums),
|
|
|
|
|
+ "taidetud": False,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@@ -380,6 +396,27 @@ def load_last_seen() -> dict[str, datetime]:
|
|
|
return {}
|
|
return {}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+def load_last_temp() -> dict[str, float | None]:
|
|
|
|
|
+ """Loeb andmebaasist viimase teadaoleva temperatuuri/niiskuse (täitmise baas)."""
|
|
|
|
|
+ query = (
|
|
|
|
|
+ f"SELECT temperature, humidity FROM {CH_DB}.{CH_TABLE} "
|
|
|
|
|
+ "ORDER BY hour DESC LIMIT 1 FORMAT JSONEachRow"
|
|
|
|
|
+ )
|
|
|
|
|
+ try:
|
|
|
|
|
+ text = _ch_query(query).strip()
|
|
|
|
|
+ if text:
|
|
|
|
|
+ obj = json.loads(text.splitlines()[0])
|
|
|
|
|
+ values = {
|
|
|
|
|
+ "temperature": _as_float(obj.get("temperature")),
|
|
|
|
|
+ "humidity": _as_float(obj.get("humidity")),
|
|
|
|
|
+ }
|
|
|
|
|
+ LOG.info("Seemendatud viimane väärtus: %s", values)
|
|
|
|
|
+ return values
|
|
|
|
|
+ except Exception:
|
|
|
|
|
+ LOG.exception("Viimase väärtuse laadimine ebaõnnestus, alustan tühjalt")
|
|
|
|
|
+ return {"temperature": None, "humidity": None}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
def main() -> int:
|
|
def main() -> int:
|
|
|
lock = threading.Lock()
|
|
lock = threading.Lock()
|
|
|
stop = threading.Event()
|
|
stop = threading.Event()
|
|
@@ -389,6 +426,7 @@ def main() -> int:
|
|
|
fault_writer = ClickHouseWriter(CH_FAULT_TABLE)
|
|
fault_writer = ClickHouseWriter(CH_FAULT_TABLE)
|
|
|
door_tracker = DoorTracker(load_last_states())
|
|
door_tracker = DoorTracker(load_last_states())
|
|
|
fault_tracker = FaultTracker(load_active_faults())
|
|
fault_tracker = FaultTracker(load_active_faults())
|
|
|
|
|
+ last_values: dict[str, float | None] = load_last_temp()
|
|
|
|
|
|
|
|
start = datetime.now(TZ)
|
|
start = datetime.now(TZ)
|
|
|
last_seen: dict[str, datetime] = {}
|
|
last_seen: dict[str, datetime] = {}
|
|
@@ -449,6 +487,12 @@ def main() -> int:
|
|
|
if msg.topic == MQTT_TOPIC:
|
|
if msg.topic == MQTT_TOPIC:
|
|
|
bucket.add(payload)
|
|
bucket.add(payload)
|
|
|
last_seen[DEVICE] = now
|
|
last_seen[DEVICE] = now
|
|
|
|
|
+ temp = _as_float(payload.get("temperature"))
|
|
|
|
|
+ hum = _as_float(payload.get("humidity"))
|
|
|
|
|
+ if temp is not None:
|
|
|
|
|
+ last_values["temperature"] = temp
|
|
|
|
|
+ if hum is not None:
|
|
|
|
|
+ last_values["humidity"] = hum
|
|
|
fault_writer.enqueue_many(fault_tracker.on_device_message(DEVICE, payload, now))
|
|
fault_writer.enqueue_many(fault_tracker.on_device_message(DEVICE, payload, now))
|
|
|
elif msg.topic in DOOR_TOPICS:
|
|
elif msg.topic in DOOR_TOPICS:
|
|
|
device = msg.topic.rsplit("/", 1)[-1]
|
|
device = msg.topic.rsplit("/", 1)[-1]
|
|
@@ -510,8 +554,20 @@ def main() -> int:
|
|
|
bucket = HourBucket(current_hour(now))
|
|
bucket = HourBucket(current_hour(now))
|
|
|
else:
|
|
else:
|
|
|
finished = None
|
|
finished = None
|
|
|
- if finished is not None and not finished.empty:
|
|
|
|
|
- temp_writer.enqueue(finished.row())
|
|
|
|
|
|
|
+ if finished is not None:
|
|
|
|
|
+ if not finished.empty:
|
|
|
|
|
+ temp_writer.enqueue(finished.row())
|
|
|
|
|
+ elif (
|
|
|
|
|
+ last_values.get("temperature") is not None
|
|
|
|
|
+ or last_values.get("humidity") is not None
|
|
|
|
|
+ ):
|
|
|
|
|
+ LOG.info(
|
|
|
|
|
+ "Tund %s ilma mõõtmisteta — täidan eelmise väärtusega (%s / %s)",
|
|
|
|
|
+ finished.hour.strftime("%Y-%m-%d %H:%M"),
|
|
|
|
|
+ last_values.get("temperature"),
|
|
|
|
|
+ last_values.get("humidity"),
|
|
|
|
|
+ )
|
|
|
|
|
+ temp_writer.enqueue(finished.row(fallback=last_values))
|
|
|
for device in sorted(MONITORED_DEVICES):
|
|
for device in sorted(MONITORED_DEVICES):
|
|
|
seen = last_seen.get(device, start)
|
|
seen = last_seen.get(device, start)
|
|
|
timeout = timedelta(minutes=offline_timeout(device))
|
|
timeout = timedelta(minutes=offline_timeout(device))
|