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)
This commit is contained in:
Pim van den Berg 2021-01-02 15:32:04 +01:00
parent d537ab80a3
commit 8b34069fc6
3 changed files with 161 additions and 0 deletions

74
convert-itho-db.py Executable file
View File

@ -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)

85
db.py Normal file
View File

@ -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()

View File

@ -1 +1,3 @@
pigpio
pyodbc
sqlite3