fix: Fixed serialization of DataFrame with empty values #118

Author: rolincova

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ### Bug Fixes
 1. [#117](https://github.com/influxdata/influxdb-client-python/pull/117): Fixed appending default tags for single Point 
+1. [#118](https://github.com/influxdata/influxdb-client-python/issues/118): Fixed serialization of DataFrame with empty values
 
 ## 1.8.0 [2020-06-19]
 

influxdb_client/client/write_api.py

@@ -1,9 +1,12 @@
 # coding: utf-8
 import logging
 import os
+import re
 from collections import defaultdict
 from datetime import timedelta
 from enum import Enum
+from functools import reduce
+from itertools import chain
 from random import random
 from time import sleep
 from typing import Union, List, Any
@@ -307,6 +310,20 @@ def _itertuples(self, data_frame):
         cols = [data_frame.iloc[:, k] for k in range(len(data_frame.columns))]
         return zip(data_frame.index, *cols)
 
+    def _replace(self, data_frame):
+        from ..extras import np
+        obj_cols = {k for k, v in dict(data_frame.dtypes).items() if v is np.dtype('O')}
+        other_cols = set(data_frame.columns) - obj_cols
+        obj_nans = (f'{k}="nan"' for k in obj_cols)
+        other_nans = (f'{k}=nani?' for k in other_cols)
+        replacements = [
+            ('|'.join(chain(obj_nans, other_nans)), ''),
+            (',{2,}', ','),
+            ('|'.join([', ,', ', ', ' ,']), ' '),
+        ]
+
+        return replacements
+
     def _data_frame_to_list_of_points(self, data_frame, precision, **kwargs):
         from ..extras import pd, np
         if not isinstance(data_frame, pd.DataFrame):
@@ -352,7 +369,16 @@ def _data_frame_to_list_of_points(self, data_frame, precision, **kwargs):
                ' ', ','.join(fields), ' {p[0].value}')
         f = eval("lambda p: f'{}'".format(''.join(fmt)))
 
-        return list(map(f, self._itertuples(data_frame)))
+        isnull = data_frame.isnull().any(axis=1)
+
+        if isnull.any():
+            lp = map(f, self._itertuples(data_frame[~isnull]))
+            rep = self._replace(data_frame)
+            lp_nan = (reduce(lambda a, b: re.sub(*b, a), rep, f(p))
+                      for p in self._itertuples(data_frame[isnull]))
+            return list(chain(lp, lp_nan))
+        else:
+            return list(map(f, self._itertuples(data_frame)))
 
     def _http(self, batch_item: _BatchItem):
 

tests/test_WriteApiDataFrame.py

@@ -86,3 +86,32 @@ def test_write_num_py(self):
         self.assertEqual(result[0].records[1].get_value(), 200.0)
 
         pass
+
+    def test_write_nan(self):
+        from influxdb_client.extras import pd, np
+
+        now = pd.Timestamp('2020-04-05 00:00+00:00')
+
+        data_frame = pd.DataFrame(data=[[3.1955, np.nan, 20.514305, np.nan],
+                                        [5.7310, np.nan, 23.328710, np.nan],
+                                        [np.nan, 3.138664, np.nan, 20.755026],
+                                        [np.nan, 5.139563, np.nan, 19.791240]],
+                                  index=[now, now + timedelta(minutes=30), now + timedelta(minutes=60),
+                                         now + timedelta(minutes=90)],
+                                  columns=["actual_kw_price", "forecast_kw_price", "actual_general_use",
+                                           "forecast_general_use"])
+
+        points = self.client.write_api(write_options=SYNCHRONOUS)\
+            ._data_frame_to_list_of_points(data_frame=data_frame,
+                                           precision=None,
+                                           data_frame_measurement_name='measurement')
+
+        self.assertEqual(4, len(points))
+        self.assertEqual("measurement actual_kw_price=3.1955,actual_general_use=20.514305 1586044800000000000",
+                         points[0])
+        self.assertEqual("measurement actual_kw_price=5.731,actual_general_use=23.32871 1586046600000000000",
+                         points[1])
+        self.assertEqual("measurement forecast_kw_price=3.138664,forecast_general_use=20.755026 1586048400000000000",
+                         points[2])
+        self.assertEqual("measurement forecast_kw_price=5.139563,forecast_general_use=19.79124 1586050200000000000",
+                         points[3])