T.py

"""Data structures for the algorithm."""
from typing import Callable, NamedTuple, Union, List
from math import inf

from evrouting.T import SoC, Wh, ChargingCoefficient, Time, Node


class SoCProfile:
    """
    The SoC Profile maps an initial SoC to the SoC after
    traversing a path.

    SoC profile of a path is fully described with two parameters:
        - cost: Cost of going from u to v.
        - out: Maximal SoC after passing the path from u to v.
    """

    def __init__(self, path_cost: Wh, capacity: SoC):
        """
        Calculates the maximum SoC after traversing the path, which is
        (for non-negative path costs) U - cost.

        :param path_cost: Cost of the path. The SoC Profile of
            a single vertex has no cost.
        :param capacity: The battery's capacity.
        """
        self.capacity: SoC = capacity
        self.path_cost: Wh = path_cost
        self.out: SoC = capacity - self.path_cost

    def __call__(self, initial_soc: SoC) -> SoC:
        """
        :param initial_soc: The initial SoC.
        :returns: The SoC after traversing the path.
        """
        if initial_soc < self.path_cost:
            return -inf
        final_soc = initial_soc - self.path_cost

        return final_soc if final_soc < self.capacity else self.capacity

    def __add__(self, other: 'SoCProfile') -> 'SoCProfile':
        """
        Combines to SoC Profiles.

        The for the combined SoC Profile, given a SoC Profile for a path a
        and a path b, holds:

            cost = cost_a + cost_b
            out = U - cost

        :return: Combined SoC Profile.
        """
        return SoCProfile(self.path_cost + other.path_cost, self.capacity)


class ChargingFunction:
    """
    Charging functions map charging time to resulting SoCs. Since they are
    monotonic, there exists also an inverse mapping SoC to charging time.

    Dummy Charging Station
    ======================

    Special case is a so called dummy charging station. This is the case
    if the charging coefficient is 0. Then there is no actual charging and
    the value of the charging station is always the same as the arrival SoC.

    A dummy charging station is necessary for initialization of the problem
    to trigger spawning of new labels.
    """

    def __init__(self, c: ChargingCoefficient, capacity: SoC,
                 initial_soc: SoC = None):
        """

        :param c: The stations charging coefficient. If c=0, the this becomes
            a so called dummy charging station, that exists to trigger label
            creation in the algorithm.
        :param capacity: The battery's capacity, respectively the maximum of
            the charging function.
        :param initial_soc: The SoC at arriving at the charging station. This is
            optional for dummy charging stations, that do not actually charge:

                cf(b) = soc

        """
        self.c: ChargingCoefficient = c
        self.capacity: SoC = capacity
        self.initial_soc: SoC = initial_soc

    @property
    def is_dummy(self) -> bool:
        """Tell if function is a dummy function."""
        return self.c == 0

    def __call__(self, t: Time, initial_soc: SoC = 0) -> SoC:
        """
        :param t: Charging time
        :param initial_soc: Initial SoC when starting to charge
        :return: SoC after charging.
        """
        if self.is_dummy:
            if self.initial_soc is None:
                raise ValueError(
                    'Charging coefficient is 0 but no initial SoC given.'
                )
            soc = self.initial_soc
        else:
            soc = initial_soc + self.c * t

        return soc if soc < self.capacity else self.capacity

    @property
    def inverse(self) -> Callable[[SoC], Time]:
        """
        :returns: Inverse charging function mapping SoC to charging
            time (assuming initial SoC=0)
        """
        if self.is_dummy:
            raise ValueError('Dummy Carging function has no inverse.')

        c = self.c
        capacity = self.capacity

        def cf_inverse(soc: SoC) -> Time:
            if soc > capacity:
                # Return max charge time
                return capacity / c
            if soc < 0:
                # Return min charge time
                return 0

            return soc / c

        return cf_inverse


class Label(NamedTuple):
    """
    Label used for the Algorithm.

    The (paleto optimal) label where u=t is the
    contains the minimum trip time necessary to get to t.

    The label of node v consists of a tuple:

        (t_trip, soc_last_cs, last_cs, soc_profile_cs_v)

    Where:

        :t_trip: Trip time to node v without charging time at the last
            charging station.
        :soc_last_cs: Arrival SoC at the last charging station.
        :last_cs: (Node ID of) The last charging station.
        :soc_profile_cs_v: The SoC Profile describing the costs going from
            the charging station to current node v.
    """
    t_trip: Time
    soc_last_cs: SoC
    last_cs: Node
    soc_profile_cs_v: SoCProfile


class Breakpoint(NamedTuple):
    """Breakpoint describing a SoC Function."""
    t: Time
    soc: SoC


