Refactor Metrics

examples/basic_example.py

@@ -100,12 +100,11 @@ def future_loading(t, x = None):
 
     # You can also use the metrics package to generate some useful metrics on the result of a prediction
     print("\nEOD Prediction Metrics")
-    toe = toe.key('EOD') # Calculate metrics for event EOL
 
-    from prog_algs.metrics import samples as metrics 
-    print('\tPercentage between 3005.2 and 3005.6: ', metrics.percentage_in_bounds(toe, [3005.2, 3005.6])*100.0, '%')
-    print('\tAssuming ground truth 3002.25: ', metrics.toe_metrics(toe, 3005.25))
-    print('\tP(Success) if mission ends at 3002.25: ', metrics.prob_success(toe, 3005.25))
+    from prog_algs.metrics import prob_success
+    print('\Portion between 3005.2 and 3005.6: ', toe.percentage_in_bounds([3005.2, 3005.6]))
+    print('\tAssuming ground truth 3002.25: ', toe.metrics(ground_truth=3005.25))
+    print('\tP(Success) if mission ends at 3002.25: ', prob_success(toe, 3005.25))
 
     # Plot state transition 
     # Here we will plot the states at t0, 25% to ToE, 50% to ToE, 75% to ToE, and ToE

src/prog_algs/metrics/__init__.py

@@ -1,3 +1,8 @@
 # Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration.  All Rights Reserved.
 
 from . import samples
+from .toe_metrics import prob_success
+from .uncertain_data_metrics import calc_metrics
+from .toe_profile_metrics import alpha_lambda
+
+__all__ = ['alpha_lambda', 'calc_metrics', 'prob_success']

src/prog_algs/metrics/samples.py

@@ -1,110 +1,23 @@
 # Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration.  All Rights Reserved.
 
-from numpy import mean, std
-from scipy import stats
-
-def toe_metrics(toe, ground_truth = None):
-    """Calculate all time of event metrics
-
-    Args:
-        toe ([double]): Times of event for a single event, output from predictor
-        ground_truth (float, optional): Ground truth end of discharge time. Defaults to None.
-
-    Returns:
-        dict: collection of metrics
-    """
-    toe.sort()
-    m = mean(toe)
-    median = toe[int(len(toe)/2)]
-    metrics = {
-        'min': toe[0],
-        'percentiles': {
-            '0.01': toe[int(len(toe)/10000)] if len(toe) >= 10000 else None,
-            '0.1': toe[int(len(toe)/1000)] if len(toe) >= 1000 else None,
-            '1': toe[int(len(toe)/100)] if len(toe) >= 100 else None,
-            '10': toe[int(len(toe)/10)] if len(toe) >= 10 else None,
-            '25': toe[int(len(toe)/4)] if len(toe) >= 4 else None,
-            '50': median,
-            '75': toe[int(3*len(toe)/4)] if len(toe) >= 4 else None,
-        },
-        'median': median,
-        'mean': m,
-        'std': std(toe),
-        'max': toe[-1],
-        'median absolute deviation': sum([abs(x - median) for x in toe])/len(toe),
-        'mean absolute deviation':   sum([abs(x - m)   for x in toe])/len(toe),
-        'number of samples': len(toe)
-    }
-
-    if ground_truth is not None:
-        # Metrics comparing to ground truth
-        metrics['mean absolute error'] = sum([abs(x - ground_truth) for x in toe])/len(toe)
-        metrics['mean absolute percentage error'] = metrics['mean absolute error']/ ground_truth
-        metrics['relative accuracy'] = 1 - abs(ground_truth - metrics['mean'])/ground_truth
-        metrics['ground truth percentile'] = stats.percentileofscore(toe, ground_truth)
-
-    return metrics
-
-def prob_success(toe, time):
-    """Calculate probability of success - i.e., probability that event will not occur within a given time (i.e., success)
-
-    Args:
-        toe ([float]): Times of event for a single event, output from predictor
-        time ([type]): time for calculation
-
-    Returns:
-        fload: Probability of success
-    """
-    return sum([e > time for e in toe])/len(toe)
-eol_metrics = toe_metrics  # For backwards compatability
-
-def alpha_lambda(times, toes, ground_truth, lambda_value, alpha, beta): 
-    """Compute alpha lambda metrics
-
-    Args:
-        times ([float]): Times corresponding to toes (output from predictors)
-        toes ([float]): Times of event for a single event, output from predictor
-        ground_truth (float): Ground Truth time of event for that event
-        lambda_value (float): lambda
-        alpha (float): alpha
-        beta (float): beta
-
-    Returns:
-        bool: if alpha-lambda met
-    """
-    for (t, toe) in zip(times, toes):
-        if (t >= lambda_value):
-            upper_bound = ground_truth + alpha*(ground_truth-t)
-            lower_bound = ground_truth - alpha*(ground_truth-t)
-            return percentage_in_bounds(toe, [lower_bound, upper_bound]) >= beta 
+# This file is kept for backwards compatability
+from .uncertain_data_metrics import calc_metrics as eol_metrics
+from .toe_metrics import prob_success
+from .toe_profile_metrics import alpha_lambda
 
