Rolling aggregation of ZTF data (#164)

* Move test file

* Make period test more frugal

* Numpy 2.0 warning

* Update test wrappers

* Add functionality to aggregate ZTF data

* PEP8

* Make name more generic

* Add time management

* PEP8

* hdfs utilities

Author: JulienPeloton

fink_utils/broker/avroUtils.py

@@ -17,7 +17,7 @@
 import io
 import fastavro
 
-from fink_utils.test.tester import regular_unit_tests
+from fink_utils.tester import regular_unit_tests
 
 __all__ = ["writeavrodata", "readschemadata", "readschemafromavrofile"]
 

fink_utils/broker/distributionUtils.py

@@ -25,8 +25,6 @@
 from pyspark.sql.functions import struct, lit
 from pyspark.sql.avro.functions import to_avro as to_avro_native
 
-# from fink_utils.test.tester import spark_unit_tests
-
 
 def get_kafka_df(
     df: DataFrame,

fink_utils/broker/sparkUtils.py

@@ -24,8 +24,6 @@
 
 from fink_utils.broker.avroUtils import readschemafromavrofile
 
-# from fink_utils.test.tester import spark_unit_tests
-
 
 def from_avro(dfcol: Column, jsonformatschema: str) -> Column:
     """Decode the Avro data contained in a DataFrame column into a struct.

fink_utils/cutouts/utils.py

@@ -17,7 +17,7 @@
 from astropy.io import fits
 import numpy as np
 
-from fink_utils.test.tester import regular_unit_tests
+from fink_utils.tester import regular_unit_tests
 
 
 def unzip_cutout(stamp):

fink_utils/test/__init__.py fink_utils/hdfs/__init__.pyfink_utils/test/__init__.py renamed to fink_utils/hdfs/__init__.py

@@ -0,0 +1,48 @@
+# Copyright 2025 AstroLab Software
+# Author: Julien Peloton
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Contains functionalities to work with HDFS"""
+
+from pyspark.sql import SparkSession
+
+
+def path_exist(path: str) -> bool:
+    """Check if a path exists on Spark shared filesystem (HDFS or S3)
+
+    Parameters
+    ----------
+    path : str
+        Path to check
+
+    Returns
+    -------
+    bool
+        True if the path exists, False otherwise
+    """
+    spark = SparkSession.builder.getOrCreate()
+
+    jvm = spark._jvm
+    jsc = spark._jsc
+
+    conf = jsc.hadoopConfiguration()
+    uri = jvm.java.net.URI(path)
+
+    fs = jvm.org.apache.hadoop.fs.FileSystem.get(uri, conf)
+
+    path_glob = jvm.org.apache.hadoop.fs.Path(path)
+    status_list = fs.globStatus(path_glob)
+    if len(list(status_list)) > 0:
+        return True
+    else:
+        return False

fink_utils/hdfs/utils.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 import numpy as np
 
-from fink_utils.test.tester import regular_unit_tests
+from fink_utils.tester import regular_unit_tests
 
 
 def is_source_behind(distnr: float, chinr: float = None, sharpnr: float = None) -> bool:

fink_utils/photometry/utils.py

@@ -20,8 +20,6 @@
 import pandas as pd
 from astropy.time import Time
 
-# from fink_utils.test.tester import spark_unit_tests_broker
-
 
 @pandas_udf(TimestampType(), PandasUDFType.SCALAR)
 def convert_to_millitime(jd: pd.Series, format=None, now=None):

fink_utils/spark/partitioning.py

@@ -34,7 +34,7 @@
 
 import logging
 
-from fink_utils.test.tester import regular_unit_tests
+from fink_utils.tester import regular_unit_tests
 
 _LOG = logging.getLogger(__name__)
 
@@ -258,10 +258,10 @@ def estimate_synodic_period(
     Examples
     --------
     >>> ssnamenr = 2363
-    >>> P, chi2 = estimate_synodic_period(ssnamenr, flavor="SHG1G2", Nterms_base=2)
-    >>> assert int(P) == 20, P
+    >>> P, chi2 = estimate_synodic_period(ssnamenr, flavor="SHG1G2", Nterms_base=1)
+    >>> assert int(P) < 48, P
 
-    >>> P_HG, chi2_HG = estimate_synodic_period(ssnamenr, flavor="HG", Nterms_base=2)
+    >>> P_HG, chi2_HG = estimate_synodic_period(ssnamenr, flavor="HG", Nterms_base=1)
     >>> assert chi2 < chi2_HG, (chi2, chi2_HG)
 
     # by default we apply the light travel correction. Disable it.

fink_utils/sso/periods.py

@@ -17,7 +17,7 @@
 from scipy import linalg
 
 from fink_utils.sso.utils import estimate_axes_ratio
-from fink_utils.test.tester import regular_unit_tests
+from fink_utils.tester import regular_unit_tests
 
 
 def sort_quantity_by_filter(filter, quantity):
@@ -910,7 +910,7 @@ def fit_legacy_models(
         lower_bounds = np.concatenate((lower_bounds, bounds[0]))
         upper_bounds = np.concatenate((upper_bounds, bounds[1]))
 
-    if not np.alltrue([i == i for i in magpsf_red]):
+    if not np.all([i == i for i in magpsf_red]):
         outdic = {"fit": 1, "status": -2}
         return outdic