class SoCFunction:
    """
    SoC Function of a node's label maps trip time plus an additional charging
    time at the last charging station to a SoC at the node.

    """

    def __init__(self, label: Label, cf_cs: ChargingFunction):
        """
        :param label: Label containing to a node. It has the fields:

            :t_trip: Trip time. Includes time to reach v from the last
                charging station in the path but excludes charging at u.
            :soc_last_cs: The SoC at reaching the last charging station
                before any charging.
            :last_cs: The last charging station in the path to current node.
            :soc_profile_cs_v: The SoC Profile of the path cs -> v.
            :cf_cs: Charging function of cs.

        :param cf_cs: The charging function of the last charging station.
        """
        # Unpack label
        self.t_trip: Time = label.t_trip
        self.last_cs: Node = label.last_cs
        self.soc_last_cs: SoC = label.soc_last_cs
        self.soc_profile_cs_v: SoCProfile = label.soc_profile_cs_v

        self.cf_cs: ChargingFunction = cf_cs

        self.minimum = self._calc_minimum()
        self.breakpoints = self._calc_breakpoints()

    def _calc_breakpoints(self) -> List[Breakpoint]:
        """
        Since all charging functions are linear functions, every SoC Function
        can be represented using not more than two breakpoints.

        If the last charging station is a dummy station (that does not change
        the battery's SoC), the only breakpoint is at the minimum trip time
        where the battery's SoC will be the SoC at the last charging station
        decreased by the costs of the path.

        For regular charging stations holds, that the SoC at the minimum
        feasible trip time to a node means, that the car will arrive with
        zero capacity left. The maximum SoC at the arriving node, which does
        not change after even longer trip times and thereby defines the other
        breakpoint, is when the battery has been fully charged at the last
        station.

        Todo: Think about alternative ways of calculation, to minimize function
            calls.
        """
        if self.cf_cs.is_dummy:
            breakpoints = [Breakpoint(self.minimum, self(self.minimum))]
        else:
            breakpoints = [
                Breakpoint(self.minimum, 0),
                Breakpoint(
                    self.minimum + self.cf_cs.inverse(self.soc_profile_cs_v.out),
                    self.soc_profile_cs_v.out
                )
            ]
        return breakpoints

    def __call__(self, t: Time) -> SoC:
        """
        Maps a new trip time to a SoC at the current node. The new trip time
        then includes and fixes charging time at the last charging station.

        :param t: t = trip time + charging at the last charging station
        :return: SoC at current node at time t after spending t - t_trip at
            the last charging station to charge the battery.

        """
        if t < self.t_trip:
            # Impossible to reach the current node for times < trip time
            return -inf

        return self.soc_profile_cs_v(
            self.cf_cs(t - self.t_trip, self.soc_last_cs)
        )

    def calc_optimal_t_charge(self, cs: ChargingFunction) -> Union[Time, None]:
        """
        Calculates the optimal time to charge at the last charging station
        given a charging function ```cs``` of another (the current) charging
        station.

        Given the fact, that all charging stations have linear charging
        functions we have to seperate two cases:

        1. The current charging stations performs better than the last one.

        In this case, it is optimal to charge at the last stop
        only the amount necessary to reach the current charging station which
        is the minimum of the SoC Function::

            t_charge = t_min(f)

        2. Because of battery constraints, the battery's maximum capacity
        at this station from charging at the previous station is given by the
        battery's maximum capacity decreased by the costs of the path to
        the current station.

        Therefore (if the path cost is > 0), it is beneficial to charge
        at the current charging station even if it charges slower than the
        previous one.

        The optimal procedure then is to charge the battery up to the maximum
        at the previous station and continue charging at the current
        station when the SoC Function reaches its maximum. In breakpoint
        representation the maximum is given by the last breakpoint of the
        piecewise linear SoC Function.

        ..Note:

            The possibility not to charge is also kept in the algorithm
            since the according label is not thrown away.


        :param cs: Charging function of the current charging station.
        :return: None if it is optimal not to charge else the charging time.
        """
        capacity: SoC = self.soc_profile_cs_v.capacity

        t_charge = None

        if cs.c > self.cf_cs.c:
            # Faster charging station -> charge as soon as possible
            t_charge = self.breakpoints[0].t - self.t_trip
        elif self.breakpoints[-1].soc < capacity:
            # Slower charging station might still be dominating
            # because the soc cannot be more than the full capacity
            # decreased by the trip costs. This will be refilled at this station.
            t_charge = self.breakpoints[-1].t - self.t_trip

        return t_charge

    def dominates(self, other: 'SoCFunction') -> bool:
        """
        Check if self dominates other function by evaluating all breakpoints.

        f dominates f' if f(t) >= f'(t) for all t

        """
        for t_i, soc_i in self.breakpoints:
            if other(t_i) > soc_i:
                return False

        for t_i, soc_i in other.breakpoints:
            if soc_i > self(t_i):
                return False

        return True

    def _calc_minimum(self) -> Time:
        """
        :returns: Least feasible trip time. That is, the minimum time when
            the SoC functions yields a SoC value that is not -inf.

            This is either the trip time, or if energy needs to be charged
            at the previous charging station to traverse the path, trip time
            plus charging time until the battery holds the minimum energy to
            traverse the path to the current node (which is the cost of the
            path). This time is:

            t = t_trip + cf_cs^(-1)(cost_cs_v) - cf_cs^(-1)(soc_last_sc)

            Thus, for t_min holds:

            t_min = max(t_trip, t)

        """
        cost_p = self.soc_profile_cs_v.path_cost

        if cost_p > self.soc_last_cs:
            if self.cf_cs.is_dummy:
                # Not feasible. Dummy stations do not load.
                return -inf

            return self.t_trip + \
                   self.cf_cs.inverse(cost_p) - \
                   self.cf_cs.inverse(self.soc_last_cs)

        return self.t_trip