Merge branch 'bug_in_tps_rt_9926' into 'develop'

Handling overflow data in tps

Closes #286

See merge request !340

sohstationviewer/view/plotting/time_power_square/time_power_squared_helper.py

@@ -109,9 +109,10 @@ def get_tps_for_discontinuous_data(
     mean_squares = np.zeros_like(start_5min_blocks, dtype=float)
 
     # Calculate mean_square for each 5m block
+    # adding dtype=float64 to prevent integer overflow
     for i, d in enumerate(split_data):
         if len(d) != 0:
-            mean_squares[i] = np.mean(np.square(d))
+            mean_squares[i] = np.mean(np.square(d, dtype='float64'))
         elif ((0 < i < len(split_data) - 1) and
               len(split_data[i - 1]) > 0 and len(split_data[i + 1]) > 0):
             """
@@ -119,7 +120,8 @@ def get_tps_for_discontinuous_data(
             data in the previous and next blocks if they both have data.
             """
             mean_squares[i] = np.mean(np.square(
-                np.hstack((split_data[i - 1], split_data[i + 1]))
+                np.hstack((split_data[i - 1], split_data[i + 1])),
+                dtype='float64'
             ))
     # reshape 1D mean_quares into 2D array in which each row contains 288 of
     # 5m blocks' mean_squares of a day

tests/view/plotting/test_time_power_square_helper.py

@@ -156,6 +156,19 @@ class TestGetTPSForDiscontinuousData(BaseTestCase):
         # last block of day0 has value
         self.assertIn(const.NUMBER_OF_5M_IN_DAY - 1, day0_indexes)
 
+    def test_overflow_data(self):
+        times = np.arange(self.start, self.end, 9*60)    # 9m apart
+        # This data reproduce overflow data
+        data = np.random.randint(-10**6, 0, times.size, dtype='i4')
+        channel_data = {'tracesInfo': [{'times': times, 'data': data}]}
+        tps = get_tps_for_discontinuous_data(
+            channel_data, self.start_5mins_blocks)
+        self.assertEqual(len(tps), 2)
+        # Before adding dtype in np.square in get_tps_for_discontinuous_data,
+        # some tps data would be less than 0
+        lessthanzero_indexes = np.where(tps[0] < 0)[0]
+        self.assertEqual(lessthanzero_indexes.size, 0)
+
 
 class TestGetTPSTimeByColorForADay(BaseTestCase):
     def setUp(self):