Pythonise setup_stats and generate_statsMP

oneflux_steps/ustar_cp_python/cpdBootstrap.py

@@ -0,0 +1,93 @@
+import numpy as np
+from typing import Dict, List
+from oneflux_steps.ustar_cp_python.utilities import dot, intersect
+
+def cpdBootstrapUStarTh4Season20100901(*args, **kwargs):
+    """
+    cpdBootstrapUStarTh4Season20100901: Bootstrap the uStarTh for a season
+
+    Args:
+        *args: Variable length argument list.
+        **kwargs: Arbitrary keyword arguments.
+
+    Returns:
+        None
+    """
+    # TODO: implement this function
+    return None
+
+def generate_statsMT() -> Dict[str, float]:
+    """
+    Initialize a stats structure with NaN values for predefined fields.
+
+    Returns:
+        Dict[str, float]: Initialized stats dictionary with NaN values.
+    """
+    fields = [
+        "n", "Cp", "Fmax", "p", "b0", "b1", "b2", "c2",
+        "cib0", "cib1", "cic2", "mt", "ti", "tf",
+        "ruStarVsT", "puStarVsT", "mT", "ciT"
+    ]
+
+    # Initialize all fields with NaN
+    stats_mt = {field: np.nan for field in fields}
+
+    return stats_mt
+
+
+def setup_Stats(n_boot: int, n_seasons: int, n_strata_x: int) -> List[List[List[Dict[str, float]]]]|dict[str, float]:
+    """
+    Initialize the Stats structure based on input dimensions.
+
+    This function is correct where the args are all integers > 1.
+
+    Args:
+        n_boot (int): Number of bootstraps.
+        n_seasons (int): Number of seasons.
+        n_strata_x (int): Number of strata in X direction.
+
+    Returns:
+        List[List[List[Dict[str, float]]]]: Preallocated stats structure.
+    """
+    args_list = [n_boot, n_seasons, n_strata_x]
+    for n in args_list:
+        if n < 2:
+            raise ValueError(f'Function "setup_stats" has been passed an argument \
+                             with value {n}. This may lead undesired behaviour')
+
+    # Preallocate stats array
+    stats = [[[generate_statsMT() for _ in range(n_strata_x)]
+              for _ in range(n_seasons)]
+             for _ in range(n_boot)]
+
+    return stats
+
+def setup_Cp(nSeasons=None, nStrataX=None, nBoot=None):
+    # TODO: check definition, may need to use the definition in utils.py
+    return dot(np.nan, np.ones([nSeasons, nStrataX, nBoot]))
+
+# TODO: rough attempt in np
+def get_itNee(NEE=None, uStar=None, T=None, iNight=None):
+    itNee = np.where(np.logical_not(np.isnan(NEE + uStar + T)))
+    # itNee = intersect(itNee, iNight)
+    return []
+
+def get_ntN(t = None, nSeasons=None):
+    # TODO
+    return []
+
+def update_uStar(x):
+    # TODO
+    return 0
+
+def get_iNight(x):
+    # TODO
+    return 0
+
+def get_nPerBin(x):
+    # TODO
+    return 0
+
+def get_nPerDay(x):
+    # TODO
+    return 0

oneflux_steps/ustar_cp_python/utilities.py

@@ -60,6 +60,35 @@ def floor(a : np.ndarray) -> np.ndarray:
     """
     return np.asanyarray(a // 1).astype(int)
 
+def intersect(a : np.ndarray, b : np.ndarray) -> np.ndarray:
+    """
+    Return the intersection of two arrays.
+
+    Parameters:
+    a : np.ndarray
+        The first array to intersect
+
+    b : np.ndarray
+        The second array to intersect
+
+    nargout : int
+        The number of output arguments to return
+
+    Returns:
+    np.ndarray
+        The intersection of the two arrays
+    """
+    from builtins import set
+
+    c = sorted(set(a.flat) & set(b.flat))
+    if isinstance(a, str):
+        return "".join(c)
+    elif isinstance(a, list):
+        return c
+    else:
+        # FIXME: the result is a column vector if
+        # both args are column vectors; otherwise row vector
+        return np.array(c).reshape((1, -1) if a.shape[1] > 1 else (-1, 1))
 
 def jsonencode(a):
     return a if isinstance(a, cellarray) else json.dumps(a)

tests/conftest.py

@@ -93,6 +93,7 @@ def mf_factory(cls, *args, **kwargs):
 from oneflux_steps.ustar_cp_python.cpdFmax2pCp3 import *
 from oneflux_steps.ustar_cp_python.utilities import *
 from oneflux_steps.ustar_cp_python.cpd_evaluate_functions import *
+from oneflux_steps.ustar_cp_python.cpdBootstrap import *
 
 def pytest_addoption(parser):
     parser.addoption("--language", action="store", default="matlab")
@@ -178,6 +179,10 @@ def newfunc(*args, **kwargs):
                 # if nargout is present in kwargs then remove it
                 if 'nargout' in kwargs:
                     kwargs.pop('nargout')
+                # if jsonencode is present in kwargs then remove it
+                if 'jsonencode' in kwargs:
+                    kwargs.pop('jsonencode')
+
                 func = globals().get(name)
                 if callable(func):
                     return func(*args, **kwargs)
@@ -547,6 +552,7 @@ def to_matlab_type(data: Any) -> Any:
     """
     if isinstance(data, dict):
         # Convert a Python dictionary to a MATLAB struct
+        # TODO: the following doesn't actually work but is not yet used
         matlab_struct = matlab.struct()
         for key, value in data.items():
             matlab_struct[key] = to_matlab_type(value)  # Recursively handle nested structures