From 4c0576377b6d68f31aada6edfcb29653f9966633 Mon Sep 17 00:00:00 2001 From: "Maxime Alves LIRMM@home" Date: Mon, 27 Jul 2020 16:00:32 +0200 Subject: [PATCH] [var_type] added the models, the database base methods, and the tests --- db/pyheatpump.db | Bin 32768 -> 32768 bytes pyheatpump/db.py | 48 ++++++++++++++++++-- pyheatpump/models/__init__.py | 0 pyheatpump/models/variable.py | 18 ++++++++ pyheatpump/models/variable_type.py | 42 +++++++++++++++++ pyheatpump/models/variable_value.py | 18 ++++++++ pyheatpump/variable_types.py | 54 ++++++++++++++++++++++ tests/test_variable_types.py | 67 ++++++++++++++++++++++++++++ 8 files changed, 244 insertions(+), 3 deletions(-) create mode 100644 pyheatpump/models/__init__.py create mode 100644 pyheatpump/models/variable.py create mode 100644 pyheatpump/models/variable_type.py create mode 100644 pyheatpump/models/variable_value.py create mode 100644 pyheatpump/variable_types.py create mode 100644 tests/test_variable_types.py diff --git a/db/pyheatpump.db b/db/pyheatpump.db index 492027527d56477810999688e7f08314c47c92a6..26df93a9019f7ef6b55994611559a5f32630fe95 100644 GIT binary patch delta 203 zcmZo@U}|V!njp={GEv5vk!54T5`Hdb{#^|GulO(X@7gRVu!cWGoSBVZTAGu|#U(R6 zvm`MmDL+3aH8GF*g(wqHLYRrk(=)FmH9fT`Gp|IrOCm>vk&RzcoRi7XF)uMEKRqoc zKe2@I3l{?e12g}72L9jt@A2VBvos^rW@Zo@Y8ewJvm__PhD8DjiyRaHmpC;0 delta 30 mcmZo@U}|V!njp={I8nx#k#S?f68_CB4u9kq1sE(!PyhgrItji2 diff --git a/pyheatpump/db.py b/pyheatpump/db.py index 6babe8d..c27799e 100755 --- a/pyheatpump/db.py +++ b/pyheatpump/db.py @@ -1,9 +1,51 @@ #!/usr/bin/env python3 import sqlite3 -from ..conf import config +from subprocess import Popen +from .conf import config +import sys + +conn = None + +def connect(): + global conn + if conn is None: + print('Will connect to database {}'.format( + config['heatpump']['database'])) + conn = sqlite3.connect(config['heatpump']['database']) + conn.row_factory = sqlite3.Row + return conn + +def initialize(filename): + p = Popen( + '/usr/bin/env sqlite3 -init {} {}'.format(filename, config['heatpump']['database']), + shell=True + ) + return True if p.wait() == 0 else False -conn = sqlite3.connect(config['heatpump']['database']) def sql(query): global conn - return conn.execute(query) + if conn is None: + connect() + + cursor = conn.cursor() + + print(f'Will execute query : \n{query}\n') + cursor.execute(query) + + return cursor + +class RowClass(object): + def __init__(self, **kwargs): + for key in kwargs.keys(): + if hasattr(self, key): + setattr(self, key, kwargs[key]) + + def select(self, key, tablename): + attr = getattr(self, key) + if type(attr) == str: + q = f"SELECT * FROM {tablename} WHERE {key} LIKE '{attr}'" + elif type(attr) == int: + q = f"SELECT * FROM {tablename} WHERE {key} = {attr}" + + return sql(q) diff --git a/pyheatpump/models/__init__.py b/pyheatpump/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyheatpump/models/variable.py b/pyheatpump/models/variable.py new file mode 100644 index 0000000..1407bbe --- /dev/null +++ b/pyheatpump/models/variable.py @@ -0,0 +1,18 @@ +from pyheatpump.db import RowClass +from pyheatpump.db import sql + +class VariableType(RowClass): + slabel: str = None + label: str = None + type: str = None + start_address: int = None + end_address: int = None + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @staticmethod + def getall(): + return dict([ + (row['label'], VariableType(**dict(row))) + for row in sql('SELECT * FROM var_type') ]) diff --git a/pyheatpump/models/variable_type.py b/pyheatpump/models/variable_type.py new file mode 100644 index 0000000..e08e7e0 --- /dev/null +++ b/pyheatpump/models/variable_type.py @@ -0,0 +1,42 @@ +from pyheatpump.db import RowClass +from pyheatpump.db import sql + +class VariableType(RowClass): + slabel: str = None + label: str = None + type: str = None + start_address: int = None + end_address: int = None + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + if self.slabel is None and self.label is not None: + self.slabel = self.label[0] + + + def select(): + try: + elt = next(super().select('slabel', 'var_type')) + except StopIteration: + print('No element exists') + + def save(self): + q = ['UPDATE var_type SET'] + updates = [] + if self.start_address is not None: + updates.append(f'start_address = {self.start_address}') + if self.end_address is not None: + updates.append(f'end_address = {self.end_address}') + if len(updates) == 0: + return + q.append(','.join(updates)) + q.append(f"WHERE slabel LIKE '{self.slabel}'") + + return sql(' '.join(q)) + + @staticmethod + def getall(): + return dict([ + (row['label'], VariableType(**dict(row))) + for row in sql('SELECT * FROM var_type') ]) diff --git a/pyheatpump/models/variable_value.py b/pyheatpump/models/variable_value.py new file mode 100644 index 0000000..1407bbe --- /dev/null +++ b/pyheatpump/models/variable_value.py @@ -0,0 +1,18 @@ +from pyheatpump.db import RowClass +from pyheatpump.db import sql + +class VariableType(RowClass): + slabel: str = None + label: str = None + type: str = None + start_address: int = None + end_address: int = None + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @staticmethod + def getall(): + return dict([ + (row['label'], VariableType(**dict(row))) + for row in sql('SELECT * FROM var_type') ]) diff --git a/pyheatpump/variable_types.py b/pyheatpump/variable_types.py new file mode 100644 index 0000000..36a2ce3 --- /dev/null +++ b/pyheatpump/variable_types.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +import os +from datetime import datetime +from configparser import ConfigParser +from starlette.routing import Route, Router +from starlette.responses import PlainTextResponse, JSONResponse +from pprint import pprint +import uvicorn +import json + +# pyHeatpump modules +from pyheatpump.db import sql +from pyheatpump.models.variable_type import VariableType + +def variable_types(): + assert type(VariableType.getall()) == list + +async def get_variable_types(request): + return JSONResponse(dict([ + (key, val.__dict__) + for key, val in VariableType.getall().items() + ])) + +async def set_variable_types(request): + + body = json.loads(await request.json()) + + for var_type_label, var_type_values in body.items(): + vt = VariableType(label=var_type_label) + for key, val in var_type_values.items(): + if key in [ 'start_address', 'end_address' ]: + setattr(vt, key, val) + + vt.save() + + return PlainTextResponse('OK') + +async def variable_types_routes(request, *args, **kwargs): + if request['method'] == 'GET': + return await get_variable_types(request) + elif request['method'] == 'POST': + return await set_variable_types(request) + + +ROUTES=[ + Route('/', variable_types_routes, methods=['GET', 'POST']) +] + +app = Router(routes=ROUTES) + +if __name__ == '__main__': + uvicorn.run('pyHeatpump:conf.app', + host='127.0.0.1', + port=8000) diff --git a/tests/test_variable_types.py b/tests/test_variable_types.py new file mode 100644 index 0000000..6aec239 --- /dev/null +++ b/tests/test_variable_types.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +import pytest +from starlette.authentication import UnauthenticatedUser +from starlette.testclient import TestClient +#from pyheatpump.conf import app, config, default_config, CONFIG_FILES, get_config, set_config, config_route, ROUTES +from unittest.mock import patch, MagicMock +from pprint import pprint +import json +from tempfile import mkstemp +from configparser import ConfigParser +import os +import sys + +from pyheatpump.conf import config +from pyheatpump.db import initialize, connect +from pyheatpump.variable_types import app, get_variable_types, set_variable_types, ROUTES + +@pytest.fixture(scope='module') +def set_test_db(): + _, tmpdb = mkstemp(suffix='.db', dir=os.getcwd(), ) + print(f'Will store database in {tmpdb}') + config['heatpump']['database'] = tmpdb + if not initialize(os.path.join(os.getcwd(), 'db/pyheatpump.sql')): + sys.exit(-1) + + yield + + os.unlink(tmpdb) + + +def test_get_(set_test_db): + c = TestClient(app) + r = c.get('/') + assert r.status_code == 200 + +class RequestMock(MagicMock): + def __get__(self, key): + if key == 'method': + return 'GET' + +@pytest.mark.asyncio +async def test_get_variable_types(set_test_db): + resp = await get_variable_types(RequestMock()) + assert resp.status_code == 200 + d_resp = json.loads(resp.body.decode()) + assert 'Analog' in d_resp.keys() + assert type(d_resp['Analog']) == dict + assert 'Integer' in d_resp.keys() + assert type(d_resp['Integer']) == dict + assert 'Digital' in d_resp.keys() + assert type(d_resp['Digital']) == dict + +def test_set_variable_types(set_test_db): + c = TestClient(app) + r = c.post('/', json=json.dumps({ + 'Analog': { + 'start_address': 42, + 'end_address': 420, + } + })) + + assert r.status_code == 200 + + r = c.get('/') + d_resp = json.loads(r.content.decode()) + assert d_resp['Analog']['start_address'] == 42 + assert d_resp['Analog']['end_address'] == 420