HistorySQLite.read_node_history() - Code Metrics - Inspection of "Implement Event History" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#172)

unknown

created 2016-05-01 07:55 UTC

HistorySQLite.read_node_history() F

↳ Parent: HistorySQLite

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	33
CRAP Score	13.0976

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	13
c	3
b	0
f	0
dl	0
loc	54
ccs	33
cts	36
cp	0.9167
crap	13.0976
rs	3.5512

How to fix Long Method Complexity

import logging
from datetime import timedelta
from datetime import datetime
from threading import Lock
from opcua import ua
from opcua.common.utils import Buffer
from opcua.server.history import HistoryStorageInterface
import sqlite3


class HistorySQLite(HistoryStorageInterface):
    """
    very minimal history backend storing data in SQLite database
    """

    def __init__(self, path="history.db"):
        self.logger = logging.getLogger(__name__)
        self._datachanges_period = {}
        self._events = {}
        self._db_file = path
        self._lock = Lock()
        self._event_attributes = {}

        self._conn = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES, check_same_thread=False)

    def new_historized_node(self, node_id, period, count=0):
        with self._lock:
            _c_new = self._conn.cursor()

            table = self._get_table_name(node_id)

            self._datachanges_period[node_id] = period, count

            # create a table for the node which will store attributes of the DataValue object
            # note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
            try:
                _c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
                               ' ServerTimestamp TIMESTAMP,'
                               ' SourceTimestamp TIMESTAMP,'
                               ' StatusCode INTEGER,'
                               ' Value TEXT,'
                               ' VariantType TEXT,'
                               ' VariantBinary BLOB)'.format(tn=table))

            except sqlite3.Error as e:
                self.logger.info('Historizing SQL Table Creation Error for %s: %s', node_id, e)

            self._conn.commit()

    def save_node_value(self, node_id, datavalue):
        with self._lock:
            _c_sub = self._conn.cursor()

            table = self._get_table_name(node_id)

            # insert the data change into the database
            try:
                _c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, ?, ?, ?, ?, ?, ?)'.format(tn=table),
                               (
                                   datavalue.ServerTimestamp,
                                   datavalue.SourceTimestamp,
                                   datavalue.StatusCode.value,
                                   str(datavalue.Value.Value),
                                   datavalue.Value.VariantType.name,
                                   datavalue.Value.to_binary()
                               )
                               )
            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Insert Error for %s: %s', node_id, e)

            self._conn.commit()

            # get this node's period from the period dict and calculate the limit
            period, count = self._datachanges_period[node_id]

            if period:
                # after the insert, if a period was specified delete all records older than period
                date_limit = datetime.now() - period

                try:
                    _c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=table),
                                   (date_limit.isoformat(' '),))
                except sqlite3.Error as e:
                    self.logger.error('Historizing SQL Delete Old Data Error for %s: %s', node_id, e)

                self._conn.commit()

    def read_node_history(self, node_id, start, end, nb_values):
        with self._lock:
            _c_read = self._conn.cursor()

            order = "ASC"

            if start is None or start == ua.DateTimeMinValue:
                order = "DESC"
                start = ua.DateTimeMinValue

            if end is None or end == ua.DateTimeMinValue:
                end = datetime.utcnow() + timedelta(days=1)

            if start < end:
                start_time = start.isoformat(' ')
                end_time = end.isoformat(' ')
            else:
                order = "DESC"
                start_time = end.isoformat(' ')
                end_time = start.isoformat(' ')

            if nb_values:
                limit = nb_values + 1  # add 1 to the number of values for retrieving a continuation point
            else:
                limit = -1  # in SQLite a LIMIT of -1 returns all results

            table = self._get_table_name(node_id)

            cont = None
            results = []

            # select values from the database; recreate UA Variant from binary
            try:
                for row in _c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
                                           'ORDER BY "Id" {dir} LIMIT ?'.format(tn=table, dir=order), (start_time, end_time, limit,)):

                    # rebuild the data value object
                    dv = ua.DataValue(ua.Variant.from_binary(Buffer(row[6])))
                    dv.ServerTimestamp = row[1]
                    dv.SourceTimestamp = row[2]
                    dv.StatusCode = ua.StatusCode(row[3])

                    results.append(dv)

            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Read Error for %s: %s', node_id, e)

            if nb_values:
                if start > ua.DateTimeMinValue and len(results) > nb_values:
                    cont = results[nb_values].ServerTimestamp

                results = results[:nb_values]

            return results, cont

    def new_historized_event(self, source_id, etype, period):
        with self._lock:
            _c_new = self._conn.cursor()

            table = self._get_table_name(source_id)

            # FIXME need a way to add new column for new fields to a table, or build the table from multiple event types

            self._datachanges_period[source_id] = period

            self._event_attributes[source_id] = etype.__dict__.keys()  # FIXME: find a better way? use vars()?

            fields = self._get_event_fields(etype)

            # create a table for the event which will store attributes of the Event object
            try:
                _c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL, '
                               'Timestamp TIMESTAMP, '
                               '{fd})'.format(tn=table, fd=fields))

            except sqlite3.Error as e:
                self.logger.info('Historizing SQL Table Creation Error for events from %s: %s', source_id, e)

            self._conn.commit()

    def save_event(self, event):
        with self._lock:
            _c_sub = self._conn.cursor()

            table = self._get_table_name(event.SourceNode)

            fields, placeholders, evtup = self._format_event(event)

            # insert the event into the database
            try:
                _c_sub.execute('INSERT INTO "{tn}" ("Id", "Timestamp", {fd}) VALUES (NULL, "{ts}", '
                               '{pl})'.format(tn=table, fd=fields, ts=event.Time, pl=placeholders), evtup)
            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Insert Error for events from %s: %s', event.SourceNode, e)

            self._conn.commit()

            # get this node's period from the period dict and calculate the limit
            period = self._datachanges_period[event.SourceNode]

            if period:
                # after the insert, if a period was specified delete all records older than period
                date_limit = datetime.now() - period

                try:
                    _c_sub.execute('DELETE FROM "{tn}" WHERE Time < ?'.format(tn=table),
                                   (date_limit.isoformat(' '),))
                except sqlite3.Error as e:
                    self.logger.error('Historizing SQL Delete Old Data Error for events from %s: %s', event.SourceNode, e)

                self._conn.commit()

    def read_event_history(self, source_id, start, end, nb_values, evfilter):
        with self._lock:

            _c_read = self._conn.cursor()

            order = "ASC"

            if start is None or start == ua.DateTimeMinValue:
                order = "DESC"
                start = ua.DateTimeMinValue

            if end is None or end == ua.DateTimeMinValue:
                end = datetime.utcnow() + timedelta(days=1)

            if start < end:
                start_time = start.isoformat(' ')
                end_time = end.isoformat(' ')
            else:
                order = "DESC"
                start_time = end.isoformat(' ')
                end_time = start.isoformat(' ')

            if nb_values:
                limit = nb_values + 1  # add 1 to the number of values for retrieving a continuation point
            else:
                limit = -1  # in SQLite a LIMIT of -1 returns all results

            table = self._get_table_name(source_id)
            clauses = self._get_select_clauses(source_id, evfilter)

            cont = None
            results = []

            # select events from the database; SQL select clause is built from EventFilter and available fields
            try:
                for row in _c_read.execute('SELECT {cl} FROM "{tn}" WHERE "Timestamp" BETWEEN ? AND ? '
                                           'ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order),
                                           (start_time, end_time, limit,)):

                    # place all the variants in the event field list object
                    hist_ev_field_list = ua.HistoryEventFieldList()
                    for field in row:
                        hist_ev_field_list.EventFields.append(ua.Variant.from_binary(Buffer(field)))

                    results.append(hist_ev_field_list)

            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Read Error events for node %s: %s', source_id, e)

            if nb_values:
                if start > ua.DateTimeMinValue and len(results) > nb_values:
                    cont = results[nb_values].Time

                results = results[:nb_values]

            return results, cont

    def _get_table_name(self, node_id):
        return str(node_id.NamespaceIndex) + '_' + str(node_id.Identifier)

    def _format_event(self, event_result):
        placeholders = []
        ev_fields = []
        ev_variant_binaries = []

        ev_variant_dict = event_result.get_field_variants()

        # split dict into two synchronized lists which will be converted to SQL strings
        # note that the variants are converted to binary objects for storing in SQL BLOB format
        for field, variant in ev_variant_dict.items():
            placeholders.append('?')
            ev_fields.append(field)
            ev_variant_binaries.append(variant.to_binary())

        return self._list_to_sql_str(ev_fields), self._list_to_sql_str(placeholders, False), tuple(ev_variant_binaries)

    def _get_event_fields(self, etype):
        fields = []
        for key, value in vars(etype).items():
            if not key.startswith("__") and key is not "_freeze":
                fields.append(key + ' BLOB')
        return self._list_to_sql_str(fields, False)

    def _get_select_clauses(self, source_id, evfilter):
        s_clauses = []
        for select_clause in evfilter.SelectClauses:
            try:
                if not select_clause.BrowsePath:
                    s_clauses.append(select_clause.Attribute.name)
                else:
                    name = select_clause.BrowsePath[0].Name
                    s_clauses.append(name)
            except AttributeError:
                pass
                # FIXME what to do here?

        # remove select clauses that the event type doesn't have; SQL will error because the column doesn't exist
        clauses = [x for x in s_clauses if self._check(source_id, x)]

        return self._list_to_sql_str(clauses)

    def _check(self, source_id, s_clause):
        if s_clause in self._event_attributes[source_id]:
            return True
        else:
            return False

    def _list_to_sql_str(self, ls, quotes=True):
        sql_str = ''
        for item in ls:
            if quotes:
                sql_str += '"' + item + '", '
            else:
                sql_str += item + ', '
        return sql_str[:-2]  # remove trailing space and comma for SQL syntax

    def stop(self):
        with self._lock:
            self._conn.close()
            self.logger.info('Historizing SQL connection closed')


