SSR/dbhandler.py at main · SupernovaRocketry/SSR · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import sqlite3
from threading import Lock

class DBHandler():
    """
    Classe para a manipulação do banco de dados
    """

    def __init__(self,tablename):
        """
        Construtor
        :param tablename: Nome da tabela (Nome da missão)
        """
        self._dbpath = 'db\\scada.db'
        self._tablename = tablename
        self._col_names = ["Altitude",
                        "Latitude",
                        "Longitude",
                        "Principal_Paraquedas_Estabilizador",
                        "Redundancia_Paraquedas_Estabilizador",
                        "Comercial_Paraquedas_Estabilizador",
                        "Principal_Paraquedas_Principal",
                        "Comercial_Paraquedas_Principal",
                        "Acelerometro_X",
                        "Acelerometro_Y",
                        "Acelerometro_Z",
                        "Giroscopio_Roll",
                        "Giroscopio_Pitch",
                        "Giroscopio_Yaw",
                        "RSSI"
        ]


    def conect(self):
        self._con = sqlite3.connect(self._dbpath, check_same_thread = False)
        self._cursor = self._con.cursor()
        self._lock = Lock()
        self._createTable()

    def __del__(self):
        try:
            self._con.close()
        except:
            pass

    def _createTable(self):
        """
        Método que cria a tabela para armazenamento dos dados caso ela não exista
        """
        try:
            sql_str = f"""
            CREATE TABLE IF NOT EXISTS {self._tablename} (
                id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                """
            for n in self._col_names:
                sql_str += f'{n} REAL,'

            sql_str = sql_str[:-1]
            sql_str += ');'
            self._lock.acquire()
            self._cursor.execute(sql_str)
            self._con.commit()
        except Exception as e:
            print("Erro CreateTable: ",e.args)
        finally:
            self._lock.release()

    def insertData(self, data):
        """
        Método para a inserção de dados no banco de dados
        :param data: json ou dicionario contendo todos os dados para a inserção
        """
        try:
            self._lock.acquire()
            self._data = self._handleData(data)
            str_cols = ','.join(self._data.keys())
            str_values = '"' + '", "'.join(str(self._data[k]) for k in self._data.keys()) + '"'
            sql_str = f'INSERT INTO {self._tablename} ({str_cols}) VALUES ({str_values});'
            self._cursor.execute(sql_str)
            self._con.commit()
        except Exception as e:
            print("Erro insertData: ",e.args())
            raise e
        finally:
            self._lock.release()


    def _handleData(self, data):
        """
        Método
        """
        try:
            newData = {"timestamp" : data['timestamp'],
                        "Altitude" : data['Altitude'],
                        "Latitude" : data['Latitude'],
                        "Longitude" : data['Longitude'],
                        "Principal_Paraquedas_Estabilizador" : data['Principal Paraquedas Estabilizador'],
                        "Redundancia_Paraquedas_Estabilizador" : data['Redundancia Paraquedas Estabilizador'],
                        "Comercial_Paraquedas_Estabilizador" : data['Comercial Paraquedas Estabilizador'],
                        "Principal_Paraquedas_Principal" : data['Principal Paraquedas Principal'],
                        "Comercial_Paraquedas_Principal" : data['Comercial Paraquedas Principal'],
                        "Acelerometro_X" : data['Acelerometro']['x'],
                        "Acelerometro_Y" : data['Acelerometro']['y'],
                        "Acelerometro_Z" : data['Acelerometro']['z'],
                        "Giroscopio_Roll" : data['Giroscopio']['x'],
                        "Giroscopio_Pitch" : data['Giroscopio']['y'],
                        "Giroscopio_Yaw" : data['Giroscopio']['z'],
                        "RSSI" : data['RSSI']}

            return newData
        except Exception as e:
            print("Falha ao tratar dados: ",e.args())


    def disconect(self):
        self._con.close()

    # def selectData(self, cols):
    #     """
    #     Método que realiza a busca no BD para utilizar nos relatorios automatizados
    #     """
    #     try:
    #         self._lock.acquire()

    #         self._cursor.execute(sql_str)
    #         self._con.commit()
    #         self._lock.release()
    #     except Exception as e:
    #         print("Erro: ",e.args())