data_extractors.py

#  ___________________________________________________________________________
#
#  Prescient
#  Copyright 2020 National Technology & Engineering Solutions of Sandia, LLC
#  (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S.
#  Government retains certain rights in this software.
#  This software is distributed under the Revised BSD License.
#  ___________________________________________________________________________

from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from .abstract_types import *
    from typing import Iterable, Dict, Sequence, Set, Tuple

from abc import ABC, abstractmethod
from typing import NamedTuple
import math 

from prescient.util.math_utils import round_small_values


class PreQuickstartCache(NamedTuple):
    """ Stores select results before quickstart is activated, used to compare after quickstart is enabled """
    quickstart_generators_off: Set[G]
    total_cost: float
    power_generated: float


class ScedDataExtractor(ABC):
    """Extracts information from operations model instances."""

    @abstractmethod
    def get_sced_duration_minutes(self, sced: OperationsModel) -> int:
        """ Get the number of minutes of operation this sced represents """
        pass

    @abstractmethod
    def get_buses(self, sced: OperationsModel) -> Iterable[B]:
        """Get all buses in the model."""
        pass

    @abstractmethod
    def get_loads(self, sced: OperationsModel) -> Iterable[L]:
        """Get all loads in the model."""
        pass    

    @abstractmethod
    def get_transmission_lines(self, sced: OperationsModel) -> Iterable[L]:
        """Get all transmission lines in the model."""
        pass

    @abstractmethod
    def get_all_storage(self, sced: OperationsModel) -> Iterable[S]:
        """Get all storage in the model."""
        pass

    @abstractmethod
    def get_all_generators(self, sced: OperationsModel) -> Iterable[G]:
        """Get all generators in the model."""
        pass

    @abstractmethod
    def get_thermal_generators(self, sced: OperationsModel) -> Iterable[G]:
        """Get all thermal generators in the model."""
        pass

    @abstractmethod
    def get_virtual_generators(self, sced: OperationsModel) -> Iterable[G]:
        """Get all thermal generators in the model."""
        pass

    @abstractmethod
    def get_nondispatchable_generators(self, sced: OperationsModel) -> Iterable[G]:
        """Get all non-dispatchable generators (renewables) in the model."""
        pass

    @abstractmethod
    def get_thermal_generators_at_bus(self, sced: OperationsModel, b: B) -> Iterable[G]:
        """Get all thermal generators in the model at bus b."""
        pass

    @abstractmethod
    def get_virtual_generators_at_bus(self, sced: OperationsModel, b: B) -> Iterable[G]:
        """Get all non-dispatchable generators (renewables) in the model at bus b."""
        pass

    @abstractmethod
    def get_nondispatchable_generators_at_bus(self, sced: OperationsModel, b: B) -> Iterable[G]:
        """Get all non-dispatchable generators (renewables) in the model at bus b."""
        pass

    @abstractmethod
    def get_quickstart_generators(self, sced) -> Iterable[G]:
        """Get all generators that are eligible to be quickstarted."""
        pass

    @abstractmethod
    def is_generator_on(self, sced: OperationsModel, gen: G) -> bool:
        """Indicate whether the generator is turned on."""
        pass

    @abstractmethod
    def generator_was_on(self, sced: OperationsModel, gen: G) -> bool:
        """Indicate whether the generator was turned on before this sced (at T0)."""
        pass

    @abstractmethod
    def get_fixed_costs(self, sced: OperationsModel) -> float:
        """Get the sum of all fixed costs."""
        pass

    @abstractmethod
    def get_variable_costs(self, sced: OperationsModel) -> float:
        """Get the sum of all variable costs."""
        pass

    @abstractmethod
    def get_power_generated(self, sced: OperationsModel, g: G) -> float:
        """Get the amount of power generated by a specific generator."""
        pass

    @abstractmethod
    def get_power_generated_T0(self, sced: OperationsModel, g: G) -> float:
        """Get the amount of power generated by a specific generator in the timestep before this model."""
        pass

    @abstractmethod
    def get_bus_demand(self, sced: OperationsModel, bus: B) -> float:
        """Get the demand at a specific bus."""
        pass

    @abstractmethod
    def get_load_bus(self, sced: OperationsModel, load: L) -> float:
        """Get the bus for a specific load."""
        pass            

    @abstractmethod
    def get_load_demand(self, sced: OperationsModel, load: L) -> float:
        """Get the demand for a specific load."""
        pass        

    @abstractmethod
    def get_bus_mismatch(self, sced: OperationsModel, bus: B) -> float:
        """Get the absolute value of the bus's load mismatch."""
        pass

    @abstractmethod
    def get_load_mismatch(self, sced: OperationsModel, b: B) -> float:
        """Get the bus's signed load mismatch."""
        pass

    @abstractmethod
    def get_positive_load_mismatch(self, sced: OperationsModel, b: B) -> float:
        """Get the amount by which load exceeds generation, if any."""
        pass

    @abstractmethod
    def get_negative_load_mismatch(self, sced: OperationsModel, b: B) -> float:
        """Get the amount of over-generation, if any, for a specific bus."""
        pass

    @abstractmethod
    def get_max_power_output(self, sced: OperationsModel, g: G) -> float:
        pass

    @abstractmethod
    def get_max_power_available(self, sced: OperationsModel, g: G) -> float:
        """Get the maximum power available from a particular thermal generator."""
        pass

    @abstractmethod
    def get_flow_level(self, sced: OperationsModel, line: L) -> float:
        pass

    @abstractmethod
    def get_storage_input_dispatch_level(self, sced: OperationsModel, storage: S) -> float:
        pass

    @abstractmethod
    def get_storage_output_dispatch_level(self, sced: OperationsModel, storage: S) -> float:
        pass

    @abstractmethod
    def get_storage_soc_dispatch_level(self, sced: OperationsModel, storage: S) -> float:
        pass

    @abstractmethod
    def get_storage_type(self, sced: OperationsModel, storage: S) -> str:
        pass

    @abstractmethod
    def get_generator_cost(self, sced: OperationsModel, g: G) -> float:
        pass

    @abstractmethod
    def get_total_demand(self, sced: OperationsModel) -> float:
        pass

    @abstractmethod
    def get_reserve_requirement(self, sced: OperationsModel) -> float:
        pass

    @abstractmethod
    def get_min_downtime(self, sced: OperationsModel, g: G) -> float:
        pass

    @abstractmethod
    def get_scaled_startup_ramp_limit(self, sced: OperationsModel, g: G) -> float:
        pass

    @abstractmethod
    def get_generator_fuel(self, sced: OperationsModel, g: G) -> str:
        pass

    def get_total_power_generated(self, sced: OperationsModel) -> float:
        return sum(self.get_power_generated(sced, g) for g in self.get_thermal_generators(sced))

    def get_pre_quickstart_data(self, sced: OperationsModel) -> PreQuickstartCache:
        """ Extract select data from an optimization run without quickstart enabled, used to 
            detect changes when the same model is run with quickstart enabled
        """
        quickstart_generators_off = {g for g in self.get_quickstart_generators() 
                                     if not self.is_generator_on(g, 1)}
        total_cost = self.get_total_costs(sced)
        power_generated = self.get_total_power_generated(sced)
        return PreQuickstartCache(quickstart_generators_off, total_cost, power_generated)

    def get_generators_used_as_quickstart(self, 
                                          cache: PreQuickstartCache,
                                          sced: OperationsModel
                                         ) -> Set[G]:
        """ Get generators that are turned on, but would be turned off if quickstart were not enabled """
        return {} if cache is None else {g for g in cache.quickstart_generators_off if self.is_generator_on(sced, g, 1)}

    def get_generator_quickstart_usage(self, 
                                       cache: PreQuickstartCache,
                                       sced: OperationsModel
                                      ) -> Dict[G, bool]:
        """ Get a dictionary with a boolean entry for every quickstart entry, with True indicating the generator 
            was used as a quickstart generator, and False indicating it was not.

            A generator was used as a quickstart generator if it was on with quickstart enabled, but would have
            been off if quickstart was not enabled.
        """
        if cache is None:
            return {g:False for g in self.get_quickstart_generators(sced)}
        else:
            return {g: g in cache.quickstart_generators_off and self.is_generator_on(sced, g, 1)
                    for g in self.get_quickstart_generators(sced)}

    def get_additional_quickstart_power_generated(self, cache: PreQuickstartCache, sced: OperationsModel) -> float:
        """ Get the increase in total power generated when quickstart generators are on """
        return 0 if cache is None else self.get_power_generated(sced) - cache.power_generated

    def get_additional_quickstart_costs(self, 
                                        cache: PreQuickstartCache, 
                                        sced: OperationsModel) -> float:
        return 0 if cache is None else self.get_total_costs(sced) - cache.total_cost

    def get_total_costs(self, sced: OperationsModel) -> float:
        return self.get_fixed_costs(sced) + \
               self.get_variable_costs(sced)

    def get_on_off_and_ramps(self, sced: OperationsModel) -> Tuple[int, float, float]:
        """
        Get information about generators that turned on or off this time period

        Returns
        -------
        tuple
           a tuple with the following values:
              * num_on_offs: The number of generators that turned on or turned off this time period
              * sum_on_off_ramps: Amount of power generated by generators that turned on or off
              * sum_nominal_ramps: Change in amount of power generated by generators that stayed on or stayed off
        """
        num_on_offs = 0
        sum_on_off_ramps = 0.0
        sum_nominal_ramps = 0.0  # this is the total ramp change for units not switching on/off
    
        for g in self.get_thermal_generators(sced):
            unit_on = self.is_generator_on(sced, g)
            power_generated = self.get_power_generated(sced, g)
            was_on = self.generator_was_on(sced, g)
            if unit_on != was_on:
                # Unit turned on or turned off
                num_on_offs += 1
                sum_on_off_ramps += power_generated
            else:
                # Unit stayed on or stayed off
                sum_nominal_ramps += math.fabs(power_generated - self.get_power_generated_T0(sced, g))

        return num_on_offs, sum_on_off_ramps, sum_nominal_ramps

    def get_load_mismatches(self, sced: OperationsModel) -> Tuple[float, float]:
        """ Get under-generation (load shedding) and over-generation.

        Returns
        -------
        load_shedding: float
            The total load shedding across all buses
        over_generation: float
            The total over-generation across all buses
        """
        load_generation_mismatch_value = round_small_values(sum(self.get_load_mismatch(sced, b)
                                                                for b in self.get_buses(sced)))
        if load_generation_mismatch_value != 0.0:
            load_shedding_value = round_small_values(sum(self.get_positive_load_mismatch(sced, b)
                                                         for b in self.get_buses(sced)))
            over_generation_value = round_small_values(sum(self.get_negative_load_mismatch(sced, b)
                                                           for b in self.get_buses(sced)))
        else:
            load_shedding_value = 0.0
            over_generation_value = 0.0

        return load_shedding_value, over_generation_value

    def has_load_shedding(self, sced: OperationsModel) -> bool:
        return any(round_small_values(self.get_positive_load_mismatch(sced, b)) > 0.0
                   for b in self.get_buses(sced))

    def get_fleet_thermal_capacity(self, sced: OperationsModel) -> float:
        return sum(self.get_max_power_output(sced, g)
                   for g in self.get_thermal_generators(sced))

    def get_thermal_headroom(self, sced: OperationsModel, g: G) -> float:
        return self.get_max_power_available(sced, g) - self.get_power_generated(sced, g)

    def get_all_generator_fuels(self, sced: OperationsModel) -> Dict[G, str]:
        """Get the power generated by each generator."""
        return {g: self.get_generator_fuel(sced, g)
                for g in self.get_all_generators(sced)}

    def get_all_thermal_dispatch_levels(self, sced: OperationsModel) -> Dict[G, float]:
        """Get the power generated by each generator."""
        return {g: self.get_power_generated(sced, g)
                for g in self.get_thermal_generators(sced)}

    def get_all_virtual_dispatch_levels(self, sced: OperationsModel) -> Dict[G, float]:
        """Get the power generated by each generator."""
        return {g: self.get_power_generated(sced, g)
                for g in self.get_virtual_generators(sced)}

    def get_all_thermal_states(self, sced: OperationsModel) -> Dict[G, bool]:
        """Get whether each thermal generator is turned on."""
        return {g: self.is_generator_on(sced, g)
                for g in self.get_thermal_generators(sced)}

    def get_all_thermal_headroom_levels(self, sced: OperationsModel) -> Dict[G, float]:
        """ Get the thermal headroom for all thermal generators """
        return {g: self.get_thermal_headroom(sced, g)
                for g in self.get_thermal_generators(sced)}

    def get_available_reserve(self, sced: OperationsModel) -> float:
        return sum(self.get_thermal_headroom(sced, g)
                   for g in self.get_thermal_generators(sced))


    def get_available_quick_start(self, sced: OperationsModel) -> float:
        """Given a SCED sced instance with commitments from the RUC,
           determine how much quick start capacity is available 
        """
        available_quick_start_capacity = 0.0 
        for g in self.get_quickstart_generators(sced):
            available = True  # until proven otherwise
            if self.is_generator_on(sced, g):
                available = False  # unit was already committed in the RUC
            elif self.generator_was_on(sced, g):
                # there cannot have been a a shutdown in the previous hour 
                available = False
            elif self.get_min_downtime(sced, g) > 1:
                # minimum downtime should be 1 or less, by definition of a quick start
                available = False
 
            if available:  # add the amount of power that can be accessed in the first hour
                # use the min() because scaled startup ramps can be larger than the generator limit
                available_quick_start_capacity += min(self.get_scaled_startup_ramp_limit(sced, g), self.get_max_power_output(sced, g))
        
        return available_quick_start_capacity

    def get_all_thermal_quickstart_capable_flag(self, sced: OperationsModel) -> Dict[G, float]:
        """ get a true/false for all thermal generators if they're availalbe for quickstart """
        fast_start_gens = set(self.get_quickstart_generators(sced))
        return { g : (g in fast_start_gens) for g in self.get_thermal_generators(sced) }

    def get_renewables_curtailment(self, sced: OperationsModel, g: G) -> float:
        return self.get_max_nondispatchable_power(sced, g) - self.get_nondispatchable_power_used(sced, g)

    def get_total_renewables_curtailment(self, sced: OperationsModel) -> float:
        """ get how much renewable power could have been used but wasn't """
        total_curtailment = round_small_values(sum(self.get_renewables_curtailment(sced, g) 
                                                   for g in self.get_nondispatchable_generators(sced)))
        return total_curtailment

    def get_all_renewables_curtailment(self, sced: OperationsModel) -> Dict[G, float]:
        """ get curtailment for each renewables generator """
        return {g: self.get_renewables_curtailment(sced, g) 
                for g in self.get_nondispatchable_generators(sced)}

    def get_renewables_available(self, sced: OperationsModel) -> float:
        renewables_available = sum(self.get_max_nondispatchable_power(sced, g) 
                                   for g in self.get_nondispatchable_generators(sced))
        return renewables_available

    def get_renewables_used(self, sced: OperationsModel) -> float:
        renewables_used = sum(self.get_nondispatchable_power_used(sced, g)
                              for g in self.get_nondispatchable_generators(sced))
        return renewables_used

    def get_price(self, sced_duration_minutes: int, demand: float, fixed_costs: float, variable_costs: float) -> float:
        # 0 demand can happen, in some odd circumstances (not at the ISO level!).
        sceds_per_hr = 60 / sced_duration_minutes
        return 0.0 if demand == 0 else ((fixed_costs + variable_costs) / demand)*sceds_per_hr

    def get_cost_per_generator(self, sced: OperationsModel) -> Dict[G, float]:
        """ get the cost to operate each thermal generator """
        return {g: self.get_generator_cost(sced, g)
                for g in self.get_thermal_generators(sced)}

    def get_all_nondispatchable_power_used(self, sced: OperationsModel):
        return {g: self.get_nondispatchable_power_used(sced, g)
                for g in self.get_nondispatchable_generators(sced)}

    def get_all_flow_levels(self, sced: OperationsModel) -> Dict[L, float]:
        return {l: self.get_flow_level(sced, l)
                for l in self.get_transmission_lines(sced)}

    def get_all_bus_demands(self, sced: OperationsModel) -> Dict[B, float]:
        return {b: self.get_bus_demand(sced, b)
                for b in self.get_buses(sced)}

    def get_all_bus_mismatches(self, sced: OperationsModel) -> Dict[B, float]: 
        return {b: self.get_bus_mismatch(sced, b)
                for b in self.get_buses(sced)}

    def get_all_storage_input_dispatch_levels(self, sced: OperationsModel) -> Dict[S, float]:
        return {s: self.get_storage_input_dispatch_level(sced, s)
                for s in self.get_all_storage(sced)}

    def get_all_storage_output_dispatch_levels(self, sced: OperationsModel) -> Dict[S, float]:
        return {s: self.get_storage_output_dispatch_level(sced, s)
                for s in self.get_all_storage(sced)}

    def get_all_storage_soc_dispatch_levels(self, sced: OperationsModel) -> Dict[S, float]:
        return {s: self.get_storage_soc_dispatch_level(sced, s)
                for s in self.get_all_storage(sced)}

    def get_all_storage_types(self, sced: OperationsModel) -> Dict[S, float]:
        return {s: self.get_storage_soc_dispatch_level(sced, s)
                for s in self.get_all_storage(sced)}


