Code Duplication - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 23-23 lines in 2 locations

opcua/server/history.py 2 locations


        if count and len(evts) > count:
            evts.pop(0)

    def read_event_history(self, source_id, start, end, nb_values, evfilter):
        cont = None
        if source_id not in self._events:
            print("Error attempt to read event history for a node which does not historize events")
            return [], cont
        else:
            if start is None:
                start = ua.DateTimeMinValue
            if end is None:
                end = ua.DateTimeMinValue
            if start == ua.DateTimeMinValue:
                results = [ev for ev in reversed(self._events[source_id]) if start <= ev.Time]
            elif end == ua.DateTimeMinValue:
                results = [ev for ev in self._events[source_id] if start <= ev.Time]
            elif start > end:
                results = [ev for ev in reversed(self._events[source_id]) if end <= ev.Time <= start]

            else:
                results = [ev for ev in self._events[source_id] if start <= ev.Time <= end]
            if nb_values and len(results) > nb_values:
                cont = results[nb_values + 1].Time
                results = results[:nb_values]
            return results, cont

    def stop(self):
        pass

        if count and len(data) > count:
            data.pop(0)

    def read_node_history(self, node_id, start, end, nb_values):
        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 start is None:
                start = ua.DateTimeMinValue
            if end is None:
                end = ua.DateTimeMinValue
            if start == ua.DateTimeMinValue:
                results = [dv for dv in reversed(self._datachanges[node_id]) if start <= dv.ServerTimestamp]
            elif end == ua.DateTimeMinValue:
                results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp]
            elif start > end:
                results = [dv for dv in reversed(self._datachanges[node_id]) if end <= dv.ServerTimestamp <= start]

            else:
                results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp <= end]
            if nb_values and len(results) > nb_values:
                cont = results[nb_values + 1].ServerTimestamp
                results = results[:nb_values]
            return results, cont

    def new_historized_event(self, source_id, etype, period, count):
        if source_id in self._events:

		@@ 146-168 (lines=23) @@
143		if count and len(evts) > count:
144		evts.pop(0)
145
146		def read_event_history(self, source_id, start, end, nb_values, evfilter):
147		cont = None
148		if source_id not in self._events:
149		print("Error attempt to read event history for a node which does not historize events")
150		return [], cont
151		else:
152		if start is None:
153		start = ua.DateTimeMinValue
154		if end is None:
155		end = ua.DateTimeMinValue
156		if start == ua.DateTimeMinValue:
157		results = [ev for ev in reversed(self._events[source_id]) if start <= ev.Time]
158		elif end == ua.DateTimeMinValue:
159		results = [ev for ev in self._events[source_id] if start <= ev.Time]
160		elif start > end:
161		results = [ev for ev in reversed(self._events[source_id]) if end <= ev.Time <= start]
162
163		else:
164		results = [ev for ev in self._events[source_id] if start <= ev.Time <= end]
165		if nb_values and len(results) > nb_values:
166		cont = results[nb_values + 1].Time
167		results = results[:nb_values]
168		return results, cont
169
170		def stop(self):
171		pass
		@@ 105-127 (lines=23) @@
102		if count and len(data) > count:
103		data.pop(0)
104
105		def read_node_history(self, node_id, start, end, nb_values):
106		cont = None
107		if node_id not in self._datachanges:
108		print("Error attempt to read history for a node which is not historized")
109		return [], cont
110		else:
111		if start is None:
112		start = ua.DateTimeMinValue
113		if end is None:
114		end = ua.DateTimeMinValue
115		if start == ua.DateTimeMinValue:
116		results = [dv for dv in reversed(self._datachanges[node_id]) if start <= dv.ServerTimestamp]
117		elif end == ua.DateTimeMinValue:
118		results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp]
119		elif start > end:
120		results = [dv for dv in reversed(self._datachanges[node_id]) if end <= dv.ServerTimestamp <= start]
121
122		else:
123		results = [dv for dv in self._datachanges[node_id] if start <= dv.ServerTimestamp <= end]
124		if nb_values and len(results) > nb_values:
125		cont = results[nb_values + 1].ServerTimestamp
126		results = results[:nb_values]
127		return results, cont
128
129		def new_historized_event(self, source_id, etype, period, count):
130		if source_id in self._events:

FreeOpcUa / python-opcua

Code Duplication Length = 23-23 lines in 2 locations

opcua/server/history.py 2 locations