Projet de remplacement du "RPiPasserelle" d'Otec.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102
  1. #!/usr/bin/env python3
  2. import sqlite3
  3. from subprocess import Popen
  4. from .config import config
  5. import sys
  6. from pprint import pprint
  7. from pyheatpump.logger import logger_init
  8. logger = logger_init()
  9. conn = None
  10. class DB():
  11. Conn = None
  12. @staticmethod
  13. def connect():
  14. if DB.Conn:
  15. try:
  16. DB.Conn.cursor()
  17. return True
  18. except sqlite3.ProgrammingError:
  19. pass
  20. except Exception as e:
  21. print(e)
  22. sys.exit(1)
  23. logger.info('Will connect to database {}'.format(
  24. config['heatpump']['database']))
  25. DB.Conn = sqlite3.connect(
  26. config['heatpump']['database'],
  27. isolation_level=None)
  28. DB.Conn.row_factory = sqlite3.Row
  29. return True
  30. def commit():
  31. """
  32. Default mode of connecting is autocommit
  33. """
  34. pass
  35. @staticmethod
  36. def initialize(filename):
  37. if DB.Conn:
  38. DB.Conn.close()
  39. p = Popen(
  40. '/usr/bin/env sqlite3 -init {} {}'.format(filename, config['heatpump']['database']),
  41. shell=True)
  42. return True if p.wait() == 0 else False
  43. @staticmethod
  44. def sql(query, params={}):
  45. logger.debug('Query : %s', query)
  46. logger.debug('Params : %s', params)
  47. if not DB.connect():
  48. raise Exception('Can\'t connect to DB')
  49. return DB.Conn.execute(query, params or {})
  50. @staticmethod
  51. def sqlf(filename):
  52. with open(filename) as f:
  53. if not DB.connect():
  54. raise Exception('Can\'t connect to DB')
  55. try:
  56. return DB.Conn.execute((' '.join(f.readlines())).replace('\n', ''))
  57. except StopIteration:
  58. print('failed')
  59. return []
  60. except sqlite3.IntegrityError as e:
  61. print(e)
  62. return []
  63. except sqlite3.OperationalError as e:
  64. raise Exception('Can\'t find DB file %s'.format(filename)) from e
  65. class RowClass(object):
  66. def __init__(self, **kwargs):
  67. for key in kwargs.keys():
  68. if hasattr(self, key):
  69. setattr(self, key, kwargs[key])
  70. def select(self, key, tablename):
  71. attr = getattr(self, key)
  72. if type(attr) == str:
  73. q = ("SELECT * FROM {tablename} WHERE ? LIKE '?'", (key, attr))
  74. elif type(attr) == int:
  75. q = ("SELECT * FROM {tablename} WHERE ? LIKE ?", (key, attr))
  76. res = []
  77. for row in DB.sql(*q):
  78. res.append(res)
  79. return res