Kneeliverse

https://mariolpantunes.github.io/knee/
Admin
A universal knee/elbow detection library for performance curves that implements multiple algorithms...
Tokens:83,110
Snippets:458
Trust Score:9.7
Update:4 months ago
Show doc for...
Tokens:
Raw
### kneeliverse.evaluation module

Source: https://mariolpantunes.github.io/knee/kneeliverse/evaluation.html

This code snippet shows the initial setup and metadata for the kneeliverse.evaluation module, including author, version, email, status, license, and copyright information. It also imports necessary libraries like enum, math, logging, numpy, and custom modules for linear fitting and metrics.

```python
1# coding: utf-8
  2
  3'''
  4The following module provides a set of methods
  5used for evaluation knee detection algorithms.
  6'''
  7
  8__author__ = 'Mário Antunes'
  9__version__ = '1.0'
 10__email__ = 'mario.antunes@ua.pt'
 11__status__ = 'Development'
 12__license__ = 'MIT'
 13__copyright__ = '''
 14Copyright (c) 2021-2023 Stony Brook University
 15Copyright (c) 2021-2023 The Research Foundation of SUNY
 16'''
 17
 18import enum
 19import math
 20import logging
 21import numpy as np
 22import kneeliverse.linear_fit as lf
 23import kneeliverse.metrics as metrics
 24
 25
 26logger = logging.getLogger(__name__)
 27
```

--------------------------------

### Shortest Distance from Points to a Line

Source: https://mariolpantunes.github.io/knee/kneeliverse/linear_fit.html

Computes the shortest distance from a set of points to a straight line defined by two points (start and end). It handles cases where the start and end points are the same.

```python
def shortest_distance_points(p: np.ndarray, a: np.ndarray, b: np.ndarray):
    """
    Computes the shortest distance from the points to the 
    straight line defined by the left and right point.

    Args:
        pt (np.ndarray): numpy array with the points (x, y)
        start (np.ndarray): the left point
        end (np.ndarray): the right point

    Returns:
        np.ndarray: the perpendicular distances
    """

    # TODO for you: consider implementing @Eskapp's suggestions
    if np.all(a == b):
        return np.linalg.norm(p - a, axis=1)

    # normalized tangent vector
    d = np.divide(b - a, np.linalg.norm(b - a))

    # signed parallel distance components
    s = np.dot(a - p, d)
    t = np.dot(p - b, d)

    # clamped parallel distance
    h = np.maximum.reduce([s, t, np.zeros(len(p))])

    # perpendicular distance component, as before
    # note that for the 3D case these will be vectors
    c = np.cross(p - a, d)

    # use hypot for Pythagoras to improve accuracy
    return np.hypot(h, c)

```

--------------------------------

### kneeliverse.lmethod

Source: https://mariolpantunes.github.io/knee/kneeliverse/lmethod.html

The following module provides knee detection method based on L-method algorithm.

