opcua.server.HistoryDict.stop() - Code Metrics - Inspection of "Add SQLite History Support for Values" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#155)

unknown

created 2016-04-08 20:07 UTC

opcua.server.HistoryDict.stop() A

↳ Parent: opcua.server.HistoryDict

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1

Metric	Value
cc	1
dl	0
loc	2
ccs	1
cts	1
cp	1
crap	1
rs	10

from datetime import timedelta
from datetime import datetime

from opcua import Subscription
from opcua import ua

import sqlite3


class HistoryStorageInterface(object):

    """
    Interface of a history backend.
    Must be implemented by backends
    """

    def new_historized_node(self, node, period, count=0):
        """
        Called when a new node is to be historized
        Returns None
        """
        raise NotImplementedError

    def save_node_value(self, node, datavalue):
        """
        Called when the value of a historized node has changed and should be saved in history
        Returns None
        """
        raise NotImplementedError

    def read_node_history(self, node, start, end, nb_values):
        """
        Called when a client make a history read request for a node
        if start or end is missing then nb_values is used to limit query
        nb_values is the max number of values to read. Ignored if 0
        Start time and end time are inclusive
        Returns a list of DataValues and a continuation point which
        is None if all nodes are read or the ServerTimeStamp of the last rejected DataValue
        """
        raise NotImplementedError

    def new_historized_event(self, event, period):
        """
        Called when historization of events is enabled on server side
        FIXME: we may need to store events per nodes in future...
        Returns None
        """
        raise NotImplementedError

    def save_event(self, event):
        """
        Called when a new event has been generated ans should be saved in history
        Returns None
        """
        raise NotImplementedError

    def read_event_history(self, start, end, evfilter):
        """
        Called when a client make a history read request for events
        Start time and end time are inclusive
        Returns a list of Events and a continuation point which
        is None if all events are read or the ServerTimeStamp of the last rejected event
        """
        raise NotImplementedError

    def stop(self):
        """
        Called when the server shuts down
        Can be used to close database connections etc.
        """
        raise NotImplementedError


class HistoryDict(HistoryStorageInterface):
    """
    very minimal history backend storing data in memory using a Python dictionnary
    """
    def __init__(self):
        self._datachanges = {}
        self._datachanges_period = {}
        self._events = {}

    def new_historized_node(self, node, period, count=0):
        node_id = node.nodeid
        self._datachanges[node_id] = []
        self._datachanges_period[node_id] = period, count

    def save_node_value(self, node, datavalue):
        node_id = node.nodeid
        data = self._datachanges[node_id]
        period, count = self._datachanges_period[node_id]
        data.append(datavalue)
        now = datetime.now()
        if period:
            while now - data[0].ServerTimestamp > period:
                data.pop(0)
        if count and len(data) > count:
            data = data[-count:]

    def read_node_history(self, node, start, end, nb_values):
        node_id = node.NodeId
        cont = None
        if node_id not in self._datachanges:
            print("Error attempt to read history for a node which is not historized")
            return [], cont
        else:
            if end is None:
                end = datetime.now() + timedelta(days=1)
            if start is None:
                start = ua.DateTimeMinValue
            results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp <= end]
            if nb_values:
                if start > ua.DateTimeMinValue and len(results) > nb_values:
                    cont = results[nb_values + 1].ServerTimestamp
                    results = results[:nb_values]
                else:
                    results = results[-nb_values:]
            return results, cont

    def new_historized_event(self, event, period):
        self._events = []

    def save_event(self, event):
        raise NotImplementedError

    def read_event_history(self, start, end, evfilter):
        raise NotImplementedError

    def stop(self):
        pass