+from numpy import mean
+from warnings import warn
 
 def mean_square_error(values, ground_truth):
     """Mean Square Error
-
     Args:
         values ([float]): time of event for a single event, output from predictor
         ground_truth (float): Ground truth ToE
-
     Returns:
         float: mean square error of toe predictions
     """
     return sum([(mean(x) - ground_truth)**2 for x in values])/len(values)
 
-def toe_profile_metrics(toe, ground_truth):
-    """Calculate toe profile metrics
-
-    Args:
-        toe ([float]): Times of event for a single event, output from predictor
-        ground_truth (float): Ground truth toe
-
-    Returns:
-        dict: toe Profile Metrics
-    """
-    return {
-        'mean square error': mean_square_error(toe, ground_truth)
-    }
-
 def percentage_in_bounds(toe, bounds):
     """Calculate percentage of ToE dist is within specified bounds
 
@@ -115,4 +28,5 @@ def percentage_in_bounds(toe, bounds):
     Returns:
         float: Percentage within bounds (where 1 = 100%)
     """
-    return sum([x < bounds[1] and x > bounds[0] for x in toe])/ len(toe)
+    warn('percentage_in_bounds has been deprecated in favor of UncertainData.percentage_in_bounds(bounds). This function will be removed in a future release')
+    return sum([x < bounds[1] and x > bounds[0] for x in toe])/ len(toe)

src/prog_algs/metrics/toe_metrics.py

@@ -0,0 +1,60 @@
+# Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration. All Rights Reserved.
+
+"""
+This file includes functions for calculating metrics specific to time of event (ToE) from a single event or multiple events given the same time of prediction
+"""
+
+from typing import Iterable
+from numpy import isscalar
+from ..uncertain_data import UncertainData, UnweightedSamples
+
+def prob_success(toe, time, **kwargs):
+    """Calculate probability of success - i.e., probability that event will not occur within a given time (i.e., success)
+
+    Args:
+        toe (UncertainData or array[float]): Times of event for a single event (array[float]) or multiple events, output from predictor
+        time (float): time for calculation
+        **kwargs (optional): Configuration parameters. Supported parameters include:
+          * n_samples (int): Number of samples to use for calculating metrics (if toe is not UnweightedSamples). Defaults to 10,000.
+          * keys (list of strings, optional): Keys to calculate metrics for. Defaults to all keys.
+
+    Returns:
+        float: Probability of success
+    """
+    params = {
+        'n_samples': 10000,  # Default 
+    }
+    params.update(kwargs)
+
+    if isinstance(toe, UncertainData):
+        # Default to all keys
+        keys = params.setdefault('keys', toe.keys())
+
+        if isinstance(toe, UnweightedSamples):
+            samples = toe
+        else:
+            # Some other distribution besides unweighted samples
+            # Generate Samples
+            samples = toe.sample(params['n_samples'])
+
+        # If unweighted_samples, calculate metrics for each key
+        return {key: prob_success(samples.key(key), 
+                time, 
+                **kwargs) for key in keys}
+    elif isinstance(toe, Iterable):
+        if len(toe) == 0:
+            raise ValueError('Time of Event must not be empty')
+        # Is list or array
+        if isscalar(toe[0]):
+            # list of numbers - this is the case that we can calculate
+            pass
+        elif isinstance(toe[0], dict):
+            # list of dicts - Supported for backwards compatabilities
+            toe = UnweightedSamples(toe)
+            return prob_success(toe, time, **kwargs)
+        else:
+            raise TypeError("ToE must be type Uncertain Data or array of dicts, was {}".format(type(toe)))
+    else:
+        raise TypeError("ToE must be type Uncertain Data or array of dicts, was {}".format(type(toe)))
+
+    return sum([e > time for e in toe])/len(toe)

