311 lines
11 KiB
Python
311 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tahoma Module
|
|
Enthält NUR Tahoma-spezifische Geräte-Discovery Logik
|
|
KEINE Datenbank-Operationen!
|
|
"""
|
|
|
|
import requests
|
|
import urllib3
|
|
import re
|
|
import logging
|
|
from typing import List, Dict, Optional, Tuple
|
|
from modules.base_module import BaseModule
|
|
|
|
# SSL-Warnungen deaktivieren
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TahomaAPI:
|
|
"""Original TahomaAPI Klasse - unverändert"""
|
|
|
|
def __init__(self, gateway_ip: str, api_token: str):
|
|
self.base_url = f"https://{gateway_ip}:8443/enduser-mobile-web/1/enduserAPI"
|
|
self.headers = {
|
|
"Authorization": f"Bearer {api_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
def get_setup(self) -> Optional[Dict]:
|
|
try:
|
|
url = f"{self.base_url}/setup"
|
|
response = requests.get(url, headers=self.headers, verify=False, timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Fehler beim Abrufen der Setup-Daten: {e}")
|
|
return None
|
|
|
|
def get_devices(self) -> List[Dict]:
|
|
setup = self.get_setup()
|
|
if not setup:
|
|
return []
|
|
devices = setup.get('devices', [])
|
|
logger.info(f"{len(devices)} Tahoma-Geräte gefunden")
|
|
return devices
|
|
|
|
|
|
class DeviceClassifier:
|
|
"""Original DeviceClassifier - unverändert"""
|
|
|
|
ACTOR_TYPES = {
|
|
'RollerShutter', 'ExteriorScreen', 'Awning', 'Blind',
|
|
'GarageDoor', 'Window', 'Light', 'OnOff', 'DimmableLight',
|
|
'HeatingSystem', 'Valve', 'Switch', 'Door', 'Curtain',
|
|
'VenetianBlind', 'PergolaScreen'
|
|
}
|
|
|
|
SENSOR_TYPES = {
|
|
'TemperatureSensor', 'LightSensor', 'HumiditySensor',
|
|
'ContactSensor', 'OccupancySensor', 'SmokeSensor',
|
|
'WaterDetectionSensor', 'WindowHandle', 'MotionSensor',
|
|
'SunSensor', 'WindSensor', 'RainSensor', 'ConsumptionSensor'
|
|
}
|
|
|
|
# Tahoma Commands mit Parametern
|
|
TAHOMA_COMMANDS = {
|
|
"setClosure": [{"name": "position", "type": "integer", "min": 0, "max": 100}],
|
|
"setClosureAndOrientation": [
|
|
{"name": "position", "type": "integer", "min": 0, "max": 100},
|
|
{"name": "neigung", "type": "integer", "min": 0, "max": 100}
|
|
],
|
|
"setOrientation": [{"name": "neigung", "type": "integer", "min": 0, "max": 100}],
|
|
"up": [], "down": [], "my": [], "stop": [], "refresh": [], "wink":[],
|
|
"setMyPosition": [{"name": "position", "type": "integer", "min": 0, "max": 100}],
|
|
"on": [], "off": [], "toggle": [],
|
|
"setIntensity": [{"name": "helligkeit", "type": "integer", "min": 0, "max": 100}],
|
|
"setColor": [
|
|
{"name": "farbton", "type": "integer", "min": 0, "max": 360},
|
|
{"name": "sättigung", "type": "integer", "min": 0, "max": 100}
|
|
],
|
|
"setColorTemperature": [{"name": "farbtemperatur", "type": "integer", "min": 2000, "max": 6500}],
|
|
"setTargetTemperature": [{"name": "temperatur", "type": "float", "min": 5.0, "max": 30.0}],
|
|
"setMode": [{"name": "betriebsart", "type": "string"}],
|
|
"pulse": [{"name": "impuls_dauer", "type": "integer", "min": 1, "max": 3600}],
|
|
"setLevel": [{"name": "ausgangs_level", "type": "integer", "min": 0, "max": 100}],
|
|
"trigger": [],
|
|
}
|
|
|
|
@classmethod
|
|
def is_actor(cls, device: Dict) -> bool:
|
|
"""Prüft ob Gerät ein Aktor ist"""
|
|
device_type = device.get('controllableName', device.get('uiClass', ''))
|
|
|
|
if device_type in cls.ACTOR_TYPES:
|
|
return True
|
|
|
|
commands = device.get('definition', {}).get('commands', [])
|
|
if commands:
|
|
command_names = [cmd.get('commandName', '') for cmd in commands]
|
|
actor_commands = {'open', 'close', 'on', 'off', 'up', 'down', 'setPosition', 'dim'}
|
|
if any(cmd in actor_commands for cmd in command_names):
|
|
return True
|
|
|
|
return False
|
|
|
|
@classmethod
|
|
def is_sensor(cls, device: Dict) -> bool:
|
|
"""Prüft ob Gerät ein Sensor ist"""
|
|
device_type = device.get('controllableName', device.get('uiClass', ''))
|
|
|
|
if device_type in cls.SENSOR_TYPES:
|
|
return True
|
|
|
|
states = device.get('states', [])
|
|
commands = device.get('definition', {}).get('commands', [])
|
|
|
|
if states and len(states) > 0 and len(commands) <= 1:
|
|
return True
|
|
|
|
return False
|
|
|
|
@classmethod
|
|
def extract_actor_data(cls, device: Dict) -> tuple:
|
|
"""Extrahiert Commands und States aus Aktor"""
|
|
commands = []
|
|
states = []
|
|
|
|
cmd_definitions = device.get('definition', {}).get('commands', [])
|
|
|
|
for cmd in cmd_definitions:
|
|
command_name = cmd.get('commandName', '')
|
|
cmd_params = cls.TAHOMA_COMMANDS.get(command_name, "Not in List")
|
|
if(cmd_params != "Not in List"): #only append command, if it is on of the listed commands, to prevent flooding the DB with bullshit.
|
|
command_entry = {
|
|
'command': command_name,
|
|
'parameters': []
|
|
}
|
|
|
|
for cmd_param in cmd_params:
|
|
param_detail = {'name': cmd_param.get('name', '')}
|
|
|
|
if 'type' in cmd_param:
|
|
param_detail['type'] = cmd_param['type']
|
|
if 'min' in cmd_param:
|
|
param_detail['min'] = cmd_param['min']
|
|
if 'max' in cmd_param:
|
|
param_detail['max'] = cmd_param['max']
|
|
if 'values' in cmd_param:
|
|
param_detail['values'] = cmd_param['values']
|
|
|
|
if param_detail['name']:
|
|
command_entry['parameters'].append(param_detail)
|
|
|
|
commands.append(command_entry)
|
|
|
|
# States extrahieren
|
|
state_definitions = device.get('states', [])
|
|
for state in state_definitions:
|
|
state_name = state.get('name', '')
|
|
if state_name:
|
|
state_entry = {
|
|
'name': state_name,
|
|
'type': state.get('type', 0)
|
|
}
|
|
if 'value' in state:
|
|
state_entry['current_value'] = state['value']
|
|
states.append(state_entry)
|
|
|
|
return commands, states
|
|
|
|
@classmethod
|
|
def extract_sensor_data(cls, device: Dict) -> list:
|
|
"""Extrahiert States aus Sensor"""
|
|
states = []
|
|
|
|
state_definitions = device.get('states', [])
|
|
for state in state_definitions:
|
|
state_name = state.get('name', '')
|
|
if state_name:
|
|
state_entry = {
|
|
'name': state_name,
|
|
'type': state.get('type', 0)
|
|
}
|
|
if 'value' in state:
|
|
state_entry['current_value'] = state['value']
|
|
states.append(state_entry)
|
|
|
|
return states
|
|
|
|
|
|
class TahomaModule(BaseModule):
|
|
"""
|
|
Tahoma Modul - Implementiert BaseModule Interface
|
|
Gibt nur Actors/Sensors zurück, KEINE DB-Operationen
|
|
"""
|
|
|
|
def is_enabled(self) -> bool:
|
|
"""Prüft ob Tahoma aktiviert ist"""
|
|
return (self.config.tahoma_enable and
|
|
self.config.tahoma_ip and
|
|
self.config.tahoma_token)
|
|
|
|
def discover(self) -> Tuple[List[Dict], List[Dict]]:
|
|
"""
|
|
Führt Tahoma Discovery durch
|
|
|
|
Returns:
|
|
Tuple (actors, sensors) - Listen von Dicts im vereinheitlichten Format
|
|
"""
|
|
logger.info("\n" + "=" * 60)
|
|
logger.info("TAHOMA-GERÄTE WERDEN ABGERUFEN")
|
|
logger.info("=" * 60)
|
|
|
|
actors = []
|
|
sensors = []
|
|
|
|
# TahomaAPI initialisieren
|
|
tahoma = TahomaAPI(self.config.tahoma_ip, self.config.tahoma_token)
|
|
devices = tahoma.get_devices()
|
|
|
|
if not devices:
|
|
logger.warning("Keine Tahoma-Geräte gefunden")
|
|
return actors, sensors
|
|
|
|
# Geräte gruppieren (Original-Logik)
|
|
device_groups = {}
|
|
standalone_devices = []
|
|
|
|
for device in devices:
|
|
device_url = device.get('deviceURL', '')
|
|
match = re.match(r'(.+)#(\d+)$', device_url)
|
|
|
|
if match:
|
|
base_url = match.group(1)
|
|
if base_url not in device_groups:
|
|
device_groups[base_url] = []
|
|
device_groups[base_url].append(device)
|
|
else:
|
|
standalone_devices.append(device)
|
|
|
|
# Gruppierte Geräte verarbeiten
|
|
for base_url, group_devices in device_groups.items():
|
|
main_device = None
|
|
for dev in group_devices:
|
|
if dev.get('deviceURL', '').endswith('#1'):
|
|
main_device = dev
|
|
break
|
|
|
|
if not main_device and group_devices:
|
|
main_device = group_devices[0]
|
|
|
|
main_name = main_device.get('label', 'Unbekannt') if main_device else 'Unbekannt'
|
|
|
|
for device in group_devices:
|
|
actor, sensor = self._process_device(device, main_name)
|
|
if actor:
|
|
actors.append(actor)
|
|
if sensor:
|
|
sensors.append(sensor)
|
|
|
|
# Standalone Geräte verarbeiten
|
|
for device in standalone_devices:
|
|
device_name = device.get('label', 'Unbekannt')
|
|
actor, sensor = self._process_device(device, device_name)
|
|
if actor:
|
|
actors.append(actor)
|
|
if sensor:
|
|
sensors.append(sensor)
|
|
|
|
logger.info(f"Tahoma: {len(actors)} Aktoren, {len(sensors)} Sensoren gefunden")
|
|
return actors, sensors
|
|
|
|
def _process_device(self, device: Dict, device_name: str) -> Tuple[Optional[Dict], Optional[Dict]]:
|
|
"""
|
|
Verarbeitet ein einzelnes Gerät
|
|
|
|
Returns:
|
|
Tuple (actor_dict or None, sensor_dict or None)
|
|
"""
|
|
device_url = device.get('deviceURL', '')
|
|
device_type = device.get('controllableName', device.get('uiClass', 'Unknown'))
|
|
|
|
is_actor = DeviceClassifier.is_actor(device)
|
|
is_sensor = DeviceClassifier.is_sensor(device)
|
|
|
|
actor = None
|
|
sensor = None
|
|
|
|
if is_actor:
|
|
commands, states = DeviceClassifier.extract_actor_data(device)
|
|
actor = {
|
|
'type': device_type,
|
|
'name': device_name,
|
|
'url': device_url,
|
|
'commands': commands,
|
|
'states': states
|
|
}
|
|
|
|
elif is_sensor:
|
|
states = DeviceClassifier.extract_sensor_data(device)
|
|
sensor = {
|
|
'type': device_type,
|
|
'name': device_name,
|
|
'url': device_url,
|
|
'states': states
|
|
}
|
|
|
|
return actor, sensor
|