```python
# coding: utf-8

'''
The following module provides knee detection method
based on L-method algorithm.
'''

__author__ = 'Mário Antunes'
__version__ = '1.0'
__email__ = 'mario.antunes@ua.pt'
__status__ = 'Development'
__license__ = 'MIT'
__copyright__ = '''
Copyright (c) 2021-2023 Stony Brook University
Copyright (c) 2021-2023 The Research Foundation of SUNY
'''

import math
import logging
import numpy as np
import kneeliverse.linear_fit as lf 
import kneeliverse.multi_knee as mk


from enum import Enum


logger = logging.getLogger(__name__)


class Fit(Enum):
    """
    Enum that defines the types of linear fitting
    """
    best_fit = 'bestfit'
    point_fit = 'pointfit'

    def __str__(self):
        return self.value


class Cost(Enum):
    """
    Enum that defines the cost used in L-method
    
    RMSE was originaly used by the original authors.
    RSS can be used to speedup the computation 
    (it has similar results in most practical cases).
    """
    rss = 'rss'
    rmse = 'rmse'

    def __str__(self):
        return self.value


class Refinement(Enum):
    """
    Enum that defines the iterative refinement used in L-method

    none as the name implies, it avoids the iterative refinement
    original apply the iterative refinement from the paper
    adjusted applies a variant that should be stable on noisy data
    """
    none='none'
    original='original'
    adjusted='adjusted'

    def __str__(self):
        return self.value


def compute_error(x: np.ndarray, y: np.ndarray, index:int, length:int, fit=Fit.point_fit, cost=Cost.rmse):
    """
    Returns the fitting error that is minimized by the L-method algorithm.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        y (np.ndarray): the value of the points in the y axis coordinates
        index (int): the index where the fitting line is divided
        lenght (int): the lenght of the points considered
        fit (Fit): select between point fit and best fit
        cost (Cost): use either RMSE (original work) or RSS (faster implementation)
    
    Returns:
        float: the fitting cost
    """
    
    left_length = x[index] - x[0]
    right_length = x[-1] - x[index]

    left_ratio = left_length/length
    right_ratio = right_length/length

    if fit is Fit.best_fit:
        coef_left, r_left, *_  = np.polyfit(x[0:index+1], y[0:index+1], 1, full=True)
        coef_right, r_rigth, *_ = np.polyfit(x[index:], y[index:], 1, full=True)
        #error = r_left[0]*(left_length/length) + r_rigth[0]*(right_length/length)
        r_left = r_left[0]
        r_rigth = r_rigth[0]
    else:
        coef_left = lf.linear_fit(x[0:index+1], y[0:index+1])
        coef_right = lf.linear_fit(x[index:], y[index:])
        r_left = lf.linear_residuals(x[0:index+1], y[0:index+1], coef_left)
        r_rigth = lf.linear_residuals(x[index:], y[index:], coef_right)
        
    if cost is Cost.rmse:
        error = left_ratio*math.sqrt(r_left*left_ratio) + right_ratio*math.sqrt(right_ratio*r_rigth)
    else:
        error = r_left*left_ratio + r_rigth*right_ratio
    
    return error, coef_left, coef_right


def get_knee(x: np.ndarray, y: np.ndarray, fit=Fit.point_fit, cost=Cost.rmse) -> int:
    """
    Returns the index of the knee point based on the L-method.

    This method does not use the iterative refinement.
    It represents a single iteration of the refinement technique.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        y (np.ndarray): the value of the points in the y axis coordinates
        fit (Fit): select between point fit and best fit

    Returns:
        int: the index of the knee point
    """

    index = 2
    length = x[-1] - x[0]
    error, coef_left, coef_right = compute_error(x, y, index, length, fit, cost)
    #logger.info("Error(%s) = %s", index, error)

    for i in range(index+1, len(x)-2):
        current_error, i_coef_left, i_coef_right = compute_error(x, y, i, length, fit, cost)
        #logger.info("Error(%s) = %s", i, error)

        if current_error < error:
            error = current_error
            index = i
            coef_left = i_coef_left
            coef_right = i_coef_right

    return (index, coef_left, coef_right)


def knee(points:np.ndarray, fit:Fit=Fit.point_fit, it:Refinement=Refinement.adjusted, limit:int=10) -> int:
    """

```

--------------------------------

### get_neighbourhood_points

Source: https://mariolpantunes.github.io/knee/kneeliverse/evaluation.html

Get the neighbourhood (closest points) from a to b. The neighbourhood is defined as the longest straight line (defined by R^2).

```python
def get_neighbourhood_points(points: np.ndarray, a: int, b: int, t: float) -> tuple:
    """
    Get the neighbourhood (closest points) from a to b.

    The neighbourhood is defined as the longest straitgh line (defined by $R^2$).

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        a (int): the initial point of the search
        b (int): the left limit of the search
        t (float): $R^2$ threshold

    Returns:
        tuple: (neighbourhood index, r2, slope)
    """

    x = points[:, 0]
    y = points[:, 1]
    return get_neighbourhood(x, y, a, b, t)
```

--------------------------------

### get_neighbourhood_binary

Source: https://mariolpantunes.github.io/knee/kneeliverse/evaluation.html

Get the index of the point within the range [b, a] where the R^2 is close to the threshold. This version uses an inaccurate binary search to speed up the search.

```python
def get_neighbourhood_binary(x: np.ndarray, y: np.ndarray, a: int, b: int, t=0.9) -> int:
    """
    Get the index of the point within the range $[b, a]$ where the $R^2$ is close to the threshold.

    This version uses a inaccurate binary search to speedup the search.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        y (np.ndarray): the value of the points in the y axis coordinates
        a (int): the initial point of the search
        b (int): the left limit of the search
        t (float): $R^2$ threshold (default 0.9)

    Returns:
        int: index of the point
    """

    i = b
    right = a

    while abs(i-right) > 1:
        coef = lf.linear_fit(x[i:a+1], y[i:a+1])
        r2 = lf.linear_r2(x[i:a+1], y[i:a+1], coef)

        if r2 < t:
            i = int((i+right)/2.0)
        else:
            right = i
            i = int((b+right)/2.0)

    return i
```