1	1	import logging
2	1	from datetime import timedelta
3	1	from datetime import datetime
4	1	from threading import Lock
5	1	from opcua import ua
6	1	from opcua.common.utils import Buffer
7	1	from opcua.server.history import HistoryStorageInterface
8	1	import sqlite3
9
10
11	1	class HistorySQLite(HistoryStorageInterface):
12		"""
13		very minimal history backend storing data in SQLite database
14		"""
15
16	1	def __init__(self, path="history.db"):
17	1	self.logger = logging.getLogger(__name__)
18	1	self._datachanges_period = {}
19	1	self._events = {}
20	1	self._db_file = path
21	1	self._lock = Lock()
22		self._event_attributes = {}
23	1
24		self._conn = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES, check_same_thread=False)
25	1
26	1	def new_historized_node(self, node_id, period, count=0):
27	1	with self._lock:
28		_c_new = self._conn.cursor()
29	1
30		table = self._get_table_name(node_id)
31	1
32		self._datachanges_period[node_id] = period, count
33
34		# create a table for the node which will store attributes of the DataValue object
35	1	# note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
36	1	try:
37		_c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
38		' ServerTimestamp TIMESTAMP,'
39		' SourceTimestamp TIMESTAMP,'
40		' StatusCode INTEGER,'
41		' Value TEXT,'
42		' VariantType TEXT,'
43		' VariantBinary BLOB)'.format(tn=table))
44
45		except sqlite3.Error as e:
46		self.logger.info('Historizing SQL Table Creation Error for %s: %s', node_id, e)
47	1
48		self._conn.commit()
49	1
50	1	def save_node_value(self, node_id, datavalue):
51	1	with self._lock:
52		_c_sub = self._conn.cursor()
53	1
54		table = self._get_table_name(node_id)
55
56	1	# insert the data change into the database
57	1	try:
58		_c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, ?, ?, ?, ?, ?, ?)'.format(tn=table),
59		(
60		datavalue.ServerTimestamp,
61		datavalue.SourceTimestamp,
62		datavalue.StatusCode.value,
63		str(datavalue.Value.Value),
64		datavalue.Value.VariantType.name,
65		datavalue.Value.to_binary()
66		)
67		)
68		except sqlite3.Error as e:
69		self.logger.error('Historizing SQL Insert Error for %s: %s', node_id, e)
70	1
71		self._conn.commit()
72
73	1	# get this node's period from the period dict and calculate the limit
74		period, count = self._datachanges_period[node_id]
75	1
76		if period:
77		# after the insert, if a period was specified delete all records older than period
78		date_limit = datetime.now() - period
79
80		try:
81		_c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=table),
82		(date_limit.isoformat(' '),))
83		except sqlite3.Error as e:
84		self.logger.error('Historizing SQL Delete Old Data Error for %s: %s', node_id, e)
85
86		self._conn.commit()
87	1
88	1	def read_node_history(self, node_id, start, end, nb_values):
89	1	with self._lock:
90		_c_read = self._conn.cursor()
91	1
92		order = "ASC"
93	1
94	1	if start is None or start == ua.DateTimeMinValue:
95	1	order = "DESC"
96		start = ua.DateTimeMinValue
97	1
98	1	if end is None or end == ua.DateTimeMinValue:
99		end = datetime.utcnow() + timedelta(days=1)
100	1
101	1	if start < end:
102	1	start_time = start.isoformat(' ')
103		end_time = end.isoformat(' ')
104	1	else:
105	1	order = "DESC"
106	1	start_time = end.isoformat(' ')
107		end_time = start.isoformat(' ')
108	1
109	1	if nb_values:
110		limit = nb_values + 1 # add 1 to the number of values for retrieving a continuation point
111	1	else:
112		limit = -1 # in SQLite a LIMIT of -1 returns all results
113	1
114		table = self._get_table_name(node_id)
115	1
116	1	cont = None
117		results = []
118
119	1	# select values from the database; recreate UA Variant from binary
120	1	try:
121		for row in _c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
122	1	'ORDER BY "Id" {dir} LIMIT ?'.format(tn=table, dir=order), (start_time, end_time, limit,)):
123	1
124	1	# rebuild the data value object
125	1	dv = ua.DataValue(ua.Variant.from_binary(Buffer(row[6])))
126		dv.ServerTimestamp = row[1]
127	1	dv.SourceTimestamp = row[2]
128		dv.StatusCode = ua.StatusCode(row[3])
129
130		results.append(dv)
131
132	1	except sqlite3.Error as e:
133	1	self.logger.error('Historizing SQL Read Error for %s: %s', node_id, e)
134	1
135		if nb_values:
136	1	if start > ua.DateTimeMinValue and len(results) > nb_values:
137		cont = results[nb_values].ServerTimestamp
138	1
139		results = results[:nb_values]
140	1
141		return results, cont
142
143	1	def new_historized_event(self, source_id, etype, period):
144		with self._lock:
145		_c_new = self._conn.cursor()
146	1
147		table = self._get_table_name(source_id)
148
149	1	# FIXME need a way to add new column for new fields to a table, or build the table from multiple event types
150	1
151		self._datachanges_period[source_id] = period
152	1
153	1	self._event_attributes[source_id] = etype.__dict__.keys() # FIXME: find a better way? use vars()?
154	1
155		fields = self._get_event_fields(etype)
156
157		# create a table for the event which will store attributes of the Event object
158		try:
159		_c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL, '
160		'Timestamp TIMESTAMP, '
161		'{fd})'.format(tn=table, fd=fields))
162
163		except sqlite3.Error as e:
164		self.logger.info('Historizing SQL Table Creation Error for events from %s: %s', source_id, e)
165
166		self._conn.commit()
167
168		def save_event(self, event):
169		with self._lock:
170		_c_sub = self._conn.cursor()
171
172		table = self._get_table_name(event.SourceNode)
173
174		fields, placeholders, evtup = self._format_event(event)
175
176		# insert the event into the database
177		try:
178		_c_sub.execute('INSERT INTO "{tn}" ("Id", "Timestamp", {fd}) VALUES (NULL, "{ts}", '
179		'{pl})'.format(tn=table, fd=fields, ts=event.Time, pl=placeholders), evtup)
180		except sqlite3.Error as e:
181		self.logger.error('Historizing SQL Insert Error for events from %s: %s', event.SourceNode, e)
182
183		self._conn.commit()
184
185		# get this node's period from the period dict and calculate the limit
186		period = self._datachanges_period[event.SourceNode]
187
188		if period:
189		# after the insert, if a period was specified delete all records older than period
190		date_limit = datetime.now() - period
191
192		try:
193		_c_sub.execute('DELETE FROM "{tn}" WHERE Time < ?'.format(tn=table),
194		(date_limit.isoformat(' '),))
195		except sqlite3.Error as e:
196		self.logger.error('Historizing SQL Delete Old Data Error for events from %s: %s', event.SourceNode, e)
197
198		self._conn.commit()
199
200		def read_event_history(self, source_id, start, end, nb_values, evfilter):
201		with self._lock:
202
203		_c_read = self._conn.cursor()
204
205		order = "ASC"
206
207		if start is None or start == ua.DateTimeMinValue:
208		order = "DESC"
209		start = ua.DateTimeMinValue
210
211		if end is None or end == ua.DateTimeMinValue:
212		end = datetime.utcnow() + timedelta(days=1)
213
214		if start < end:
215		start_time = start.isoformat(' ')
216		end_time = end.isoformat(' ')
217		else:
218		order = "DESC"
219		start_time = end.isoformat(' ')
220		end_time = start.isoformat(' ')
221
222		if nb_values:
223		limit = nb_values + 1 # add 1 to the number of values for retrieving a continuation point
224		else:
225		limit = -1 # in SQLite a LIMIT of -1 returns all results
226
227		table = self._get_table_name(source_id)
228		clauses = self._get_select_clauses(source_id, evfilter)
229
230		cont = None
231		results = []
232
233		# select events from the database; SQL select clause is built from EventFilter and available fields
234		try:
235		for row in _c_read.execute('SELECT {cl} FROM "{tn}" WHERE "Timestamp" BETWEEN ? AND ? '
236		'ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order),
237		(start_time, end_time, limit,)):
238
239		# place all the variants in the event field list object
240		hist_ev_field_list = ua.HistoryEventFieldList()
241		for field in row:
242		hist_ev_field_list.EventFields.append(ua.Variant.from_binary(Buffer(field)))
243
244		results.append(hist_ev_field_list)
245
246		except sqlite3.Error as e:
247		self.logger.error('Historizing SQL Read Error events for node %s: %s', source_id, e)
248
249		if nb_values:
250		if start > ua.DateTimeMinValue and len(results) > nb_values:
251		cont = results[nb_values].Time
252
253		results = results[:nb_values]
254
255		return results, cont
256
257		def _get_table_name(self, node_id):
258		return str(node_id.NamespaceIndex) + '_' + str(node_id.Identifier)
259
260		def _format_event(self, event_result):
261		placeholders = []
262		ev_fields = []
263		ev_variant_binaries = []
264
265		ev_variant_dict = event_result.get_field_variants()
266
267		# split dict into two synchronized lists which will be converted to SQL strings
268		# note that the variants are converted to binary objects for storing in SQL BLOB format
269		for field, variant in ev_variant_dict.items():
270		placeholders.append('?')
271		ev_fields.append(field)
272		ev_variant_binaries.append(variant.to_binary())
273
274		return self._list_to_sql_str(ev_fields), self._list_to_sql_str(placeholders, False), tuple(ev_variant_binaries)
275
276		def _get_event_fields(self, etype):
277		fields = []
278		for key, value in vars(etype).items():
279		if not key.startswith("__") and key is not "_freeze":
280		fields.append(key + ' BLOB')
281		return self._list_to_sql_str(fields, False)
282
283		def _get_select_clauses(self, source_id, evfilter):
284		s_clauses = []
285		for select_clause in evfilter.SelectClauses:
286		try:
287		if not select_clause.BrowsePath:
288		s_clauses.append(select_clause.Attribute.name)
289		else:
290		name = select_clause.BrowsePath[0].Name
291		s_clauses.append(name)
292		except AttributeError:
293		pass
294		# FIXME what to do here?
295
296		# remove select clauses that the event type doesn't have; SQL will error because the column doesn't exist
297		clauses = [x for x in s_clauses if self._check(source_id, x)]
298
299		return self._list_to_sql_str(clauses)
300
301		def _check(self, source_id, s_clause):
302		if s_clause in self._event_attributes[source_id]:
303		return True
304		else:
305		return False
306
307		def _list_to_sql_str(self, ls, quotes=True):
308		sql_str = ''
309		for item in ls:
310		if quotes:
311		sql_str += '"' + item + '", '
312		else:
313		sql_str += item + ', '
314		return sql_str[:-2] # remove trailing space and comma for SQL syntax
315
316		def stop(self):
317		with self._lock:
318		self._conn.close()
319		self.logger.info('Historizing SQL connection closed')
320

FreeOpcUa / python-opcua

Pull Request — master (#172)

HistorySQLite.read_node_history() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like