Code Duplication    Length = 7-10 lines in 2 locations

opcua/common/subscription.py 1 location

@@ 315-324 (lines=10) @@
312
        mids = []
313
        # process result, add server_handle, or remove it if failed
314
        with self._lock:
315
            for idx, result in enumerate(results):
316
                mi = params.ItemsToCreate[idx]
317
                if not result.StatusCode.is_good():
318
                    del self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
319
                    mids.append(result.StatusCode)
320
                    continue
321
                data = self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
322
                data.server_handle = result.MonitoredItemId
323
                mids.append(result.MonitoredItemId)
324
        return mids
325
326
    def unsubscribe(self, handle):
327
        """

opcua/server/event.py 1 location

@@ 111-117 (lines=7) @@
108
                self.EventType = node.nodeid
109
                curr_node = node
110
111
                while curr_node.nodeid.Identifier != parent_identifier:
112
                    for prop in curr_node.get_properties():
113
                        setattr(self, prop.get_browse_name().Name, prop.get_value())
114
                    parents = curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
115
                    if len(parents) != 1: # Something went wrong
116
                        return None
117
                    curr_node = parents[0]
118
119
                self._freeze = True
120