KNX config file creation script from CSV exports
This commit is contained in:
Martin Bauer
parent
639f14a438
commit
6d2a193fb4
|
|
@ -0,0 +1,180 @@
|
|||
from typing import List
|
||||
from util import DeviceInfo
|
||||
|
||||
__all__ = ['extent', 'import_ets5_csv_file', 'create_power_plug', 'create_lights', 'create_shutters', 'create_switches']
|
||||
|
||||
|
||||
def first_lower(s):
|
||||
if len(s) == 0:
|
||||
return s
|
||||
else:
|
||||
return s[0].lower() + s[1:]
|
||||
|
||||
|
||||
def extent(result_dict, input_dict):
|
||||
for k, v in input_dict.items():
|
||||
if k not in result_dict:
|
||||
result_dict[k] = []
|
||||
for entry in v:
|
||||
entry['platform'] = 'knx_data'
|
||||
result_dict[k] += v
|
||||
|
||||
|
||||
def import_ets5_csv_file(csv_file):
|
||||
result = dict()
|
||||
with open(csv_file, encoding='utf-8') as f:
|
||||
for line in f:
|
||||
splitted_line = line.split(",")
|
||||
name = splitted_line[0].replace('"', "")
|
||||
group_address = splitted_line[1].replace('"', "")
|
||||
if '-' not in group_address:
|
||||
result[name.strip()] = splitted_line[1].replace('"', "").strip()
|
||||
return result
|
||||
|
||||
|
||||
def create_power_plug(device_info: List[DeviceInfo],
|
||||
csv_contents,
|
||||
postfix_on_off_write=" Schalten",
|
||||
postfix_on_off_read=" RM Schalten",
|
||||
postfix_counter_value=" BSZ Wert",
|
||||
postfix_consumption=" Verbrauch",
|
||||
postfix_consumption_sum=" VerbrauchSumme",
|
||||
postfix_consumption_reset=" Verbrauch Neustart",
|
||||
postfix_counter_reset=" BSZ Neustart"):
|
||||
result = {'switch': [],
|
||||
'sensor': []}
|
||||
for entry in device_info:
|
||||
try:
|
||||
# Switching
|
||||
on_off_write_addr = csv_contents.get(entry.csv_name + postfix_on_off_write, None)
|
||||
on_off_read_addr = csv_contents.get(entry.csv_name + postfix_on_off_read, None)
|
||||
if on_off_write_addr:
|
||||
device_entry = {
|
||||
'name': entry.display_name,
|
||||
'address': on_off_write_addr,
|
||||
}
|
||||
if on_off_read_addr:
|
||||
device_entry['state_address'] = on_off_read_addr
|
||||
result['switch'].append(device_entry)
|
||||
|
||||
# Counter for time (in hours) how long device was switched on
|
||||
counter_value = csv_contents.get(entry.csv_name + postfix_counter_value, None)
|
||||
if counter_value:
|
||||
result['sensor'].append({
|
||||
'name': entry.display_name + " Betriebsstunden",
|
||||
'address': counter_value,
|
||||
'type': '2byte_unsigned',
|
||||
})
|
||||
counter_reset = csv_contents.get(entry.csv_name + postfix_counter_reset, None)
|
||||
if counter_reset:
|
||||
result['switch'].append({
|
||||
'name': entry.display_name + " Betriebsstunden Reset",
|
||||
'address': counter_reset
|
||||
})
|
||||
|
||||
# Consumption
|
||||
consumption = csv_contents.get(entry.csv_name + postfix_consumption, None)
|
||||
if consumption:
|
||||
result['sensor'].append({
|
||||
'name': entry.display_name + " Verbrauch",
|
||||
'address': consumption,
|
||||
'type': 'current',
|
||||
})
|
||||
consumption_sum = csv_contents.get(entry.csv_name + postfix_consumption_sum, None)
|
||||
if consumption_sum:
|
||||
result['sensor'].append({
|
||||
'name': entry.display_name + " Verbrauch Summe",
|
||||
'address': consumption_sum,
|
||||
'type': '4byte_unsigned',
|
||||
})
|
||||
consumption_reset = csv_contents.get(entry.csv_name + postfix_consumption_reset, None)
|
||||
if consumption_reset:
|
||||
result['switch'].append({
|
||||
'name': entry.display_name + " Verbrauch Reset",
|
||||
'address': consumption_reset
|
||||
})
|
||||
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Skipping light {entry.csv_name} - Could not find CSV File entry: {e}")
|
||||
return result
|
||||
|
||||
|
||||
def create_lights(device_info: List[DeviceInfo],
|
||||
csv_contents,
|
||||
postfix_on_off_write=" Schalten",
|
||||
postfix_on_off_read=" RM Schalten",
|
||||
postfix_brightness_write=" Helligkeit",
|
||||
postfix_brightness_read=" RM Helligkeit"):
|
||||
result = []
|
||||
for entry in device_info:
|
||||
try:
|
||||
on_off_write_addr = csv_contents[entry.csv_name + postfix_on_off_write]
|
||||
on_off_read_addr = csv_contents.get(entry.csv_name + postfix_on_off_read, None)
|
||||
brightness_write_addr = csv_contents.get(entry.csv_name + postfix_brightness_write, None)
|
||||
brightness_read_addr = csv_contents.get(entry.csv_name + postfix_brightness_read, None)
|
||||
|
||||
entry = {
|
||||
'name': entry.display_name,
|
||||
'address': on_off_write_addr,
|
||||
}
|
||||
if on_off_read_addr:
|
||||
entry['state_address'] = on_off_read_addr
|
||||
if brightness_write_addr:
|
||||
entry['brightness_address'] = brightness_write_addr
|
||||
if brightness_read_addr:
|
||||
entry['brightness_state_address'] = brightness_read_addr
|
||||
result.append(entry)
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Skipping light {entry.csv_name} - Could not find CSV File entry: {e}")
|
||||
|
||||
return {'light': result}
|
||||
|
||||
|
||||
def create_switches(device_info: List[DeviceInfo], csv_contents,
|
||||
postfix_on_off_write=" Schalten",
|
||||
postfix_on_off_read=" RM Schalten"):
|
||||
result = []
|
||||
for entry in device_info:
|
||||
try:
|
||||
on_off_write_addr = csv_contents.get(entry.csv_name + postfix_on_off_write, None)
|
||||
if on_off_write_addr is None:
|
||||
on_off_write_addr = csv_contents[entry.csv_name]
|
||||
on_off_read_addr = csv_contents.get(entry.csv_name + postfix_on_off_read, None)
|
||||
|
||||
entry = {
|
||||
'name': entry.display_name,
|
||||
'address': on_off_write_addr,
|
||||
}
|
||||
if on_off_read_addr:
|
||||
entry['state_address'] = on_off_read_addr
|
||||
result.append(entry)
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Skipping switch {entry.csv_name} - Could not find CSV File entry: {e}")
|
||||
|
||||
return {'switch': result}
|
||||
|
||||
|
||||
def create_shutters(device_info: List[DeviceInfo],
|
||||
csv_contents,
|
||||
postfix_long_addr=" Lang", postfix_short_addr=" Kurz", postfix_read_position=" RM Position",
|
||||
postfix_set_position=" Position"):
|
||||
result = []
|
||||
|
||||
for entry in device_info:
|
||||
try:
|
||||
long_addr = csv_contents[entry.csv_name + postfix_long_addr]
|
||||
short_addr = csv_contents[entry.csv_name + postfix_short_addr]
|
||||
read_pos_addr = csv_contents[entry.csv_name + postfix_read_position]
|
||||
set_pos_addr = csv_contents[entry.csv_name + postfix_set_position]
|
||||
|
||||
result.append({
|
||||
'name': entry.display_name,
|
||||
'move_long_address': long_addr,
|
||||
'move_short_address': short_addr,
|
||||
'position_address': set_pos_addr,
|
||||
'position_state_address': read_pos_addr,
|
||||
})
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Skipping shutter {entry.csv_name} - Could not find CSV File entry: {e}")
|
||||
|
||||
return {'cover': result}
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
from util import DeviceInfo
|
||||
from ruamel.yaml import YAML
|
||||
import knx_conf as knx
|
||||
|
||||
|
||||
def create_config():
|
||||
lights = [
|
||||
# Dimmers
|
||||
DeviceInfo('Wohnzimmerlampe', 'Wohnzimmer Deckenlampe'),
|
||||
DeviceInfo('EsszimmerlampeWest', 'Esszimmer Deckenlampe West'),
|
||||
DeviceInfo('EsszimmerlampeMitte', 'Esszimmer Deckenlampe Mitte'),
|
||||
DeviceInfo("EsszimmerWandlampe", 'Esszimmer Schrankleuchte'),
|
||||
DeviceInfo("Küchenlampe", "Küche Deckenlampe"),
|
||||
DeviceInfo("AussenleuchteUntenSO", "Aussen Terassenlicht"),
|
||||
DeviceInfo("Gang", "Gang Licht"),
|
||||
DeviceInfo("Bad", "Bad Licht"),
|
||||
DeviceInfo("GangWindfang", "Gang Einganglicht"),
|
||||
DeviceInfo("LichtWaschküche", "Waschküche Licht"),
|
||||
# Normal lights
|
||||
DeviceInfo('AussenleuchteHaustüren', 'Haustür Licht'),
|
||||
DeviceInfo('AussenleuchteObenNW', 'Haustür Licht NW'),
|
||||
DeviceInfo('TreppenhausLicht', "Treppenhaus Licht"),
|
||||
DeviceInfo('WCLicht', "WC Licht"),
|
||||
DeviceInfo('LampeVorratsraum', "Vorratsraum Licht"),
|
||||
]
|
||||
|
||||
shutters = [
|
||||
DeviceInfo('Wohnzimmer Fenster Rollo', 'Wohnzimmer Fenster Rollo'),
|
||||
DeviceInfo('Terassentür Rollo', 'Wohnzimmer Terrassentür Rollo'),
|
||||
DeviceInfo('Küchenfenster Rollo', 'Küche Fenster Rollo'),
|
||||
DeviceInfo('Esszimmerfenster Rollo', 'Esszimmer Fenster Rollo'),
|
||||
]
|
||||
|
||||
switches = [
|
||||
# Bells
|
||||
DeviceInfo("KlingelOben", "Klingel Oben"),
|
||||
DeviceInfo("Klingel Innen", "Klingel Innentür"),
|
||||
DeviceInfo("Klingel Aussen", "Klingen Außentür"),
|
||||
# Bewegungsmelder LEDs
|
||||
DeviceInfo("BewegungsmelderMitte LED", "Bewegungsmelder Mitte LED"),
|
||||
DeviceInfo("BewegungsmelderWest LED", "Bewegungsmelder West LED"),
|
||||
DeviceInfo("BewegungsmelderOst LED", "Bewegungsmelder Ost LED"),
|
||||
]
|
||||
scene_button_names = ['ObenLinks', 'ObenRechts', 'MitteLinks', 'MitteRechts', 'UntenLinks', 'UntenRechts']
|
||||
scene_button_names = [(i, e) for i, e in enumerate(scene_button_names)]
|
||||
switches += [DeviceInfo(f"SzeneEsszimmer{n}", "Esszimmer Szene {i}") for i, n in scene_button_names]
|
||||
switches += [DeviceInfo(f"SzeneWohnzimmer{n}", "Wohnzimmer Szene {i}") for i, n in scene_button_names]
|
||||
switches += [DeviceInfo(f"SzeneEingang{n}", "Eingang Szene {i}") for i, n in scene_button_names[2:]]
|
||||
switches += [DeviceInfo(f"SzeneTerrassentuer{n}", "Wohnzimmer Terrassentür Szene {i}")
|
||||
for i, n in scene_button_names[2: 4]]
|
||||
|
||||
power_plugs = [
|
||||
# Vorratsraum
|
||||
DeviceInfo("VorratsraumSteckdose1", "Vorratsraum Steckdose"),
|
||||
DeviceInfo("VorratsraumSteckdose2", "Vorratsraum Steckdose"),
|
||||
DeviceInfo("VorratsraumSteckdose3", "Gefrierschrank"),
|
||||
# Waschraum
|
||||
DeviceInfo("Trockner", "Trockner"),
|
||||
DeviceInfo("Waschmaschine", "Waschmaschine"),
|
||||
# Küche
|
||||
DeviceInfo("KücheSteckdose1", "Küche Steckdose 1"),
|
||||
DeviceInfo("KücheSteckdose2", "Küche Steckdose 2"),
|
||||
DeviceInfo("Spülmaschine", "Spülmaschine"),
|
||||
DeviceInfo("Backofen", "Backofen"),
|
||||
DeviceInfo("HerdP1", "Herd Phase 1"),
|
||||
DeviceInfo("HerdP2", "Herd Phase 2"),
|
||||
DeviceInfo("HerdP3", "Herd Phase 3"),
|
||||
# Rest
|
||||
DeviceInfo("ArbeitszimmerSteckdose", "Arbeitszimmer Steckdose"),
|
||||
DeviceInfo("WohnzimmerSteckdose1", "Wohnzimmer Steckdose 1"),
|
||||
DeviceInfo("WohnzimmerSteckdose2", "Wohnzimmer Steckdose 2"),
|
||||
]
|
||||
|
||||
imported_csv = knx.import_ets5_csv_file('knx_data/export_project1.csv')
|
||||
imported_csv.update(knx.import_ets5_csv_file('knx_data/export_project2.csv'))
|
||||
|
||||
devices = {}
|
||||
knx.extent(devices, knx.create_lights(lights, imported_csv))
|
||||
knx.extent(devices, knx.create_shutters(shutters, imported_csv))
|
||||
knx.extent(devices, knx.create_switches(switches, imported_csv))
|
||||
knx.extent(devices, knx.create_power_plug(power_plugs, imported_csv))
|
||||
YAML().dump(devices, open('knx_devices.yaml', 'w'))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
create_config()
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
from collections import namedtuple
|
||||
|
||||
DeviceInfo = namedtuple("DeviceInfo", ['csv_name', 'display_name'])
|
||||
Loading…
Reference in New Issue