Merge pull request #153 from HydrologicEngineeringCenter/fix/mutltitimeseries

fix to save mutitimeseries call

Author: Enovotny

cwms/cwms_types.py

@@ -62,8 +62,8 @@ def get_df_data(data: JSON, selector: str) -> JSON:
         def rating_type(data: JSON) -> DataFrame:
             # grab the correct point values for a rating table
             df = DataFrame(data["point"]) if data["point"] else DataFrame()
-            df = df.apply(to_numeric)
-            return df
+            df_numeric = df.apply(to_numeric, axis=0, result_type="expand")
+            return DataFrame(df_numeric)
 
         def timeseries_type(orig_json: JSON, value_json: JSON) -> DataFrame:
             # if timeseries values are present then grab the values and put into

cwms/timeseries/timeseries.py

@@ -247,64 +247,69 @@ def timeseries_df_to_json(
     df = df.reindex(columns=["date-time", "value", "quality-code"])
     if df.isnull().values.any():
         raise ValueError("Null/NaN data must be removed from the dataframe")
-
+    if version_date:
+        version_date_iso = version_date.isoformat()
+    else:
+        version_date_iso = None
     ts_dict = {
         "name": ts_id,
         "office-id": office_id,
         "units": units,
         "values": df.values.tolist(),
-        "version-date": version_date,
+        "version-date": version_date_iso,
     }
 
     return ts_dict
 
 
 def store_multi_timeseries_df(
-    ts_data: pd.DataFrame, office_id: str, max_workers: Optional[int] = 30
+    data: pd.DataFrame, office_id: str, max_workers: Optional[int] = 30
 ) -> None:
-
     def store_ts_ids(
         data: pd.DataFrame,
         ts_id: str,
         office_id: str,
         version_date: Optional[datetime] = None,
     ) -> None:
-        units = data["units"].iloc[0]
-        data_json = timeseries_df_to_json(
-            data=data,
-            ts_id=ts_id,
-            units=units,
-            office_id=office_id,
-            version_date=version_date,
-        )
-        store_timeseries(data=data_json)
+        try:
+            units = data["units"].iloc[0]
+            data_json = timeseries_df_to_json(
+                data=data,
+                ts_id=ts_id,
+                units=units,
+                office_id=office_id,
+                version_date=version_date,
+            )
+            store_timeseries(data=data_json)
+        except Exception as e:
+            print(f"Error processing {ts_id}: {e}")
         return None
 
+    ts_data_all = data.copy()
+    if "version_date" not in ts_data_all.columns:
+        ts_data_all = ts_data_all.assign(version_date=pd.to_datetime(pd.Series([])))
     unique_tsids = (
-        ts_data["ts_id"].astype(str) + ":" + ts_data["version_date"].astype(str)
+        ts_data_all["ts_id"].astype(str) + ":" + ts_data_all["version_date"].astype(str)
     ).unique()
 
     with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-        for ts_id_all in unique_tsids:
-            try:
-                ts_id, version_date = ts_id_all.split(":", 1)
-                if version_date != "NaT":
-                    version_date_dt = pd.to_datetime(version_date)
-                    data = ts_data[
-                        (ts_data["ts_id"] == ts_id)
-                        & (ts_data["version_date"] == version_date_dt)
-                    ]
-                else:
-                    version_date_dt = None
-                    data = ts_data[
-                        (ts_data["ts_id"] == ts_id) & ts_data["version_date"].isna()
-                    ]
-                if not data.empty:
-                    executor.submit(
-                        store_ts_ids, data, ts_id, office_id, version_date_dt
-                    )
-            except Exception as e:
-                print(f"Error processing {ts_id}: {e}")
+        for unique_tsid in unique_tsids:
+            ts_id, version_date = unique_tsid.split(":", 1)
+            if version_date != "NaT":
+                version_date_dt = pd.to_datetime(version_date)
+                ts_data = ts_data_all[
+                    (ts_data_all["ts_id"] == ts_id)
+                    & (ts_data_all["version_date"] == version_date_dt)
+                ]
+            else:
+                version_date_dt = None
+                ts_data = ts_data_all[
+                    (ts_data_all["ts_id"] == ts_id) & ts_data_all["version_date"].isna()
+                ]
+            if not data.empty:
+                executor.submit(
+                    store_ts_ids, ts_data, ts_id, office_id, version_date_dt
+                )
 
 
 def store_timeseries(

pyproject.toml

@@ -2,7 +2,7 @@
 name = "cwms-python"
 repository = "https://github.com/HydrologicEngineeringCenter/cwms-python"
 
-version = "0.7.0"
+version = "0.7.1"
 
 
 packages = [