diff --git a/python/lsst/consdb/transform_efd.py b/python/lsst/consdb/transform_efd.py index d7822eb5..9c7f4095 100644 --- a/python/lsst/consdb/transform_efd.py +++ b/python/lsst/consdb/transform_efd.py @@ -1,6 +1,6 @@ import argparse import asyncio -from typing import Any, Callable, Optional +from typing import Any, Callable, Optional, Union import astropy.time import lsst_efd_client @@ -8,12 +8,120 @@ import yaml from lsst.daf.butler import Butler, DimensionRecord from sqlalchemy import Engine, create_engine - +import numpy class Summary: # TODO define summary pass +class Transform: + """ + A class that performs various transformations on a list or numpy array of values. + """ + + def __init__(self, values: Union[List[Union[float, int, bool]], numpy.ndarray]): + """ + Initialize the Transform object. + + Args: + values: A list or numpy array of values. + + Returns: + None + """ + self.values = np.array(values) + + def mean(self) -> float: + """ + Calculate the mean of the values. + + Returns: + The mean value as a float. + """ + return numpy.nanmean(self.values) + + def std(self, ddof: Optional[int] = 1) -> float: + """ + Calculate the standard deviation of the values. + + Args: + ddof: Delta degrees of freedom. + + Returns: + The standard deviation as a float. + """ + return numpy.nanstd(self.values, ddof=ddof) + + def max(self) -> Union[float, int, bool]: + """ + Find the maximum value in the values. + + Returns: + The maximum value as a float, int, or bool. + """ + return numpy.nanmax(self.values) + + def min(self) -> Union[float, int, bool]: + """ + Find the minimum value in the values. + + Returns: + The minimum value as a float, int, or bool. + """ + return numpy.nanmin(self.values) + + def logical_and(self) -> Union[bool, numpy.ndarray]: + """ + Perform element-wise logical AND operation on the values. + + Returns: + The result of the logical AND operation as a bool or numpy array. + """ + return numpy.logical_and(self.values) + + def logical_or(self) -> Union[bool, numpy.ndarray]: + """ + Perform element-wise logical OR operation on the values. + + Returns: + The result of the logical OR operation as a bool or numpy array. + """ + return numpy.logical_or(self.values) + + def logical_not(self) -> numpy.ndarray: + """ + Perform element-wise logical NOT operation on the values. + + Returns: + The result of the logical NOT operation as a numpy array. + """ + return numpy.logical_not(self.values) + + def logical_xor(self, other_values: Union[List[Union[float, int, bool]], numpy.ndarray]) -> numpy.ndarray: + """ + Perform element-wise logical XOR operation between the values and other values. + + Args: + other_values: A list or numpy array of other values. + + Returns: + The result of the logical XOR operation as a numpy array. + """ + other_values = numpy.array(other_values) + return numpy.logical_xor(self.values, other_values) + + def percentile(self, q: float) -> Union[float, int, bool]: + """ + Calculate the q-th percentile of the values. + + Args: + q: The percentile value. + + Returns: + The q-th percentile value as a float, int, or bool. + """ + return numpy.nanpercentile(self.values, q) + # TODO add all summarizing functions def gen_mean( @@ -157,7 +265,7 @@ def get_visits_by_period( def butler_query_results_to_pandas(query): return pandas.DataFrame([q.toDict() for q in query]) -def get_topic_correponding_indexes(butler_table: pandas.DataFrame, topic_time_array: Callable[[list],numpy.ndarray]) -> List: +def get_topic_correponding_indexes(butler_table: pandas.DataFrame, topic_time_array: Union[[list],numpy.ndarray]) -> List: """ Returns a list of topic indexes corresponding to each row in the butler_table. It is assumed that the butler_table and the topic_time_array are sorted in time ascending order.