class HistorySQLite(HistoryStorageInterface):
    """
    very minimal history backend storing data in SQLite database
    """
    # FIXME: need to check on if sql_conn.commit() should be inside try block; and if .rollback() needs to be except

    def __init__(self):
        self._datachanges_period = {}
        self._events = {}
        self._db_file = "history.db"

        # SQL objects must be accessed in a single thread
        # adding a new node to be historized probably happens on the main thread
        self._conn_new = None
        self._c_new = None

        # subscriptions are in another thread so it needs it's own sqlite connection object
        self._conn_sub = None
        self._c_sub = None

        # FIXME: no idea what thread the read values happen in, just make a new conn object for now
        self._conn_read = None
        self._c_read = None

    def new_historized_node(self, node, period, count=0):
        if self._conn_new is None:
            self._conn_new = sqlite3.connect(self._db_file)
            self._c_new = self._conn_new.cursor()

        node_id = str(node.nodeid.NamespaceIndex) + '_' + str(node.nodeid.Identifier)
        self._datachanges_period[node_id] = period

        self._conn_new = sqlite3.connect(self._db_file)
        self._c_new = self._conn_new.cursor()

        sql_type = self._get_sql_type(node)

        # create a table for the node which will store attributes of the DataValue object
        try:
            self._c_new.execute('CREATE TABLE "{tn}" (ServerTimestamp TIMESTAMP,'
                               ' SourceTimestamp TIMESTAMP,'
                               ' StatusCode INTEGER,'
                               ' Value {type},'
                               ' VariantType INTEGER)'.format(tn=node_id, type=sql_type))

        except sqlite3.Error as e:
            print(node_id, 'Historizing SQL Table Creation Error:', e)

        self._conn_new.commit()
        self._conn_new.close()

    def save_node_value(self, node, datavalue):
        if self._conn_sub is None:
            self._conn_sub = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES)
            self._c_sub = self._conn_sub.cursor()

        node_id = str(node.nodeid.NamespaceIndex) + '_' + str(node.nodeid.Identifier)

        # insert the data change into the database
        try:
            self._c_sub.execute('INSERT INTO "{tn}" VALUES (?, ?, ?, ?, ?)'.format(tn=node_id), (datavalue.ServerTimestamp,
                                                                                                 datavalue.SourceTimestamp,
                                                                                                 datavalue.StatusCode.value,
                                                                                                 datavalue.Value.Value,
                                                                                                 datavalue.Value.VariantType.value))
        except sqlite3.Error as e:
            print(node_id, 'Historizing SQL Insert Error:', e)

        # get this node's period from the period dict and calculate the limit
        period = self._datachanges_period[node_id]
        date_limit = datetime.now() - period

        # after the insert, delete all values older than period
        try:
            self._c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=node_id), (date_limit.isoformat(' '),))
        except sqlite3.Error as e:
            print(node_id, 'Historizing SQL Delete Old Data Error:', e)

        self._conn_sub.commit()

    def read_node_history(self, node, start, end, nb_values):
        if self._conn_read is None:
            self._conn_read = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES)
            self._c_read = self._conn_read.cursor()

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

        node_id = str(node.NodeId.NamespaceIndex) + '_' + str(node.NodeId.Identifier)
        cont = None
        results = []

        start_time = start.isoformat(' ')
        end_time = end.isoformat(' ')

        # select values from the database
        try:
            for row in self._c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
                                           'LIMIT ?'.format(tn=node_id), (start_time, end_time, nb_values,)):

                dv = ua.DataValue(ua.Variant(row[3], self._get_variant_type(row[4])))
                dv.ServerTimestamp = row[0]
                dv.SourceTimestamp = row[1]
                dv.StatusCode = ua.StatusCode(row[2])

                results.append(dv)

        except sqlite3.Error as e:
            print(node_id, 'Historizing SQL Read Error:', e)

        return results, cont

    def new_historized_event(self, event, period):
        raise NotImplementedError

    def save_event(self, event):
        raise NotImplementedError

    def read_event_history(self, start, end, evfilter):
        raise NotImplementedError

    # convert the node UA Variant type to an SQL supported type
    # FIXME: this could lead to lost precision! Better way to store the data? Add custom types to SQL?
    def _get_sql_type(self, node):
        node_type = node.get_data_type()
        # see object_ids.py
        if node_type.Identifier in (10, 11,):  # Float, Double
            return 'REAL'
        elif node_type.Identifier in (4, 5, 6, 7, 8, 9,):  # Ints
            return "INT"
        elif node_type.Identifier in (12,):  # String
            return "TEXT"
        else:
            return "NULL"

    # convert the OPC UA variant identifier stored in SQL back to a UA Variant
    # FIXME: is there a util method someplace for getting this?
    def _get_variant_type(self, identifier):
        if identifier is 10:
            return ua.VariantType.Float
        elif identifier is 11:
            return ua.VariantType.Double
        elif identifier is 4:
            return ua.VariantType.Int16
        elif identifier is 5:
            return ua.VariantType.UInt16
        elif identifier is 6:
            return ua.VariantType.Int32
        elif identifier is 7:
            return ua.VariantType.UInt32
        elif identifier is 8:
            return ua.VariantType.Int64
        elif identifier is 9:
            return ua.VariantType.UInt64
        elif identifier is 12:
            return ua.VariantType.String

    # close connections to the history database when the server stops
    def stop(self):
        pass
        # FIXME: Should close the database connections when the server stops, but because server.stop() is called
        # FIXME: on a different thread than the SQL conn object, no idea how to do this at the moment