src/prog_algs/metrics/toe_profile_metrics.py

@@ -0,0 +1,43 @@
+# Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration. All Rights Reserved.
+
+"""
+This file includes functions for calculating metrics given a time of event (ToE) profile (i.e., ToE's calculated at different times of prediction resulting from running prognostics multiple times, e.g., on playback data). The metrics calculated here are specific to multiple ToE estimates (e.g. alpha-lambda metric)
+"""
+from ..predictors import ToEPredictionProfile
+
+def alpha_lambda(toe_profile : ToEPredictionProfile, ground_truth : float, lambda_value : float, alpha : float, beta : float, **kwargs): 
+    """
+    Compute alpha lambda metric, a common metric in prognostics. Alpha-Lambda is met if alpha % of the Time to Event (TtE) distribution is within beta % of the ground truth at prediction time lambda.
+
+    Args:
+        toe_profile (ToEPredictionProfile): A profile of predictions, the combination of multiple predictions
+        ground_truth (float): Ground Truth time of event for that event
+        lambda_value (float): Prediction time at or after which metric is evaluated. Evaluation occurs at this time (if a prediction exists) or the next prediction following.
+        alpha (float): percentage bounds around time to event (where 0.2 allows 20% error TtE)
+        beta (float): portion of prediction that must be within those bounds
+        kwargs (optional): configuration arguments. Accepted arge include:
+            * keys (list[string], optional): list of keys to use. If not provided, all keys are used.
+
+    Returns:
+        bool: if alpha-lambda met
+    """
+    params = {
+        'print': False
+    }
+    params.update(kwargs)
+
+    for (t_prediction, toe) in toe_profile.items():
+        if (t_prediction >= lambda_value):
+            # If keys not provided, use all
+            keys = params.setdefault('keys', toe.keys())
+
+            result = {}
+            for key in keys:
+                upper_bound = ground_truth[key] + alpha*(ground_truth[key]-t_prediction)
+                lower_bound = ground_truth[key] - alpha*(ground_truth[key]-t_prediction)
+                result[key] = toe.percentage_in_bounds([lower_bound, upper_bound])[key] >= beta 
+                if params['print']:
+                    print('\n', key)
+                    print('\ttoe:', toe.key(key))
+                    print('\tBounds: [{} - {}]({}%)'.format(lower_bound, upper_bound, toe.percentage_in_bounds([lower_bound, upper_bound])[key]))
+            return result

src/prog_algs/metrics/uncertain_data_metrics.py

