ebuild_rasp2/ebuild/io/ds18b20.py

159 lines
4.8 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
from __future__ import annotations
__title__ = "ds18b20"
__author__ = 'Mehmet Karatay & "Saraswati" (ChatGPT)'
__purpose__ = "DS18B20 1-Wire sıcaklık sensörü sürücüsü"
__version__ = "0.1.0"
__date__ = "2025-11-21"
"""
ebuild/io/ds18b20.py
Revision : 2025-11-21
Authors : Mehmet Karatay & "Saraswati" (ChatGPT)
Amaç
-----
- DS18B20 sensörlerini 1-Wire üzerinden /sys/bus/w1/devices yolundan okuyarak
sıcaklık (°C) bilgisi sağlamak.
- Tek sensör için DS18B20Sensor sınıfı,
- Otomatik cihaz keşfi için DS18B20Bus yardımcı sınıfı sunar.
Notlar
------
- 1-Wire kernel modüllerinin (w1_gpio, w1_therm) yüklü olması gerekir.
- Bu sürücü yalnızca dosya sisteminden okuma yapar; filtreleme, smoothing,
bina/daire eşlemesi gibi işlemler üst katmanlarda yapılır.
"""
import glob
import os
from typing import Dict, List, Optional
class DS18B20Sensor:
"""
Tek bir DS18B20 sensörünü temsil eder.
Özellikler:
-----------
- serial : 1-Wire cihaz id'si (örn: "28-00000660e983")
- base_path : /sys/bus/w1/devices (varsayılan)
- read_temperature() : son sıcaklığı °C cinsinden döndürür
"""
def __init__(
self,
serial: str,
base_path: str = "/sys/bus/w1/devices",
name: Optional[str] = None,
) -> None:
self.serial = serial
self.base_path = base_path
self.device_path = os.path.join(base_path, serial, "w1_slave")
self.name = name or serial
self.is_connected: bool = True
self.error_count: int = 0
self.last_temperature: Optional[float] = None
# ------------------------------------------------------------------
def read_temperature(self) -> Optional[float]:
"""
Sensörden anlık sıcaklık okur (°C).
Dönüş:
- Başarı: float (örn: 23.437)
- Hata: None (error_count artar, is_connected False olur)
"""
try:
with open(self.device_path, "r") as f:
lines = f.readlines()
if not lines:
raise IOError("w1_slave boş okundu")
# İlk satır CRC ve 'YES/NO' bilgisini içerir.
if not lines[0].strip().endswith("YES"):
raise IOError("CRC hatalı veya sensör doğrulanamadı")
# İkinci satırda 't=xxxxx' ifadesini arıyoruz.
pos = lines[1].find("t=")
if pos == -1:
raise ValueError("t= alanı bulunamadı")
raw = lines[1][pos + 2 :].strip()
t_c = float(raw) / 1000.0
self.is_connected = True
self.last_temperature = t_c
return t_c
except Exception:
self.error_count += 1
self.is_connected = False
self.last_temperature = None
return None
# ------------------------------------------------------------------
def exists(self) -> bool:
"""
Cihaz dosyasının mevcut olup olmadığını kontrol eder.
"""
return os.path.exists(self.device_path)
def summary(self) -> str:
"""
Sensör hakkında kısa bir özet döndürür.
"""
status = "OK" if self.is_connected else "ERR"
return f"DS18B20Sensor(name={self.name}, serial={self.serial}, status={status}, errors={self.error_count})"
class DS18B20Bus:
"""
Bir 1-Wire hattı üzerindeki DS18B20 cihazlarını keşfetmek için yardımcı sınıf.
"""
def __init__(self, base_path: str = "/sys/bus/w1/devices") -> None:
self.base_path = base_path
def discover(self) -> List[str]:
"""
Sistemdeki tüm DS18B20 cihazlarının seri numaralarını listeler.
Örnek:
["28-00000660e983", "28-0000066144f9", ...]
"""
pattern = os.path.join(self.base_path, "28-*")
devices = glob.glob(pattern)
serials = [os.path.basename(p) for p in devices]
return serials
def get_sensors(self) -> Dict[str, DS18B20Sensor]:
"""
Otomatik keşif yaparak her seri için bir DS18B20Sensor nesnesi döndürür.
"""
result: Dict[str, DS18B20Sensor] = {}
for serial in self.discover():
result[serial] = DS18B20Sensor(serial=serial, base_path=self.base_path)
return result
# ----------------------------------------------------------------------
# Basit test
# ----------------------------------------------------------------------
if __name__ == "__main__":
bus = DS18B20Bus()
serials = bus.discover()
print("Bulunan DS18B20 cihazları:")
for s in serials:
print(" -", s)
sensors = bus.get_sensors()
for serial, sensor in sensors.items():
t = sensor.read_temperature()
print(f"{serial}: {t} °C ({sensor.summary()})")