# -*- coding: utf-8 -*- from __future__ import annotations __title__ = "building" __author__ = 'Mehmet Karatay & "Saraswati" (ChatGPT)' __purpose__ = "Bina / daire / sensör ve brülör topolojisi sürücüsü" __version__ = "0.4.3" __date__ = "2025-11-22" """ ebuild/core/building.py Revision : 2025-11-22 Authors : Mehmet Karatay & "Saraswati" (ChatGPT) Amaç ----- - Statik konfigürasyondan (config_statics) bina, geo ve sensör topolojisini yükler. - FLAT_AREA içindeki DS18B20 sensörlerini daire/oda bazında yönetir. - Bina ortalama sıcaklık istatistiklerini üretir. - Brülör grupları (BURNER_GROUPS) ve hat sensörleri (dış, çıkış, dönüş) için başlangıçta okunabilir bir özet verir. - SeasonController ile mevsim / güneş / tatil bilgisi sağlar. Not --- Bu modül, hem bağımsız test aracı olarak `python3 -m ebuild.core.building` şeklinde çalışabilir, hem de BurnerController gibi üst seviye kontrol modülleri tarafından kullanılabilir. """ import os import json import time import datetime import statistics from collections import defaultdict from typing import Dict, List, Optional, Any from .season import SeasonController from ..io.dbtext import DBText from .. import config_statics as cfg_s from .. import config_runtime as cfg_v # ------------------------------------------------------------------ # Yardımcı sabitler # ------------------------------------------------------------------ FlatDirection = { 0: "Kuzey-Kuzey", 1: "Kuzey-Doğu", 2: "Kuzey-Batı", 3: "Güney-Güney", 4: "Güney-Doğu", 5: "Güney-Batı", 6: "Doğu-Doğu", 7: "Batı-Batı", } ValveType = { 0: "Yok", 1: "Dijital", 2: "PWM", 3: "Analog", 4: "Enkoder", 5: "Modbus", 6: "CANbus", } # ------------------------------------------------------------------ # DS18B20 sensör sınıfı # ------------------------------------------------------------------ class EDSensor: """Tek bir DS18B20 sensörünü temsil eder.""" def __init__(self, serial: str, flat_no: int, flat_direction: int, floor: int, room_no: int = 0) -> None: self.serial = serial self.flat_no = flat_no self.floor = floor self.room_no = room_no self.flat_direction = flat_direction self.device_path = f"/sys/bus/w1/devices/{serial}/w1_slave" self.flat_check = 0 self.is_connected = True def read_temperature(self) -> Optional[float]: try: with open(self.device_path, "r") as f: lines = f.readlines() if lines[0].strip().endswith("YES"): pos = lines[1].find("t=") if pos != -1: raw = lines[1][pos + 2 :] t_c = float(raw) / 1000.0 self.is_connected = True return t_c except Exception: self.flat_check += 1 self.is_connected = False return None # ------------------------------------------------------------------ # Daire nesnesi # ------------------------------------------------------------------ class Flat: """Daire nesnesi: yön, kat ve sensör ilişkisi.""" def __init__(self, flat_no: int, flat_direction: int, sensor: EDSensor, floor: int, room_no: int = 0) -> None: self.flat_no = flat_no self.flat_direction = flat_direction self.sensor = sensor self.floor = floor self.room_no = room_no self.last_heat: Optional[float] = None self.last_read_time: Optional[datetime.datetime] = None # ------------------------------------------------------------------ # Bina sınıfı # ------------------------------------------------------------------ class Building: """ Bina seviyesi sürücü. config_statics üzerinden: BUILDING_NAME BUILDING_LOCATION BUILDING_LABEL GEO_CITY GEO_COUNTRY GEO_TZ GEO_LAT GEO_LON FLAT_AREA BURNER_GROUPS OUTSIDE_SENSOR_ID / NAME BURNER_OUT_SENSOR_ID / NAME RETURN_LINE_SENSOR_IDS / NAME_MAP """ def __init__( self, name: Optional[str] = None, location: Optional[str] = None, bina_adi: Optional[str] = None, logtime: bool = False, logger: Optional[DBText] = None, ) -> None: # Bina bilgileri config_statics'ten self.name = name or getattr(cfg_s, "BUILDING_NAME", "Bina") self.location = location or getattr(cfg_s, "BUILDING_LOCATION", "Ankara") self.bina_adi = bina_adi or getattr(cfg_s, "BUILDING_LABEL", "A_Binasi") # Coğrafi bilgiler self.geo_city = getattr(cfg_s, "GEO_CITY", "Ankara") self.geo_country = getattr(cfg_s, "GEO_COUNTRY", "Turkey") self.geo_tz = getattr(cfg_s, "GEO_TZ", "Europe/Istanbul") self.geo_lat = float(getattr(cfg_s, "GEO_LAT", 39.92077)) self.geo_lon = float(getattr(cfg_s, "GEO_LON", 32.85411)) # Logger self.logger = logger or DBText( filename=getattr(cfg_s, "BUILDING_LOG_FILE", "ebina_log.sql"), table=getattr(cfg_s, "BUILDING_LOG_TABLE", "ebrulor_log"), app="ESYSTEM", ) # Çalışma aralığı: 3 dakika (eski koddaki gibi) interval_min = getattr(cfg_v, "BUILDING_SAMPLE_MINUTES", 3) self.interval = datetime.timedelta(seconds=60 * interval_min) self.next_run = datetime.datetime.now() self.flats: Dict[int, Flat] = {} self.last_max_floor: Optional[int] = None self.logtime = logtime # Season / tatil bilgisi try: self.season = SeasonController.from_now() except Exception as e: print(f"⚠️ SeasonController.from_now() hata: {e}") self.season = None # Daireleri FLAT_AREA'dan oluştur self._load_statics_from_flat_area() # Bina genel istatistiği (ilk deneme) try: self.get_building_temperature_stats() except Exception: pass # Konsol özet self._print_header() self._print_sun_and_season() self._print_burner_topology() self._print_line_sensors() self.print_config() def get_hat_sensor_by_id(self, sensor_id: str): """ Verilen DS18B20 ID'sine göre hat sensör nesnesini döndür. Yoksa None döndür. """ return self.hat_sensors.get(sensor_id) # yapıya göre uyarlarsın # ------------------------------------------------------------------ # Statik konfigürasyondan daire/sensör yükleme # ------------------------------------------------------------------ def _load_statics_from_flat_area(self) -> None: """config_statics.FLAT_AREA içinden daire/sensörleri yükler.""" self.flats = {} flat_area = getattr(cfg_s, "FLAT_AREA", []) for item in flat_area: serial = item.get("serial") flat_no = int(item.get("flat_no", 0)) room_no = int(item.get("room_no", 0)) floor = int(item.get("floor", 0)) direction = int(item.get("direction", 0)) if not serial: continue sensor = EDSensor(serial, flat_no, direction, floor, room_no=room_no) # bağlantı testi sensor.read_temperature() flat = Flat(flat_no, direction, sensor, floor, room_no=room_no) self.flats[flat_no] = flat print("✅ Building: statics yüklendi") # ------------------------------------------------------------------ # Başlangıç çıktıları # ------------------------------------------------------------------ def _print_header(self) -> None: print(f"🏢 Bina adı : {self.bina_adi}") print(f"📍 Konum : {self.name} / {self.location}") print( f"🌍 Geo : {self.geo_city}, {self.geo_country} " f"({self.geo_lat}, {self.geo_lon}) tz={self.geo_tz}" ) print(f"🏠 Daire/oda sayısı: {len(self.flats)}\n") def _print_sun_and_season(self) -> None: """SeasonController üzerinden sunrise / sunset ve mevsim / tatil bilgisini yazar.""" if not self.season or not getattr(self.season, "info", None): print("☀️ Sunrise/Sunset : --:-- / --:--") print("🧭 Mevsim / Tatil : (bilgi yok)\n") return info = self.season.info sunrise = info.sunrise.strftime("%H:%M") if info.sunrise else "--:--" sunset = info.sunset.strftime("%H:%M") if info.sunset else "--:--" print(f"☀️ Sunrise/Sunset : {sunrise} / {sunset}") if info.season_start and info.season_end: print( f"🧭 Mevsim / Tatil : {info.season} " f"({info.season_start} → {info.season_end}), " f"Tatil={info.is_holiday} {info.holiday_label}" ) else: print( f"🧭 Mevsim / Tatil : {info.season}, " f"Tatil={info.is_holiday} {info.holiday_label}" ) print() def _print_burner_topology(self) -> None: """config_statics.BURNER_GROUPS bilgisini okunaklı yazar.""" groups = getattr(cfg_s, "BURNER_GROUPS", {}) if not groups: print("🔥 Brülör topolojisi: tanımlı değil (BURNER_GROUPS boş).\n") return print("🔥 Brülör Topolojisi (config_statics.BURNER_GROUPS):") for bid, info in groups.items(): name = info.get("name", f"Burner{bid}") loc = info.get("location", "-") ign = info.get("igniter_pin", None) print(f" #{bid}: {name} @ {loc}") print(f" Igniter pin : {ign}") circ = info.get("circulation", {}) if not circ: print(" Pompalar : (tanımlı değil)") else: print(" Pompalar :") for cname, cinfo in circ.items(): pin = cinfo.get("pin") default = cinfo.get("default", 0) print(f" - {cname:<7} → pin={pin:<3} default={default}") print() # ------------------------------------------------------------------ # Hat sensörleri: dış / çıkış / dönüş # ------------------------------------------------------------------ def _read_ds18b20_raw(self, serial: str) -> Optional[float]: """Seri numarası verilen DS18B20'den tek seferlik okuma.""" path = f"/sys/bus/w1/devices/{serial}/w1_slave" try: with open(path, "r") as f: lines = f.readlines() if not lines or not lines[0].strip().endswith("YES"): return None pos = lines[1].find("t=") if pos == -1: return None raw = lines[1][pos + 2 :] return float(raw) / 1000.0 except Exception: return None def _print_line_sensors(self) -> None: """ Dış ısı, kazan çıkış ısısı ve dönüş hat sensörlerini (varsa) ekrana yazar. config_statics içindeki: OUTSIDE_SENSOR_ID / OUTSIDE_SENSOR_NAME BURNER_OUT_SENSOR_ID / BURNER_OUT_SENSOR_NAME RETURN_LINE_SENSOR_IDS RETURN_LINE_SENSOR_NAME_MAP kullanılır. """ outside_id = getattr(cfg_s, "OUTSIDE_SENSOR_ID", None) outside_nm = getattr(cfg_s, "OUTSIDE_SENSOR_NAME", "Dış Isı 1") burner_id = getattr(cfg_s, "BURNER_OUT_SENSOR_ID", None) burner_nm = getattr(cfg_s, "BURNER_OUT_SENSOR_NAME", "Çıkış Isı 1") return_ids = list(getattr(cfg_s, "RETURN_LINE_SENSOR_IDS", [])) name_map = dict(getattr(cfg_s, "RETURN_LINE_SENSOR_NAME_MAP", {})) if not (outside_id or burner_id or return_ids): # Hiç tanım yoksa sessizce geç return print("🌡️ Hat sensörleri:") # Dış ısı if outside_id: t = self._read_ds18b20_raw(outside_id) if t is None: print(f" - {outside_nm:<15}: ❌ Okunamadı ({outside_id})") else: print(f" - {outside_nm:<15}: {t:5.2f}°C ({outside_id})") # Kazan çıkış if burner_id: t = self._read_ds18b20_raw(burner_id) if t is None: print(f" - {burner_nm:<15}: ❌ Okunamadı ({burner_id})") else: print(f" - {burner_nm:<15}: {t:5.2f}°C ({burner_id})") # Dönüş hatları for sid in return_ids: nm = name_map.get(sid, sid) t = self._read_ds18b20_raw(sid) if t is None: print(f" - Dönüş {nm:<10}: ❌ Okunamadı ({sid})") else: print(f" - Dönüş {nm:<10}: {t:5.2f}°C ({sid})") print() # ------------------------------------------------------------------ # İstatistik fonksiyonları # ------------------------------------------------------------------ def _summarize(self, values: List[float]) -> dict: if not values: return {"min": None, "max": None, "avg": None, "count": 0} return { "min": min(values), "max": max(values), "avg": round(statistics.mean(values), 2), "count": len(values), } def get_temperature_stats(self) -> str: stats = { "floors": defaultdict(list), "directions": defaultdict(list), "top_floor": [], "building": [], } max_floor = self.last_max_floor for flat in self.flats.values(): temp = flat.sensor.read_temperature() if temp is not None: flat.last_heat = temp flat.last_read_time = datetime.datetime.now() stats["floors"][flat.floor].append(temp) stats["directions"][flat.flat_direction].append(temp) stats["building"].append(temp) if max_floor is not None and flat.floor == max_floor: stats["top_floor"].append(temp) result = { "floors": {f: self._summarize(v) for f, v in stats["floors"].items()}, "directions": { d: self._summarize(v) for d, v in stats["directions"].items() }, "top_floor": self._summarize(stats["top_floor"]), "building": self._summarize(stats["building"]), } return json.dumps(result, indent=4, ensure_ascii=False) def get_last_heat_report(self) -> str: report = [] for flat in self.flats.values(): temp = flat.sensor.read_temperature() now = datetime.datetime.now() if temp is not None: flat.last_heat = temp flat.last_read_time = now report.append( { "flat_no": flat.flat_no, "room_no": flat.room_no, "direction": flat.flat_direction, "floor": flat.floor, "serial": flat.sensor.serial, "last_heat": flat.last_heat, "last_read_time": flat.last_read_time.strftime( "%Y-%m-%d %H:%M:%S" ) if flat.last_read_time else None, "is_connected": flat.sensor.is_connected, "flat_check": flat.sensor.flat_check, } ) return json.dumps(report, indent=4, ensure_ascii=False) def get_building_temperature_stats(self) -> Optional[dict]: print("\n🌡️ Bina Genel Isı Bilgisi:") all_temps: List[float] = [] floor_temps: Dict[int, List[float]] = {} now = datetime.datetime.now() # Log zamanı mı? if now >= self.next_run: self.logtime = True self.next_run = now + self.interval else: self.logtime = False self.last_max_floor = None for flat in self.flats.values(): temp = flat.sensor.read_temperature() floor = flat.floor if temp is not None: flat.last_heat = temp flat.last_read_time = now if self.last_max_floor is None or floor > self.last_max_floor: self.last_max_floor = floor else: flat.last_heat = None flat.last_read_time = None floor_temps.setdefault(floor, []) if temp is not None: all_temps.append(temp) floor_temps[floor].append(temp) if self.logtime: self.logger.insert_event( source=f"Sensor:{flat.sensor.serial}", event_type="temperature", value=temp, unit="°C", timestamp=now, extra=f"Daire {flat.flat_no} Oda {flat.room_no} Kat {flat.floor}, Yön {flat.flat_direction}", ) if not all_temps or not floor_temps: print("❌ Hiç sıcaklık verisi alınamadı.\n") return None min_temp = min(all_temps) max_temp = max(all_temps) avg_temp = sum(all_temps) / len(all_temps) min_floor = min(floor_temps.keys()) max_floor = max(floor_temps.keys()) avg_min_floor = sum(floor_temps[min_floor]) / max(len(floor_temps[min_floor]), 1) avg_max_floor = sum(floor_temps[max_floor]) / max(len(floor_temps[max_floor]), 1) delta_t = abs(avg_max_floor - avg_min_floor) print( " Min: {:.2f}°C | Max: {:.2f}°C | Avg: {:.2f}°C | Count: {}".format( min_temp, max_temp, avg_temp, len(all_temps) ) ) print(f" En Alt Kat ({min_floor}): {avg_min_floor:.2f}°C") print(f" En Üst Kat ({max_floor}): {avg_max_floor:.2f}°C") print(f" 🔺 Delta Isı (Üst - Alt): {delta_t:.2f}°C") if self.logtime: self.logger.insert_event( source="Sensor:min", event_type="temperature", value=min_temp, unit="°C", timestamp=now, extra="", ) self.logger.insert_event( source="Sensor:max", event_type="temperature", value=max_temp, unit="°C", timestamp=now, extra="", ) self.logger.insert_event( source="Sensor:avg", event_type="temperature", value=avg_temp, unit="°C", timestamp=now, extra="", ) return { "min": min_temp, "max": max_temp, "avg": avg_temp, "count": len(all_temps), "min_floor": min_floor, "max_floor": max_floor, "avg_min_floor": avg_min_floor, "avg_max_floor": avg_max_floor, "delta_t": delta_t, } # ------------------------------------------------------------------ # Yardımcı # ------------------------------------------------------------------ def print_config(self) -> None: if not self.flats: print(f"📦 {self.bina_adi} için tanımlı daire yok.") return print(f"\n📦 {self.bina_adi} içeriği:") for flat_no, flat in self.flats.items(): temp = flat.sensor.read_temperature() durum = "✅ Bağlı" if temp is not None else "❌ Yok" print( " Daire {:<3} Oda {:<1} | Kat: {:<2} | Yön: {:<12} | Sensör: {:<18} | Durum: {} | AKTIF".format( flat.flat_no, flat.room_no, flat.floor, FlatDirection.get(flat.flat_direction, "Bilinmiyor"), flat.sensor.serial, durum, ) ) def run_forever(self, sleep_s: int = 5) -> None: """Eski test aracı gibi: sürekli bina istatistiği üretir.""" while True: self.get_building_temperature_stats() time.sleep(sleep_s) # ---------------------------------------------------------------------- # CLI # ---------------------------------------------------------------------- if __name__ == "__main__": bina = Building() # Config'ten her şeyi alır bina.run_forever()