class SubHandler(object):
    def __init__(self, storage):
        self.storage = storage

    def datachange_notification(self, node, val, data):
        self.storage.save_node_value(node, data.monitored_item.Value)  # SHOULD NOT GET NODE ID HERE

    def event_notification(self, event):
        self.storage.save_event(event)


class HistoryManager(object):
    def __init__(self, iserver):
        self.iserver = iserver
        self.storage = HistorySQLite()  # HistoryDict() HistorySQLite()
        self._sub = None
        self._handlers = {}

    def set_storage(self, storage):
        self.storage = storage

    def _create_subscription(self, handler):
        params = ua.CreateSubscriptionParameters()
        params.RequestedPublishingInterval = 10
        params.RequestedLifetimeCount = 3000
        params.RequestedMaxKeepAliveCount = 10000
        params.MaxNotificationsPerPublish = 0
        params.PublishingEnabled = True
        params.Priority = 0
        return Subscription(self.iserver.isession, params, handler)

    def historize(self, node, period=timedelta(days=7), count=0):
        if not self._sub:
            self._sub = self._create_subscription(SubHandler(self.storage))
        if node in self._handlers:
            raise ua.UaError("Node {} is allready historized".format(node))
        self.storage.new_historized_node(node, period, count)  # SHOULD NOT GET NODE ID HERE
        handler = self._sub.subscribe_data_change(node)
        self._handlers[node] = handler

    def dehistorize(self, node):
        self._sub.unsubscribe(self._handlers[node])
        del(self._handlers[node])

    def read_history(self, params):
        """
        Read history for a node
        This is the part AttributeService, but implemented as its own service
        since it requires more logic than other attribute service methods
        """
        results = []
        
        for rv in params.NodesToRead:
            res = self._read_history(params.HistoryReadDetails, rv)
            results.append(res)
        return results
        
    def _read_history(self, details, rv):
        """
        read history for a node
        """
        result = ua.HistoryReadResult()
        if isinstance(details, ua.ReadRawModifiedDetails):
            if details.IsReadModified:
                result.HistoryData = ua.HistoryModifiedData()
                # we do not support modified history by design so we return what we have
            else:
                result.HistoryData = ua.HistoryData()
            dv, cont = self._read_datavalue_history(rv, details)
            result.HistoryData.DataValues = dv
            result.ContinuationPoint = cont

        elif isinstance(details, ua.ReadEventDetails):
            result.HistoryData = ua.HistoryEvent()
            # FIXME: filter is a cumbersome type, maybe transform it something easier
            # to handle for storage
            result.HistoryData.Events = self.storage.read_event_history(details.StartTime,
                                                                        details.EndTime,
                                                                        details.Filter)
        else:
            # we do not currently support the other types, clients can process data themselves
            result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNotImplemented)
        return result

    def _read_datavalue_history(self, rv, details):
        starttime = details.StartTime
        if rv.ContinuationPoint:
            # Spec says we should ignore details if cont point is present
            # but they also say we can use cont point as timestamp to enable stateless
            # implementation. This is contradictory, so we assume details is
            # send correctly with continuation point
            # starttime = bytes_to_datetime(rv.ContinuationPoint)
            starttime = ua.unpack_datetime(rv.ContinuationPoint)

        dv, cont = self.storage.read_node_history(rv,  # SHOULD NOT GET NODE ID HERE!
                                                  starttime,
                                                  details.EndTime,
                                                  details.NumValuesPerNode)
        if cont:
            # cont = datetime_to_bytes(dv[-1].ServerTimestamp)
            cont = ua.pack_datetime(dv[-1].ServerTimestamp)
        # FIXME, parse index range and filter out if necessary
        # rv.IndexRange
        # rv.DataEncoding # xml or binary, seems spec say we can ignore that one
        return dv, cont

    def update_history(self, params):
        """
        Update history for a node
        This is the part AttributeService, but implemented as its own service
        since it requires more logic than other attribute service methods
        """
        results = []
        for _ in params.HistoryUpdateDetails:
            result = ua.HistoryUpdateResult()
            # we do not accept to rewrite history
            result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNotWritable)
            results.append(results)
        return results

    def stop(self):
        self.storage.stop()