--------------------------------

### Kneedle Module

Source: https://mariolpantunes.github.io/knee/kneeliverse/kneedle.html

This is the main module for the Kneedle algorithm, including enums for direction, concavity, and peak detection, as well as functions for calculating differences and identifying knee points.

```python
# coding: utf-8

'''
The following module provides knee detection method
based on Kneedle algorithm.
'''

__author__ = 'Mário Antunes'
__version__ = '1.0'
__email__ = 'mario.antunes@ua.pt'
__status__ = 'Development'
__license__ = 'MIT'
__copyright__ = '''
Copyright (c) 2021-2023 Stony Brook University
Copyright (c) 2021-2023 The Research Foundation of SUNY
'''

import math
import enum
import logging
import numpy as np
import uts.ema as ema
import uts.peak_detection as pd
import kneeliverse.multi_knee as mk
import kneeliverse.linear_fit as lf


logger = logging.getLogger(__name__)


class Direction(enum.Enum):
    """
    Enum data type that represents the direction of a concavity.
    """
    Increasing = 'increasing'
    Decreasing = 'decreasing'

    def __str__(self):
        return self.value


class Concavity(enum.Enum):
    """
    Enum data type that represents the rotation of a concavity.
    """
    Counterclockwise = 'counter-clockwise'
    Clockwise = 'clockwise'

    def __str__(self):
        return self.value


class PeakDetection(enum.Enum):
    """
    Enum data type that identifies the peak selection algorithm.
    """
    Kneedle = 'Kneedle'
    ZScore = 'ZScore'
    Significant = 'Significant'
    All = 'All'

    def __str__(self):
        return self.value


def differences(points: np.ndarray, cd: Direction, cc: Concavity) -> np.ndarray:
    """
    Computes the differences from the y axis.

    These differences represent a rotation within the original algorithm.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        cd (Direction): direction of the concavity
        cc (Concavity): rotation of the concavity

    Returns:
        np.ndarray: the points array with the differences
    """

    rv = np.empty(points.shape)

    if cd is Direction.Decreasing and cc is Concavity.Clockwise:
        for i in range(0, len(points)):
            rv[i][0] = points[i][0]
            rv[i][1] = points[i][0] + points[i][1]  # x + y
    elif cd is Direction.Decreasing and cc is Concavity.Counterclockwise:
        for i in range(0, len(points)):
            rv[i][0] = points[i][0]
            rv[i][1] = 1.0 - (points[i][0] + points[i][1])  # 1.0 - (x + y)
    elif cd is Direction.Increasing and cc is Concavity.Clockwise:
        for i in range(0, len(points)):
            rv[i][0] = points[i][0]
            rv[i][1] = points[i][1] - points[i][0]  # y - x
    else:
        for i in range(0, len(points)):
            rv[i][0] = points[i][0]
            rv[i][1] = math.fabs(points[i][1] - points[i][0])  # abs(y - x)

    return rv


def _knee(points: np.ndarray, t: float, cd: Direction, cc: Concavity) -> int:
    """
    Returns the index of the knee point based on the Kneedle method.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        t (float): tau of the side window used to smooth the curve
        cd (Direction): direction of the concavity
        cc (Concavity): rotation of the concavity

    Returns:
        int: the index of the knee point
    """

    Ds = ema.linear(points, t)
    pmin = Ds.min(axis=0)
    pmax = Ds.max(axis=0)
    diff = pmax - pmin
    diff[diff == 0] = 1.0

    Dn = (Ds - pmin)/diff 

    Dd = differences(Dn, cd, cc)

    peaks = pd.all_peaks(Dd)
    
    idx = pd.highest_peak(Dd, peaks)
    if idx == -1:
        return None
    else:
        return idx


def _knees(points: np.ndarray, t: float, cd: Direction, cc: Concavity, sensitivity:float=1.0, p:PeakDetection=PeakDetection.Kneedle, debug:bool=False) -> np.ndarray:
    """
    Returns the index of the knees point based on the Kneedle method.

    This implementation uses an heuristic to automatically define
    the direction and rotation of the concavity.

    Furthermore, it support three different methods to select the 
    relevant knees:
    1. Kneedle    : classical algorithm
    2. Significant: significant knee peak detection
    3. ZScore     : significant knee peak detection based on zscore

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        t (float): tau of the sliding window used to smooth the curve
        cd (Direction): direction of the concavity
        cc (Concavity): rotation of the concavity
        sensitivity (float): controls the sensitivity of the peak detection (default 1.0)

```