@@ -0,0 +1,104 @@
+# Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration. All Rights Reserved.
+
+"""
+This file includes functions for calculating general metrics (i.e. mean, std, percentiles, etc.) on any distribution of type UncertainData (e.g. states, event_states, an EOL distribution, etc.)
+"""
+from typing import Iterable
+from numpy import isscalar, mean, std, array
+from scipy import stats
+from ..uncertain_data import UncertainData, UnweightedSamples
+
+def calc_metrics(data, ground_truth = None, **kwargs):
+    """Calculate all time of event metrics
+
+    Args:
+        data (array[float] or UncertainData): data from a single event
+        ground_truth (float, optional): Ground truth value. Defaults to None.
+        **kwargs (optional): Configuration parameters. Supported parameters include:
+          * n_samples (int): Number of samples to use for calculating metrics (if data is not UnweightedSamples). Defaults to 10,000.
+          * keys (list of strings, optional): Keys to calculate metrics for. Defaults to all keys.
+
+    Returns:
+        dict: collection of metrics
+    """
+    params = {
+        'n_samples': 10000,  # Default is enough to get every percentile
+    }
+    params.update(kwargs)
+
+    if isinstance(data, UncertainData):
+        # Default to all keys
+        keys = params.setdefault('keys', data.keys())
+
+        if ground_truth and isscalar(ground_truth):
+            # If ground truth is scalar, create dict (expected below)
+            ground_truth = {key: ground_truth for key in keys}
+
+        if isinstance(data, UnweightedSamples):
+            samples = data
+        else:
+            # Some other distribution besides unweighted samples
+            # Generate Samples
+            samples = data.sample(params['n_samples'])
+
+        # If unweighted_samples, calculate metrics for each key
+        result = {key: calc_metrics(samples.key(key), 
+                ground_truth if not ground_truth else ground_truth[key],  # If ground_truth is a dict, use key
+                **kwargs) for key in keys}
+
+        # Set values specific to distribution
+        for key in keys:
+            result[key]['mean'] = data.mean[key]
+            result[key]['median'] = data.median[key]
+            result[key]['percentiles']['50'] = data.median[key]
+
+        return result
+    elif isinstance(data, Iterable):
+        if len(data) == 0:
+            raise ValueError('Data must not be empty')
+        # Is list or array
+        if isscalar(data[0]):
+            # list of numbers - this is the case that we can calculate
+            pass
+        elif isinstance(data[0], dict):
+            # list of dicts - Supported for backwards compatabilities
+            data = UnweightedSamples(data)
+            return calc_metrics(data, ground_truth, **kwargs)
+        else:
+            raise TypeError("Data must be type Uncertain Data or array of dicts, was {}".format(type(data)))
+    else:
+        raise TypeError("Data must be type Uncertain Data or array of dicts, was {}".format(type(data)))
+
+    # If we get here then Data is a list of numbers- calculate metrics for numbers
+    data = array(data)  # Must be array
+    data.sort()
+    m = mean(data)
+    median = data[int(len(data)/2)]
+    metrics = {
+        'min': data[0],
+        'percentiles': {
+            '0.01': data[int(len(data)/10000)] if len(data) >= 10000 else None,
+            '0.1': data[int(len(data)/1000)] if len(data) >= 1000 else None,
+            '1': data[int(len(data)/100)] if len(data) >= 100 else None,
+            '10': data[int(len(data)/10)] if len(data) >= 10 else None,
+            '25': data[int(len(data)/4)] if len(data) >= 4 else None,
+            '50': median,
+            '75': data[int(3*len(data)/4)] if len(data) >= 4 else None,
+        },
+        'median': median,
+        'mean': m,
+        'std': std(data),
+        'max': data[-1],
+        'median absolute deviation': sum([abs(x - median) for x in data])/len(data),
+        'mean absolute deviation':   sum([abs(x - m)   for x in data])/len(data),
+        'number of samples': len(data)
+    }
+
+    if ground_truth is not None:
+        # Metrics comparing to ground truth
+        metrics['mean absolute error'] = sum([abs(x - ground_truth) for x in data])/len(data)
+        metrics['mean absolute percentage error'] = metrics['mean absolute error']/ ground_truth
+        metrics['relative accuracy'] = 1 - abs(ground_truth - metrics['mean'])/ground_truth
+        metrics['ground truth percentile'] = stats.percentileofscore(data, ground_truth)
+
+    return metrics

src/prog_algs/predictors/toe_prediction_profile.py

@@ -28,3 +28,7 @@ def keys(self):
 
     def values(self):
         return [self[k] for k in self.keys()]
+
+    def alpha_lambda(self, ground_truth : float, lambda_value : float, alpha : float, beta : float, **kwargs):
+        from ..metrics import alpha_lambda
+        return alpha_lambda(self, ground_truth, lambda_value, alpha, beta, **kwargs)

src/prog_algs/uncertain_data/__init__.py

@@ -1,6 +1,5 @@
 # Copyright © 2021 United States Government as represented by the Administrator of the National Aeronautics and Space Administration. All Rights Reserved.
 
-
 from .uncertain_data import UncertainData
 from .unweighted_samples import UnweightedSamples
 from .scalar_data import ScalarData

src/prog_algs/uncertain_data/scalar_data.py

@@ -39,3 +39,10 @@ def sample(self, num_samples = 1):
 
     def __str__(self):
         return 'ScalarData({})'.format(self.__state)
+
+    def percentage_in_bounds(self, bounds):
+        if isinstance(bounds, list):
+            bounds = {key: bounds for key in self.keys()}
+        if not isinstance(bounds, dict) and all([isinstance(b, list) for b in bounds]):
+            raise TypeError("Bounds must be list [lower, upper] or dict (key: [lower, upper]), was {}".format(type(bounds)))
+        return {key: (1 if bounds[key][0] < x and bounds[key][1] > x else 0) for (key, x) in self.__state.items()}