change datatype function and test

Author: otzi5300

mikeio/generic.py

@@ -51,6 +51,7 @@
     "quantile",
     "scale",
     "sum",
+    "change_datatype",
 ]
 
 
@@ -94,14 +95,19 @@ def _clone(
     start_time: datetime | None = None,
     timestep: float | None = None,
     items: Sequence[int | DfsDynamicItemInfo] | None = None,
+    datatype: int = None,
 ) -> DfsFile:
     source = DfsFileFactory.DfsGenericOpen(str(infilename))
     fi = source.FileInfo
 
     builder = DfsBuilder.Create(fi.FileTitle, "mikeio", __dfs_version__)
 
     # Set up the header
-    builder.SetDataType(fi.DataType)
+    if datatype is None:
+        builder.SetDataType(fi.DataType)
+    else:
+        builder.SetDataType(datatype)
+
     builder.SetGeographicalProjection(fi.Projection)
 
     # Copy time axis
@@ -959,3 +965,50 @@ def _get_repeated_items(
             new_items.append(item)
 
     return new_items
+
+
+def change_datatype(
+    infilename: str | pathlib.Path,
+    outfilename: str | pathlib.Path,
+    datatype: int,
+):
+    """Create temporal quantiles of all items in dfs file.
+
+    Parameters
+    ----------
+    infilename : str | pathlib.Path
+        input filename
+    outfilename : str | pathlib.Path
+        output filename
+    datatype: int
+        DataType to be used for the output file
+
+    Examples
+    --------
+    >>> quantile("in.dfsu", "out.dfsu", datatype=107)
+
+    >>> quantile("huge.dfsu", "Q01.dfsu", q=0.1, buffer_size=5.0e9)
+
+    """
+    dfs_out = _clone(infilename, outfilename, datatype=datatype)
+    dfs_in = DfsFileFactory.DfsGenericOpen(infilename)
+
+    item_numbers = _valid_item_numbers(dfs_in.ItemInfo)
+    n_items = len(item_numbers)
+    n_time_steps = dfs_in.FileInfo.TimeAxis.NumberOfTimeSteps
+    deletevalue = dfs_in.FileInfo.DeleteValueFloat
+
+    # Rewrite the data FIXME: can we do this in a more elegant way?
+    for timestep in trange(n_time_steps, disable=True):
+        for item in range(n_items):
+            itemdata = dfs_in.ReadItemTimeStep(item_numbers[item] + 1, timestep)
+            time = itemdata.Time
+            d = itemdata.Data
+            d[d == deletevalue] = np.nan
+            outdata = d
+            outdata[np.isnan(outdata)] = deletevalue
+            darray = outdata.astype(np.float32)
+            dfs_out.WriteItemTimeStep(item_numbers[item] + 1, timestep, time, darray)
+
+    dfs_out.Close()
+    dfs_in.Close()

tests/test_generic.py

@@ -3,8 +3,18 @@
 import pandas as pd
 import mikeio
 from mikeio import generic
-from mikeio.generic import scale, diff, sum, extract, avg_time, fill_corrupt, add
+from mikeio.generic import (
+    scale,
+    diff,
+    sum,
+    extract,
+    avg_time,
+    fill_corrupt,
+    add,
+    change_datatype,
+)
 import pytest
+from mikecore.DfsFileFactory import DfsFileFactory
 
 
 def test_add_constant(tmp_path: Path) -> None:
@@ -638,3 +648,43 @@ def test_fill_corrupt_data(tmp_path: Path) -> None:
     orig = mikeio.read(infile)
     extracted = mikeio.read(fp)
     assert extracted.n_timesteps == orig.n_timesteps
+
+
+def test_change_datatype_dfs0(tmp_path: Path) -> None:
+    infilename = "tests/testdata/random.dfs0"
+    outfilename = str(tmp_path / "random_datatype107.dfs0")
+    OUT_DATA_TYPE = 107
+
+    change_datatype(infilename, outfilename, datatype=OUT_DATA_TYPE)
+    dfs_out = DfsFileFactory.DfsGenericOpen(outfilename)
+    dfs_in = DfsFileFactory.DfsGenericOpen(infilename)
+
+    n_timesteps_in = dfs_in.FileInfo.TimeAxis.NumberOfTimeSteps
+    n_timesteps_out = dfs_out.FileInfo.TimeAxis.NumberOfTimeSteps
+    datatype_out = dfs_out.FileInfo.DataType
+
+    dfs_out.Close()
+    dfs_in.Close()
+
+    assert datatype_out == OUT_DATA_TYPE
+    assert n_timesteps_in == n_timesteps_out
+
+
+def test_change_datatype_dfsu(tmp_path: Path) -> None:
+    infilename = "tests/testdata/oresund_sigma_z.dfsu"
+    outfilename = str(tmp_path / "oresund_sigma_z_datatype107.dfsu")
+    OUT_DATA_TYPE = 107
+
+    change_datatype(infilename, outfilename, datatype=OUT_DATA_TYPE)
+    dfs_out = DfsFileFactory.DfsGenericOpen(outfilename)
+    dfs_in = DfsFileFactory.DfsGenericOpen(infilename)
+
+    n_timesteps_in = dfs_in.FileInfo.TimeAxis.NumberOfTimeSteps
+    n_timesteps_out = dfs_out.FileInfo.TimeAxis.NumberOfTimeSteps
+    datatype_out = dfs_out.FileInfo.DataType
+
+    dfs_out.Close()
+    dfs_in.Close()
+
+    assert datatype_out == OUT_DATA_TYPE
+    assert n_timesteps_in == n_timesteps_out