--------------------------------

### get_neighbourhood_fast_points

Source: https://mariolpantunes.github.io/knee/kneeliverse/evaluation.html

Get the neighbourhood (closest points) from a to b. The neighbourhood is defined as the longest straight line (defined by R^2). This version uses an inaccurate binary search to speed up the search.

```python
def get_neighbourhood_fast_points(points: np.ndarray, a: int, b: int, t: float) -> tuple:
    """
    Get the neighbourhood (closest points) from a to b.

    The neighbourhood is defined as the longest straitgh line (defined by $R^2$).
    This version uses a inaccurate binary search to speedup the search.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        a (int): the initial point of the search
        b (int): the left limit of the search
        t (float): $R^2$ threshold

    Returns:
        tuple: (neighbourhood index, r2, slope)
    """

    x = points[:, 0]
    y = points[:, 1]
    return get_neighbourhood_fast(x, y, a, b, t)
```

--------------------------------

### min_point_rdp function

Source: https://mariolpantunes.github.io/knee/kneeliverse/rdp.html

This function implements a minimal points RDP algorithm. It iterates through different threshold values to find a reduced set of points that meets a minimum size requirement. If necessary, it falls back to a fixed RDP version to achieve the exact number of points.

```python
def min_point_rdp(points: np.ndarray, t:list=[0.01, 0.001, 0.0001], min_points:int=10) -> tuple:
    """
    Minimal points RDP.

    Given a minimal amount of points, this version runs Global RDP with different threshold 
    values. The thresholds are sorted in decreasing order.
    The method returns as soon as a threshold value returns a minimal amount of points.
    If necessary the method will execute the fixed version of RDP to get the exact amount of points.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        t (float): a list of coefficient of determination thresholds (default [0.01, 0.001, 0.0001])
        min_points (int): the minimal amount of points (default 10)

    Returns:
        tuple: the index of the reduced space, the points that were removed
    """    
    
    # sort the threshold in 
    t.sort(reverse=True)

    for current_t in t:
        reduced, removed = grdp(points, t=current_t)
        if len(reduced) >= min_points:
            return reduced, removed
    
    return rdp_fixed(points, min_points)
```

--------------------------------

### kneeliverse.knee_ranking

Source: https://mariolpantunes.github.io/knee/kneeliverse/knee_ranking.html

The following module provides a set of methods used for ranking knees from multi-knees approaches.

