From 8b34069fc6a226f385232d6e5141616cb252dfee Mon Sep 17 00:00:00 2001 From: Pim van den Berg Date: Sat, 2 Jan 2021 15:32:04 +0100 Subject: [PATCH] feat(convert-itho-db): convert an itho servicetool database to sqlite The SQLite database (a subset of the Itho database) can be used to dynamically interact with an Itho WPU based on the WPU version. > usage: convert-itho-db.py [-h] --itho-db [ITHO_DB] [--sqlite-db [SQLITE_DB]] > [--force] > > Convert Itho Servicetool database to SQLite > > optional arguments: > -h, --help show this help message and exit > --itho-db [ITHO_DB] Itho Database file (default: None) > --sqlite-db [SQLITE_DB] > Itho Database file (default: heatpump.sqlite) > --force Force overwrite SQLite database (default: False) --- convert-itho-db.py | 74 ++++++++++++++++++++++++++++++++++++++++ db.py | 85 ++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 2 ++ 3 files changed, 161 insertions(+) create mode 100755 convert-itho-db.py create mode 100644 db.py diff --git a/convert-itho-db.py b/convert-itho-db.py new file mode 100755 index 0000000..bbb5434 --- /dev/null +++ b/convert-itho-db.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +# +# Convert the required data for python-itho-wpu from Itho Servicetool +# $_parameters_HeatPump.par (Microsoft Access database) to SQLite SQL. +# +# Dependencies: python3-pyodbc, mdbtools (>= 0.9.0), odbc-mdbtools (>= 0.9.0) + +import argparse +import db +import os +import pyodbc +import re +import sys + + +def parse_args(): + parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter, + description='Convert Itho Servicetool database to SQLite') + parser.add_argument('--itho-db', nargs='?', required=True, help="Itho Database file") + parser.add_argument('--sqlite-db', nargs='?', default='heatpump.sqlite', help="Itho Database file") + parser.add_argument('--force', action='store_true', help="Force overwrite SQLite database") + args = parser.parse_args() + return args + + +def convert(par_file, sqlite_db): + par_file = par_file.replace("$", "\\$") + par_conn = pyodbc.connect(f'DRIVER={{MDBTools}};DBQ={par_file};') + par_conn.setencoding('UTF-8') + par_conn.setdecoding(pyodbc.SQL_CHAR, encoding='UTF-8') + par_cur = par_conn.cursor() + + sqlite_db = db.sqlite(sqlite_db) + + tables = [] + for table_info in par_cur.tables(tableType='TABLE'): + if re.match('^(VersieBeheer|Data[Ll]abel|Parameterlijst)', table_info.table_name): + tables.append(table_info.table_name) + + for t in sorted(tables): + sqlite_db.create_table(t.lower()) + data = [] + if re.match("^Data[Ll]abel", t): + par_cur.execute(f"select Index, Naam, Tekst_NL, Tooltip_NL, Eenheid_NL from {t}") + rows = par_cur.fetchall() + for r in sorted(rows): + data.append((r.Index, r.Naam, r.Tekst_NL, r.Tooltip_NL, r.Eenheid_NL)) + if re.match("^Parameterlijst", t): + par_cur.execute(f"select Index, Naam, Naam_fabriek, Min, Max, Default, Tekst_NL, Omschrijving_NL, Eenheid_NL from {t}") + rows = par_cur.fetchall() + for r in sorted(rows): + data.append((r.Index, r.Naam, r.Naam_fabriek, r.Min, r.Max, r.Default, r.Tekst_NL, r.Omschrijving_NL, r.Eenheid_NL)) + if re.match("^VersieBeheer", t): + par_cur.execute(f"select VersieNummer, DataLabel, ParameterLijst from {t}") + rows = par_cur.fetchall() + for r in sorted(rows): + data.append((r.VersieNummer, r.DataLabel, r.ParameterLijst)) + sqlite_db.insert(t.lower(), data) + + +if __name__ == "__main__": + args = parse_args() + print(f"Converting {args.itho_db} to {args.sqlite_db} ...") + if not os.path.exists(args.itho_db): + print(f"Itho database does not exist: {args.itho_db}") + sys.exit(1) + if os.path.exists(args.sqlite_db): + if args.force: + print(f"Removing existing SQLite database: {args.sqlite_db}") + os.remove(args.sqlite_db) + else: + print(f"Error: SQLite database exists: {args.sqlite_db}") + sys.exit(1) + convert(args.itho_db, args.sqlite_db) diff --git a/db.py b/db.py new file mode 100644 index 0000000..c7367c6 --- /dev/null +++ b/db.py @@ -0,0 +1,85 @@ +import sqlite3 +from sqlite3 import Error + + +class sqlite(): + def __init__(self, db_file): + self.conn = self.connect(db_file) + + def connect(self, db_file): + conn = None + try: + conn = sqlite3.connect(db_file) + return conn + except Error as e: + print(e) + return conn + + def execute(self, query, params=()): + try: + self.conn.row_factory = sqlite3.Row + c = self.conn.cursor() + c.execute(query, params) + return [dict(row) for row in c.fetchall()] + except Error as e: + print("sqlite_execute failed for: {}, {}".format(query, params)) + print("Error:", e) + + def executemany(self, query, data): + try: + c = self.conn.cursor() + c.executemany(query, data) + except Error as e: + print("sqlite_executemany failed for: {}, {}".format(query, data)) + print("Error:", e) + + def create_table(self, t): + if t.startswith('datalabel'): + query = """CREATE TABLE {} ( + id real, + name text, + title text, + tooltip text, + unit text + );""".format(t) + elif t.startswith('parameterlijst'): + query = """ + CREATE TABLE {} ( + id real, + name text, + name_factory text, + min real, + max real, + def real, + title text, + description text, + unit text + );""".format(t) + elif t.startswith('versiebeheer'): + query = """ + CREATE TABLE {} ( + version integer primary key, + datalabel integer, + parameterlist integer + );""".format(t) + self.execute(query) + self.conn.commit() + + def insert(self, t, data): + if t.startswith('datalabel'): + query = """ + INSERT INTO {} (id, name, title, tooltip, unit) + VALUES (?, ?, ?, ?, ?); + """.format(t) + elif t.startswith('parameterlijst'): + query = """ + INSERT INTO {} (id, name, name_factory, min, max, def, title, description, unit) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?); + """.format(t) + elif t.startswith('versiebeheer'): + query = """ + INSERT INTO {} (version, datalabel, parameterlist) + VALUES (?, ?, ?); + """.format(t) + self.executemany(query, data) + self.conn.commit() diff --git a/requirements.txt b/requirements.txt index ad1e415..94d804f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,3 @@ pigpio +pyodbc +sqlite3