### LMP Methods ###
#
#
    @abstractmethod
    def get_reserve_RT_price(self, lmp_sced: OperationsModel) -> float:
        """
        Gets the reserve RT price

        Parameters
        ----------
        lmp_sced
            An operations model that has been solved for LMP
        """
        pass

    @abstractmethod
    def get_bus_LMP(self, lmp_sced: OperationsModel, bus: B) -> float:
        """
        Gets the LMP for a single bus

        Parameters
        ----------
        lmp_sced: 
            An operations model that has been solved for LMP

        bus: 
            The bus whose LMP is wanted
        """

    def get_all_bus_LMPs(self, lmp_sced: OperationsModel) -> Dict[B, float]:
        """
        Gets a dictionary where keys are buses and values are the LMP for the bus

        There is one entry for each bus in the model

        Parameters
        ----------
        lmp_sced:
            A schedule that has been solved for LMP
        """
        return {b: self.get_bus_LMP(lmp_sced, b)
                for b in self.get_buses(lmp_sced)}

#
#
### LMP Methods ###

class RucDataExtractor(ABC):
    """
    Extracts information from RUC model instances
    """

    @abstractmethod
    def get_num_time_periods(self, ruc: RucModel) -> int:
        """ Get the number of time periods for which data is available.
            
            Time periods are numbered 1..N, where N is the value returned by this method.
        """
        pass

    @abstractmethod
    def get_buses(self, ruc: RucModel) -> Iterable[B]:
        """ Get all buses in the model """
        pass

    @abstractmethod
    def get_bus_demand(self, ruc: RucModel, bus: B, time: int) -> float:
        """ get the demand on a bus in a given time period """
        pass

    @abstractmethod
    def get_nondispatchable_generators(self, ruc: RucModel) -> Iterable[G]:
        """Get all non-dispatchable generators in the model"""
        pass

    @abstractmethod
    def get_min_nondispatchable_power(self, ruc: RucModel, gen: G, time: int) -> float:
        pass

    @abstractmethod
    def get_max_nondispatchable_power(self, ruc: RucModel, gen: G, time: int) -> float:
        pass