```python
# coding: utf-8

'''
The following module provides a set of methods
used for ranking knees from multi-knees approaches.
'''

__author__ = 'Mário Antunes'
__version__ = '0.1'
__email__ = 'mario.antunes@ua.pt'
__status__ = 'Development'
__license__ = 'MIT'
__copyright__ = '''
Copyright (c) 2021-2023 Stony Brook University
Copyright (c) 2021-2023 The Research Foundation of SUNY
'''

import math
import enum
import logging
import numpy as np
import kneeliverse.linear_fit as lf
import kneeliverse.evaluation as ev


logger = logging.getLogger(__name__)


class ClusterRanking(enum.Enum):
    """
    Enum data type that represents the direction of the ranking within a cluster.
    """
    left = 'left'
    linear = 'linear'
    right = 'right'
    hull = 'hull'

    def __str__(self):
        return self.value


def distances(point:np.ndarray, points:np.ndarray) -> np.ndarray:
    """
    Computes the euclidean distance from a single point to a vector of points.

    Args:
        point (np.ndarray): the point
        points (np.ndarray): the vector of points
    
    Returns:
        np.ndarray: a vector with the distances from point to all the points. 
    """
    return np.sqrt(np.sum(np.power(points - point, 2), axis=1))


def rect_overlap(amin: np.ndarray, amax: np.ndarray, bmin: np.ndarray, bmax: np.ndarray) -> float:
    """
    Computes the percentage of the overlap for two rectangles.

    Args:
        amin (np.ndarray): the low point in rectangle A
        amax (np.ndarray): the high point in rectangle A
        bmin (np.ndarray): the low point in rectangle B
        bmax (np.ndarray): the high point in rectangle B

    Returns:
        float: percentage of the overlap of two rectangles
    """
    #logger.info('%s %s %s %s', amin, amax, bmin, bmax)
    dx = max(0.0, min(amax[0], bmax[0]) - max(amin[0], bmin[0]))
    dy = max(0.0, min(amax[1], bmax[1]) - max(amin[1], bmin[1]))
    #logger.info('dx %s dy %s', dx, dy)
    overlap = dx * dy
    #logger.info('overlap = %s', overlap)
    if overlap > 0.0:
        a = np.abs(amax-amin)
        b = np.abs(bmax-bmin)
        total_area = a[0]*a[1] + b[0]*b[1] - overlap
        #print(f'overlap area = {overlap} total area =  {total_area}')
        return overlap / total_area
    else:
        return 0.0


def rect(p1: np.ndarray, p2: np.ndarray) -> tuple:
    """
    Creates the low and high rectangle coordinates from 2 points.

    Args:
        p1 (np.ndarray): one of the points in the rectangle
        p2 (np.ndarray): one of the points in the rectangle

    Returns:
        tuple: tuple with two points (low and high)
    """
    p1x, p1y = p1
    p2x, p2y = p2
    return np.array([min(p1x, p2x), min(p1y, p2y)]), np.array([max(p1x, p2x), max(p1y, p2y)])


def distance_to_similarity(array: np.ndarray) -> np.ndarray:
    """
    Converts an array of distances into an array of similarities.

    Args:
        array (np.ndarray): array with distances values

    Returns:
        np.ndarray: an array with similarity values
    """
    return max(array) - array


def rank(array: np.ndarray) -> np.ndarray:
    """
    Computes the rank of an array of values.

    Args:
        array (np.ndarray): array with values

    Returns:
        np.ndarray: an array with the ranks of each value
    """
    temp = array.argsort()
    ranks = np.empty_like(temp)
    ranks[temp] = np.arange(len(array))
    return ranks


def slope_ranking(points: np.ndarray, knees: np.ndarray, t: float = 0.8) -> np.ndarray:
    """
    Computes the rank of a set of knees in a curve.

    The ranking is based on the slope of the left of the knee point.
    The left neighbourhood is computed based on the R2 metric.
    
    Args:
        points (np.ndarray): numpy array with the points (x, y)
        knees (np.ndarray): knees indexes
        t (float): the R2 threshold for the neighbourhood (default 0.8)

    Returns:
        np.ndarray: an array with the ranks of each value
    """
    # corner case
    if len(knees) == 1.0:
        rankings = np.array([1.0])
    else:
        rankings = []

        x = points[:, 0]
        y = points[:, 1]

        _, _, slope = ev.get_neighbourhood(x, y, knees[0], 0, t)
        rankings.append(math.fabs(slope))

        for i in range(1, len(knees)):
            _, _, slope = ev.get_neighbourhood(x, y, knees[i], knees[i-1], t)

```

--------------------------------

### get_neighbourhood function

Source: https://mariolpantunes.github.io/knee/kneeliverse/evaluation.html

This function is intended to get the neighbourhood (closest points) from point 'a' to 'b', defining the neighbourhood by the longest straight line based on R^2. The implementation details are not fully shown in the provided snippet.

```python
def get_neighbourhood(x: np.ndarray, y: np.ndarray, a: int, b: int, t: float = 0.9) -> tuple:
    """
    Get the neighbourhood (closest points) from a to b.

    The neighbourhood is defined as the longest straitgh line (defined by $R^2$).

    
```

--------------------------------

### Global RDP (gRDP) Main Loop

Source: https://mariolpantunes.github.io/knee/kneeliverse/rdp.html

This is the core loop for the generalized Ramer-Douglas-Peucker (gRDP) algorithm. It iteratively refines a set of points based on a global cost threshold (`t`) and a chosen cost metric. It uses a stack to manage segments to process and a cache to optimize global cost computations. The loop continues as long as the global cost indicates the curve is 'curved' (i.e., needs further simplification) and there are segments left to process.