1	1	from datetime import timedelta
2	1	from datetime import datetime
3
4	1	from opcua import Subscription
5	1	from opcua import ua
6
7		import sqlite3
8	1
9
10		class HistoryStorageInterface(object):
11
12		"""
13		Interface of a history backend.
14		Must be implemented by backends
15	1	"""
16
17		def new_historized_node(self, node, period, count=0):
18		"""
19		Called when a new node is to be historized
20		Returns None
21		"""
22	1	raise NotImplementedError
23
24		def save_node_value(self, node, datavalue):
25		"""
26		Called when the value of a historized node has changed and should be saved in history
27		Returns None
28		"""
29	1	raise NotImplementedError
30
31		def read_node_history(self, node, start, end, nb_values):
32		"""
33		Called when a client make a history read request for a node
34		if start or end is missing then nb_values is used to limit query
35		nb_values is the max number of values to read. Ignored if 0
36		Start time and end time are inclusive
37		Returns a list of DataValues and a continuation point which
38		is None if all nodes are read or the ServerTimeStamp of the last rejected DataValue
39		"""
40	1	raise NotImplementedError
41
42		def new_historized_event(self, event, period):
43		"""
44		Called when historization of events is enabled on server side
45		FIXME: we may need to store events per nodes in future...
46		Returns None
47		"""
48	1	raise NotImplementedError
49
50		def save_event(self, event):
51		"""
52		Called when a new event has been generated ans should be saved in history
53		Returns None
54		"""
55	1	raise NotImplementedError
56
57		def read_event_history(self, start, end, evfilter):
58		"""
59		Called when a client make a history read request for events
60		Start time and end time are inclusive
61		Returns a list of Events and a continuation point which
62		is None if all events are read or the ServerTimeStamp of the last rejected event
63		"""
64		raise NotImplementedError
65	1
66		def stop(self):
67		"""
68		Called when the server shuts down
69	1	Can be used to close database connections etc.
70	1	"""
71	1	raise NotImplementedError
72	1
73
74	1	class HistoryDict(HistoryStorageInterface):
75	1	"""
76	1	very minimal history backend storing data in memory using a Python dictionnary
77		"""
78	1	def __init__(self):
79		self._datachanges = {}
80		self._datachanges_period = {}
81	1	self._events = {}
82	1
83	1	def new_historized_node(self, node, period, count=0):
84	1	node_id = node.nodeid
85	1	self._datachanges[node_id] = []
86	1	self._datachanges_period[node_id] = period, count
87	1
88		def save_node_value(self, node, datavalue):
89	1	node_id = node.nodeid
90	1	data = self._datachanges[node_id]
91		period, count = self._datachanges_period[node_id]
92	1	data.append(datavalue)
93	1	now = datetime.now()
94	1	if period:
95		while now - data[0].ServerTimestamp > period:
96		data.pop(0)
97		if count and len(data) > count:
98	1	data = data[-count:]
99
100	1	def read_node_history(self, node, start, end, nb_values):
101		node_id = node.NodeId
102	1	cont = None
103	1	if node_id not in self._datachanges:
104	1	print("Error attempt to read history for a node which is not historized")
105	1	return [], cont
106	1	else:
107		if end is None:
108	1	end = datetime.now() + timedelta(days=1)
109	1	if start is None:
110		start = ua.DateTimeMinValue
111	1	results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp <= end]
112		if nb_values:
113		if start > ua.DateTimeMinValue and len(results) > nb_values:
114	1	cont = results[nb_values + 1].ServerTimestamp
115		results = results[:nb_values]
116		else:
117		results = results[-nb_values:]
118	1	return results, cont
119	1
120	1	def new_historized_event(self, event, period):
121		self._events = []
122	1
123	1	def save_event(self, event):
124		raise NotImplementedError
125
126	1	def read_event_history(self, start, end, evfilter):
127		raise NotImplementedError
128
129		def stop(self):
130	1	pass
131	1
132	1
133	1	class HistorySQLite(HistoryStorageInterface):
134	1	"""
135	1	very minimal history backend storing data in SQLite database
136		"""
137	1	# FIXME: need to check on if sql_conn.commit() should be inside try block; and if .rollback() needs to be except
138
139		def __init__(self):
140	1	self._datachanges_period = {}
141	1	self._events = {}
142	1	self._db_file = "history.db"
143	1
144	1	# SQL objects must be accessed in a single thread
145	1	# adding a new node to be historized probably happens on the main thread
146	1	self._conn_new = None
147	1	self._c_new = None
148	1
149		# subscriptions are in another thread so it needs it's own sqlite connection object
150	1	self._conn_sub = None
151	1	self._c_sub = None
152	1
153	1	# FIXME: no idea what thread the read values happen in, just make a new conn object for now
154		self._conn_read = None
155	1	self._c_read = None
156	1
157	1	def new_historized_node(self, node, period, count=0):
158		if self._conn_new is None:
159	1	self._conn_new = sqlite3.connect(self._db_file)
160	1	self._c_new = self._conn_new.cursor()
161	1
162		node_id = str(node.nodeid.NamespaceIndex) + '_' + str(node.nodeid.Identifier)
163	1	self._datachanges_period[node_id] = period
164
165		self._conn_new = sqlite3.connect(self._db_file)
166		self._c_new = self._conn_new.cursor()
167
168		sql_type = self._get_sql_type(node)
169	1
170		# create a table for the node which will store attributes of the DataValue object
171	1	try:
172	1	self._c_new.execute('CREATE TABLE "{tn}" (ServerTimestamp TIMESTAMP,'
173	1	' SourceTimestamp TIMESTAMP,'
174	1	' StatusCode INTEGER,'
175		' Value {type},'
176	1	' VariantType INTEGER)'.format(tn=node_id, type=sql_type))
177
178		except sqlite3.Error as e:
179	1	print(node_id, 'Historizing SQL Table Creation Error:', e)
180	1
181	1	self._conn_new.commit()
182		self._conn_new.close()
183
184		def save_node_value(self, node, datavalue):
185	1	if self._conn_sub is None:
186	1	self._conn_sub = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES)
187	1	self._c_sub = self._conn_sub.cursor()
188	1
189		node_id = str(node.nodeid.NamespaceIndex) + '_' + str(node.nodeid.Identifier)
190
191		# insert the data change into the database
192		try:
193		self._c_sub.execute('INSERT INTO "{tn}" VALUES (?, ?, ?, ?, ?)'.format(tn=node_id), (datavalue.ServerTimestamp,
194		datavalue.SourceTimestamp,
195		datavalue.StatusCode.value,
196		datavalue.Value.Value,
197		datavalue.Value.VariantType.value))
198		except sqlite3.Error as e:
199		print(node_id, 'Historizing SQL Insert Error:', e)
200	1
201		# get this node's period from the period dict and calculate the limit
202	1	period = self._datachanges_period[node_id]
203	1	date_limit = datetime.now() - period
204	1
205		# after the insert, delete all values older than period
206		try:
207		self._c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=node_id), (date_limit.isoformat(' '),))
208		except sqlite3.Error as e:
209		print(node_id, 'Historizing SQL Delete Old Data Error:', e)
210
211		self._conn_sub.commit()
212	1
213		def read_node_history(self, node, start, end, nb_values):
214		if self._conn_read is None:
215		self._conn_read = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES)
216	1	self._c_read = self._conn_read.cursor()
217
218	1	if end is None:
219		end = datetime.now() + timedelta(days=1)
220		if start is None:
221		start = ua.DateTimeMinValue
222	1
223		node_id = str(node.NodeId.NamespaceIndex) + '_' + str(node.NodeId.Identifier)
224	1	cont = None
225		results = []
226
227		start_time = start.isoformat(' ')
228		end_time = end.isoformat(' ')
229
230		# select values from the database
231		try:
232		for row in self._c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
233		'LIMIT ?'.format(tn=node_id), (start_time, end_time, nb_values,)):
234
235		dv = ua.DataValue(ua.Variant(row[3], self._get_variant_type(row[4])))
236		dv.ServerTimestamp = row[0]
237		dv.SourceTimestamp = row[1]
238		dv.StatusCode = ua.StatusCode(row[2])
239
240		results.append(dv)
241
242		except sqlite3.Error as e:
243		print(node_id, 'Historizing SQL Read Error:', e)
244
245		return results, cont
246
247		def new_historized_event(self, event, period):
248		raise NotImplementedError
249
250		def save_event(self, event):
251		raise NotImplementedError
252
253		def read_event_history(self, start, end, evfilter):
254		raise NotImplementedError
255
256		# convert the node UA Variant type to an SQL supported type
257		# FIXME: this could lead to lost precision! Better way to store the data? Add custom types to SQL?
258		def _get_sql_type(self, node):
259		node_type = node.get_data_type()
260		# see object_ids.py
261		if node_type.Identifier in (10, 11,): # Float, Double
262		return 'REAL'
263		elif node_type.Identifier in (4, 5, 6, 7, 8, 9,): # Ints
264		return "INT"
265		elif node_type.Identifier in (12,): # String
266		return "TEXT"
267		else:
268		return "NULL"
269
270		# convert the OPC UA variant identifier stored in SQL back to a UA Variant
271		# FIXME: is there a util method someplace for getting this?
272		def _get_variant_type(self, identifier):
273		if identifier is 10:
274		return ua.VariantType.Float
275		elif identifier is 11:
276		return ua.VariantType.Double
277		elif identifier is 4:
278		return ua.VariantType.Int16
279		elif identifier is 5:
280		return ua.VariantType.UInt16
281		elif identifier is 6:
282		return ua.VariantType.Int32
283		elif identifier is 7:
284		return ua.VariantType.UInt32
285		elif identifier is 8:
286		return ua.VariantType.Int64
287		elif identifier is 9:
288		return ua.VariantType.UInt64
289		elif identifier is 12:
290		return ua.VariantType.String
291
292		# close connections to the history database when the server stops
293		def stop(self):
294		pass
295		# FIXME: Should close the database connections when the server stops, but because server.stop() is called
296		# FIXME: on a different thread than the SQL conn object, no idea how to do this at the moment
297
298
299		class SubHandler(object):
300		def __init__(self, storage):
301		self.storage = storage
302
303		def datachange_notification(self, node, val, data):
304		self.storage.save_node_value(node, data.monitored_item.Value) # SHOULD NOT GET NODE ID HERE
305
306		def event_notification(self, event):
307		self.storage.save_event(event)
308
309
310		class HistoryManager(object):
311		def __init__(self, iserver):
312		self.iserver = iserver
313		self.storage = HistorySQLite() # HistoryDict() HistorySQLite()
314		self._sub = None
315		self._handlers = {}
316
317		def set_storage(self, storage):
318		self.storage = storage
319
320		def _create_subscription(self, handler):
321		params = ua.CreateSubscriptionParameters()
322		params.RequestedPublishingInterval = 10
323		params.RequestedLifetimeCount = 3000
324		params.RequestedMaxKeepAliveCount = 10000
325		params.MaxNotificationsPerPublish = 0
326		params.PublishingEnabled = True
327		params.Priority = 0
328		return Subscription(self.iserver.isession, params, handler)
329
330		def historize(self, node, period=timedelta(days=7), count=0):
331		if not self._sub:
332		self._sub = self._create_subscription(SubHandler(self.storage))
333		if node in self._handlers:
334		raise ua.UaError("Node {} is allready historized".format(node))
335		self.storage.new_historized_node(node, period, count) # SHOULD NOT GET NODE ID HERE
336		handler = self._sub.subscribe_data_change(node)
337		self._handlers[node] = handler
338
339		def dehistorize(self, node):
340		self._sub.unsubscribe(self._handlers[node])
341		del(self._handlers[node])
342
343		def read_history(self, params):
344		"""
345		Read history for a node
346		This is the part AttributeService, but implemented as its own service
347		since it requires more logic than other attribute service methods
348		"""
349		results = []
350
351		for rv in params.NodesToRead:
352		res = self._read_history(params.HistoryReadDetails, rv)
353		results.append(res)
354		return results
355
356		def _read_history(self, details, rv):
357		"""
358		read history for a node
359		"""
360		result = ua.HistoryReadResult()
361		if isinstance(details, ua.ReadRawModifiedDetails):
362		if details.IsReadModified:
363		result.HistoryData = ua.HistoryModifiedData()
364		# we do not support modified history by design so we return what we have
365		else:
366		result.HistoryData = ua.HistoryData()
367		dv, cont = self._read_datavalue_history(rv, details)
368		result.HistoryData.DataValues = dv
369		result.ContinuationPoint = cont
370
371		elif isinstance(details, ua.ReadEventDetails):
372		result.HistoryData = ua.HistoryEvent()
373		# FIXME: filter is a cumbersome type, maybe transform it something easier
374		# to handle for storage
375		result.HistoryData.Events = self.storage.read_event_history(details.StartTime,
376		details.EndTime,
377		details.Filter)
378		else:
379		# we do not currently support the other types, clients can process data themselves
380		result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNotImplemented)
381		return result
382
383		def _read_datavalue_history(self, rv, details):
384		starttime = details.StartTime
385		if rv.ContinuationPoint:
386		# Spec says we should ignore details if cont point is present
387		# but they also say we can use cont point as timestamp to enable stateless
388		# implementation. This is contradictory, so we assume details is
389		# send correctly with continuation point
390		# starttime = bytes_to_datetime(rv.ContinuationPoint)
391		starttime = ua.unpack_datetime(rv.ContinuationPoint)
392
393		dv, cont = self.storage.read_node_history(rv, # SHOULD NOT GET NODE ID HERE!
394		starttime,
395		details.EndTime,
396		details.NumValuesPerNode)
397		if cont:
398		# cont = datetime_to_bytes(dv[-1].ServerTimestamp)
399		cont = ua.pack_datetime(dv[-1].ServerTimestamp)
400		# FIXME, parse index range and filter out if necessary
401		# rv.IndexRange
402		# rv.DataEncoding # xml or binary, seems spec say we can ignore that one
403		return dv, cont
404
405		def update_history(self, params):
406		"""
407		Update history for a node
408		This is the part AttributeService, but implemented as its own service
409		since it requires more logic than other attribute service methods
410		"""
411		results = []
412		for _ in params.HistoryUpdateDetails:
413		result = ua.HistoryUpdateResult()
414		# we do not accept to rewrite history
415		result.StatusCode = ua.StatusCode(ua.StatusCodes.BadNotWritable)
416		results.append(results)
417		return results
418
419		def stop(self):
420		self.storage.stop()
421

FreeOpcUa / python-opcua

Pull Request — master (#155)

opcua.server.HistoryDict.stop() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like