#802: first PoC of (not working) wrap function

psutil/_common.py

@@ -16,6 +16,7 @@
 import socket
 import stat
 import sys
+import threading
 import warnings
 from collections import namedtuple
 from socket import AF_INET
@@ -465,3 +466,67 @@ def inner(self, *args, **kwargs):
             return getattr(self, replacement)(*args, **kwargs)
         return inner
     return outer
+
+
+_wrapn_lock = threading.Lock()
+_wrapn_cache = {}
+
+
+def wrap_numbers(new_dict, name):
+    def did_nums_wrap(new_nt, old_nt):
+        # Return True if one of the numbers of the new ntuple is smaller
+        # than the number of the old ntuple in the same position.
+        for i in range(len(new_nt)):
+            new_value = new_nt[i]
+            old_value = old_nt[i]
+            if new_value < old_value:
+                return True
+        return False
+
+    def replace_ntuple(new_nt, old_nt):
+        # Return a new ntuple with the "adjusted" numbers.
+        bits = []
+        for i in range(len(new_nt)):
+            new_value = new_nt[i]
+            old_value = old_nt[i]
+            if new_value < old_value:
+                # they wrapped!
+                num = old_value + new_value
+            else:
+                num = new_value
+            bits.append(num)
+        return new_nt._make(bits)
+
+    with _wrapn_lock:
+        if name not in _wrapn_cache:
+            # This was the first call.
+            _wrapn_cache[name] = new_dict
+            return new_dict
+
+        old_dict = _wrapn_cache[name]
+        for key, new_nt in new_dict.items():
+            try:
+                old_nt = old_dict[key]
+            except KeyError:
+                # We may get here if net_io_counters() returned a new NIC
+                # or disk_io_counters() returned a new disk.
+                continue
+            else:
+                assert new_nt._fields == old_nt._fields, (new_nt, old_nt)
+                if did_nums_wrap(new_nt, old_nt):
+                    # we wrapped!
+                    new_dict[key] = replace_ntuple(new_nt, old_nt)
+
+        _wrapn_cache[name] = new_dict
+        return new_dict
+
+
+def _wrapn_cache_clear(name=None):
+    with _wrapn_lock:
+        if name is None:
+            _wrapn_cache.clear()
+        else:
+            _wrapn_cache.pop(name)
+
+
+wrap_numbers.cache_clear = _wrapn_cache_clear

psutil/tests/test_misc.py

@@ -20,13 +20,15 @@
 import socket
 import stat
 import sys
+from collections import namedtuple
 
 from psutil import LINUX
 from psutil import POSIX
 from psutil import WINDOWS
 from psutil._common import memoize
 from psutil._common import memoize_when_activated
 from psutil._common import supports_ipv6
+from psutil._common import wrap_numbers
 from psutil._compat import PY3
 from psutil.tests import APPVEYOR
 from psutil.tests import bind_socket
@@ -377,6 +379,49 @@ def test_sanity_version_check(self):
             self.assertIn("version conflict", str(cm.exception).lower())
 
 
+nt = namedtuple('foo', 'a b c')
+
+
+class TestWrapNumbers(unittest.TestCase):
+
+    def tearDown(self):
+        wrap_numbers.cache_clear()
+
+    def test_first_call(self):
+        input = {'foo': nt(5, 5, 5)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+
+    def test_input_hasnt_changed(self):
+        input = {'foo': nt(5, 5, 5)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+
+    def test_increase_but_no_wrap(self):
+        input = {'foo': nt(5, 5, 5)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+        input = {'foo': nt(10, 15, 20)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+
+    def test_wrap_once(self):
+        input = {'foo': nt(5, 5, 5)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+        input = {'foo': nt(5, 5, 3)}
+        self.assertEqual(wrap_numbers(input, 'funname'),
+                         {'foo': nt(5, 5, 8)})
+
+    def test_keep_wrapping(self):
+        input = {'foo': nt(100, 100, 100)}
+        self.assertEqual(wrap_numbers(input, 'funname'), input)
+        # wrap from 5, expect 105
+        input = {'foo': nt(100, 100, 5)}
+        self.assertEqual(wrap_numbers(input, 'funname'),
+                         {'foo': nt(100, 100, 105)})
+        # next go to 10, expect 115
+        input = {'foo': nt(100, 100, 10)}
+        self.assertEqual(wrap_numbers(input, 'funname'),
+                         {'foo': nt(100, 100, 115)})
+
+
 # ===================================================================
 # --- Example script tests
 # ===================================================================