```python
def _grdp(points:np.ndarray, t:float, cost:metrics.Metrics, order:Order, 
distance_points:callable, stack:list, reduced:list) -> tuple:
    """
    Main loop of the gRDP version.

    Not intended to be used as a single method.
    It is used internally on grdp and mp_grdp.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        t (float): the coefficient of determination threshold
        cost (metrics.Metrics): the cost method used to evaluate a point set 
        order (Order): the metric used to sort the segments
        distance_points (callable): the distance function
        stack (list): stack used to explore the points space
        reduced (list): set of reduced points

    Returns:
        tuple: the index of the reduced space, stack
    """
    # Setup cache that is used to speedup the global cost computation
    cache = {}

    global_cost = evaluation.compute_global_cost(points, reduced, cost, cache)
    curved = global_cost < t if cost is metrics.Metrics.r2 else global_cost >= t

    ti = 0

    while curved and stack:
        _, left, right = stack.pop()
        pt = points[left:right]
        d = distance_points(pt, pt[0], pt[-1])

        # fix issue when all distances are 0 (perfect fit)
        # changed to EPS for improve robustness
        #if d.sum() < np.finfo(float).eps:
        if np.all((d < np.finfo(float).eps)):

```

--------------------------------

### RDP Module Boilerplate

Source: https://mariolpantunes.github.io/knee/kneeliverse/rdp.html

This snippet includes the module's metadata, imports, and enum definitions for distance and order metrics.

```python
1# coding: utf-8
  2
  3'''
  4The following module provides a set of methods
  5used for curve simplification. It offers several
  6versions of the RDP algorhtm.
  7'''
  8
  9__author__ = 'Mário Antunes'
 10__version__ = '1.0'
 11__email__ = 'mario.antunes@ua.pt'
 12__status__ = 'Development'
 13__license__ = 'MIT'
 14__copyright__ = '''
 15Copyright (c) 2021-2023 Stony Brook University
 16Copyright (c) 2021-2023 The Research Foundation of SUNY
 17'''
 18
 19#import math
 20import enum
 21import logging
 22import numpy as np
 23import kneeliverse.linear_fit as lf
 24import kneeliverse.metrics as metrics
 25import kneeliverse.evaluation as evaluation
 26
 27
 28#import matplotlib.pyplot as plt
 29
 30logger = logging.getLogger(__name__)
 31
 32
 33class Distance(enum.Enum):
 34    """
 35    Enum that defines the distance method for the RDP 
 36    """
 37    perpendicular = 'perpendicular'
 38    shortest = 'shortest'
 39
 40    def __str__(self):
 41        return self.value
 42
 43
 44class Order(enum.Enum):
 45    """
 46    Enum that defines the distance method for the RDP 
 47    """
 48    triangle = 'triangle'
 49    area = 'area'
 50    segment = 'segment'
 51
 52    def __str__(self):
 53        return self.value
```

--------------------------------

### kneeliverse.postprocessing module

Source: https://mariolpantunes.github.io/knee/kneeliverse/postprocessing.html

The kneeliverse.postprocessing module provides methods for post-processing knee candidates to improve their quality.

```python
# coding: utf-8

'''
The following module provides a set of methods
used for post-processing knees. These filter were
designed to improve the quality of the knee candidates.
'''

__author__ = 'Mário Antunes'
__version__ = '1.0'
__email__ = 'mario.antunes@ua.pt'
__status__ = 'Development'
__license__ = 'MIT'
__copyright__ = '''
Copyright (c) 2021-2023 Stony Brook University
Copyright (c) 2021-2023 The Research Foundation of SUNY
'''

import math
import logging
import numpy as np
import kneeliverse.rdp as rdp
import kneeliverse.linear_fit as lf
import kneeliverse.convex_hull as ch
import kneeliverse.knee_ranking as kr


logger = logging.getLogger(__name__)


def filter_corner_knees(points: np.ndarray, knees: np.ndarray, t:float = .33) -> np.ndarray:
    """
    Filter the left upper corner knees points.

    A left upper knee corner does not provide a significant improvement to be considered.
    The detection method relies on a three point rectangle fitting and overlap.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        knees (np.ndarray): knees indexes
        t (float): overlap treshold (default 0.33)

    Returns:
        np.ndarray: the filtered knees
    """

    filtered_knees = []

    for i in range(len(knees)):
        idx = knees[i]
        if idx-1 >=0 and idx+1 < len(points):
            p0, p1 ,p2 = points[idx-1:idx+2]
            corner0 = np.array([p0[0], p2[1]])
            amin, amax = kr.rect(corner0, p1)
            bmin, bmax = kr.rect(p0, p2)
            p = kr.rect_overlap(amin, amax, bmin, bmax)
            
            if p < t:
                filtered_knees.append(idx)
        else:
            filtered_knees.append(idx)

    return np.array(filtered_knees)


def select_corner_knees(points: np.ndarray, knees: np.ndarray, t:float = .33) -> np.ndarray:
    """
    Detect and keep the left upper corner knees points.

    The detection method relies on a three point rectangle fitting and overlap.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        knees (np.ndarray): knees indexes
        t (float): overlap treshold (default 0.33)

    Returns:
        np.ndarray: the filtered knees
    """

    filtered_knees = []

    for i in range(len(knees)):
        idx = knees[i]
        if idx-1 >=0 and idx+1 < len(points):
            p0, p1 ,p2 = points[idx-1:idx+2]
            corner0 = np.array([p0[0], p2[1]])
            amin, amax = kr.rect(corner0, p1)
            bmin, bmax = kr.rect(p0, p2)
            p = kr.rect_overlap(amin, amax, bmin, bmax)
            
            if p >= t:
                filtered_knees.append(idx)

    return np.array(filtered_knees)


def filter_worst_knees(points: np.ndarray, knees: np.ndarray) -> np.ndarray:
    """
    Filter the worst knees points.

    A worst knee is a knee that is higher (at y axis) than a previous knee.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        knees (np.ndarray): knees indexes

    Returns:
        np.ndarray: the filtered knees
    """
    
    if len(knees) <= 1:
        return knees
    else:
        filtered_knees = []

        filtered_knees.append(knees[0])
        h_min = points[knees[0]][1]

        for i in range(1, len(knees)):
            h = points[knees[i]][1]

            if h <= h_min:
                filtered_knees.append(knees[i])
                h_min = h

        return np.array(filtered_knees)


def filter_clusters(points: np.ndarray, knees: np.ndarray,
clustering: callable, t: float = 0.01,
method: kr.ClusterRanking = kr.ClusterRanking.linear) -> np.ndarray:
    """
    Filter the knee points based on clustering.

    For each cluster a single point is selected based on the ranking.
    The ranking is computed based on the slope and the improvement (on the y axis).

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        knees (np.ndarray): knees indexes
        clustering (callable): the clustering function
        t (float): the threshold for merging (in percentage, default 0.01)
        method (ranking.ClusterRanking): represents the direction of the ranking within a cluster (default ranking.ClusterRanking.linear)

    Returns:
        np.ndarray: the filtered knees
    """
    if method is kr.ClusterRanking.hull:

```

--------------------------------

### MP GRDP (Min Points Global RDP)

Source: https://mariolpantunes.github.io/knee/kneeliverse/rdp.html

Implements the Global RDP algorithm with an additional constraint on the minimum number of points. If the constraint is not met, it falls back to the `rdp_fixed` method.

```python
def mp_grdp(points: np.ndarray, t: float = 0.01, min_points:int = 10, distance: Distance = Distance.shortest,
cost: metrics.Metrics = metrics.Metrics.smape, order:Order=Order.segment) -> tuple:
    """
    MP gRDP (Min Points Global RDP)

    This version computes the gRDP with the given threshold.
    At the end, if the minimum number of points constraint is not satisfied
    it executes the rdp_fixed version.

    This method stores the reduced set and stack of the gRDP, meaning that
    the rdp_fixed method continues from the previous point onwards.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        t (float): the coefficient of determination threshold (default 0.01)
        min_points (int): the minimal amount of points (default 10)
        distance (Distance): the distance metric used to decide the split point (default: Distance.shortest)
        cost (metrics.Metrics): the cost method used to evaluate a point set (default: metrics.Metrics.smape)
        order (Order): the metric used to sort the segments (default: Order.segment)
    
    Returns:
        tuple: the index of the reduced space, the points that were removed
    """

    stack = [(0, 0, len(points))]
    reduced = [0, len(points)-1]
    
    # select the distance metric to be used
    distance_points = None
    if distance is Distance.shortest:
        distance_points = lf.shortest_distance_points
    elif distance is Distance.perpendicular:
        distance_points = lf.perpendicular_distance_points
    else:
        distance_points = lf.shortest_distance_points


```

--------------------------------

### Python Module for Linear Fitting

Source: https://mariolpantunes.github.io/knee/kneeliverse/linear_fit.html

This Python module provides methods for fast linear fitting, used in the L-methods knee algorithm. It includes functions for fitting points, transforming points, and calculating residuals for both horizontal and vertical line fitting.

```python
# coding: utf-8

'''
The following module provides a set of methods
used for fast linear fitting. This methods are
them used on the L-methods knee algorithm.
'''

__author__ = 'Mário Antunes'
__version__ = '1.0'
__email__ = 'mario.antunes@ua.pt'
__status__ = 'Development'
__license__ = 'MIT'
__copyright__ = '''
Copyright (c) 2021-2023 Stony Brook University
Copyright (c) 2021-2023 The Research Foundation of SUNY
'''

import math
import logging
import numpy as np
import kneeliverse.metrics as metrics


from typing import Union


logger = logging.getLogger(__name__)


def linear_fit_points(points: np.ndarray) -> tuple:
    """
    Computes the linear fit for the points.

    This methods approximates the linear fit using only the
    first and last points in a curve.

    Args:
        points (np.ndarray): numpy array with the points (x, y)

    Returns:
        tuple: (b, m)
    """
    x = points[:, 0]
    y = points[:, 1]
    return linear_fit(x, y)


def linear_fit(x: np.ndarray, y: np.ndarray) -> tuple:
    """
    Computes the linear fit for the points.

    This methods approximates the linear fit using only the
    first and last points in a curve.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        y (np.ndarray): the value of the points in the y axis coordinates

    Returns:
        tuple: (b, m)
    """

    d = x[0] - x[-1]
    if d != 0:
        m = (y[0] - y[-1])/(x[0] - x[-1])
        b = y[0] - (m*x[0])
        return (b, m)
    else:
        return (0, 0)


def linear_transform_points(points: np.ndarray, coef: tuple) -> np.ndarray:
    """
    Computes the y values for an x array and the given coefficients.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        coef (tuple): the coefficients from the linear fit

    Returns:
        np.ndarray: the corresponding y values
    """
    x = points[:, 0]
    return linear_transform(x, coef)


def linear_transform(x: np.ndarray, coef: tuple) -> np.ndarray:
    """
    Computes the y values for an x array and the given coefficients.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        coef (tuple): the coefficients from the linear fit

    Returns:
        np.ndarray: the corresponding y values
    """
    b, m = coef
    y_hat = x * m + b
    return y_hat


def linear_hv_residuals_points(points: np.ndarray) -> float:
    """
    Computes the residual error from a straight line fitting.

    This methods tries a horizontal (typical $f(x)=m \times x + b$ equation)
    and a vertinal line (a non typical $f(y) = m \times y + b$ equation).
    It returns the smaller residual errors.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
    
    Returns:
        float: the residuals of the line fitting.
    """
    x = points[:, 0]
    y = points[:, 1]
    return linear_hv_residuals(x,y)


def linear_hv_residuals(x: np.ndarray, y: np.ndarray) -> float:
    """
    Computes the residual error from a straight line fitting.

    This methods tries a horizontal (typical $f(x)=m \times x + b$ equation)
    and a vertinal line (a non typical $f(y) = m \times y + b$ equation).
    It returns the smaller residual errors.

    Args:
        x (np.ndarray): the value of the points in the x axis coordinates
        y (np.ndarray): the value of the points in the y axis coordinates
    
    Returns:
        float: the residuals of the line fitting.
    """
    # try a tipical y = mx + b line
    coef1 = linear_fit(x, y)
    y_residuals = linear_residuals(x, y, coef1)

    # try a non-typical x = my + b line
    coef2 = linear_fit(y, x)
    x_residuals = linear_residuals(y, x, coef2)

    if y_residuals <= x_residuals:
        return y_residuals
    else:
        return x_residuals


def linear_fit_transform_points(points: np.ndarray, vertical:bool=False) -> Union[np.ndarray, tuple]:
    """
    Computes the horizontal or vertical line fitting.

    A horizontal line fitting uses the typical $f(x)=m \times x + b$ equation.
    The vertinal line fitting uses the non typical $f(y) = m \times y + b$ equation.

    Args:
        points (np.ndarray): numpy array with the points (x, y)
        vertical (bool): if True computes the vertical line fitting (default: False)
    

```
Tokens:
Raw