<s> """<STR_LIT>""" <EOL> import logging <EOL> try : <EOL> import matplotlib . pyplot as plt <EOL> from matplotlib . colors import ListedColormap <EOL> except ImportError : <EOL> logging . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> import numpy as np <EOL> import pandas as pd <EOL> from . import core <EOL> from . import utils <EOL> from . import analysis_utils as au <EOL> def plot_carrier_production ( solution , carrier = '<STR_LIT>' , subset = dict ( ) , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> data = solution [ '<STR_LIT:e>' ] . loc [ dict ( c = carrier , ** subset ) ] . sum ( dim = '<STR_LIT:x>' ) <EOL> return plot_timeseries ( solution , data , carrier = carrier , ** kwargs ) <EOL> def plot_timeseries ( solution , data , carrier = '<STR_LIT>' , demand = '<STR_LIT>' , <EOL> types = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> colormap = None , ticks = None , <EOL> resample_options = None , resample_func = None ) : <EOL> """<STR_LIT>""" <EOL> if not ticks : <EOL> days = ( data . coords [ '<STR_LIT:t>' ] . values [ - <NUM_LIT:1> ] - data . coords [ '<STR_LIT:t>' ] . values [ <NUM_LIT:0> ] ) <EOL> timespan = days / np . timedelta64 ( <NUM_LIT:1> , '<STR_LIT:D>' ) <EOL> if timespan <= <NUM_LIT:2> : <EOL> ticks = '<STR_LIT>' <EOL> elif timespan < <NUM_LIT> : <EOL> ticks = '<STR_LIT>' <EOL> else : <EOL> ticks = '<STR_LIT>' <EOL> time_res = solution [ '<STR_LIT>' ] . to_pandas ( ) <EOL> data = data . to_pandas ( ) . T <EOL> plot_df = data . divide ( time_res , axis = '<STR_LIT:index>' ) <EOL> if resample_options and resample_func : <EOL> plot_df = getattr ( plot_df . resample ( ** resample_options ) , resample_func ) ( ) <EOL> df = solution [ '<STR_LIT>' ] . to_pandas ( ) . query ( '<STR_LIT>' . format ( carrier ) ) <EOL> query_string = au . _get_query_string ( types ) <EOL> stacked_techs = df . query ( query_string ) . index . tolist ( ) <EOL> weighted = df . stack_weight . sort_values ( ascending = False ) . index . tolist ( ) <EOL> stacked_techs = [ y for y in weighted if y in stacked_techs ] <EOL> names = [ df . at [ y , '<STR_LIT:name>' ] for y in stacked_techs ] <EOL> if not colormap : <EOL> colors = [ df . at [ i , '<STR_LIT>' ] for i in stacked_techs ] <EOL> colormap = ListedColormap ( colors ) <EOL> ax = au . stack_plot ( plot_df , stacked_techs , colormap = colormap , <EOL> alpha = <NUM_LIT> , ticks = ticks , legend = None , names = names ) <EOL> ax . plot ( plot_df [ demand ] . index , <EOL> plot_df [ demand ] * - <NUM_LIT:1> , <EOL> color = '<STR_LIT>' , lw = <NUM_LIT> , ls = '<STR_LIT>' , label = df . at [ demand , '<STR_LIT:name>' ] ) <EOL> au . legend_outside_ax ( ax , where = '<STR_LIT:right>' ) <EOL> return ax <EOL> def plot_installed_capacities ( solution , <EOL> types = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> unit_multiplier = <NUM_LIT:1.0> , <EOL> unit_label = '<STR_LIT>' , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> query_string = au . _get_query_string ( types ) <EOL> md = solution . metadata . to_pandas ( ) <EOL> supply_cap = md . query ( query_string ) . index . tolist ( ) <EOL> df = solution [ '<STR_LIT>' ] . loc [ dict ( y = supply_cap ) ] . to_pandas ( ) <EOL> weighted = md . stack_weight . sort_values ( ascending = False ) . index . tolist ( ) <EOL> stacked_techs = [ y for y in weighted if y in df . columns ] <EOL> df = df . loc [ : , stacked_techs ] * unit_multiplier <EOL> names = [ md . at [ y , '<STR_LIT:name>' ] for y in df . columns ] <EOL> colors = [ md . at [ i , '<STR_LIT>' ] for i in df . columns ] <EOL> colormap = ListedColormap ( colors ) <EOL> proxies = [ plt . Rectangle ( ( <NUM_LIT:0> , <NUM_LIT:0> ) , <NUM_LIT:1> , <NUM_LIT:1> , fc = i ) <EOL> for i in colors ] <EOL> if ( '<STR_LIT>' in solution . config_model and <EOL> '<STR_LIT>' in solution . config_model . metadata ) : <EOL> meta_config = solution . config_model . metadata <EOL> for index , item in enumerate ( meta_config . location_ordering ) : <EOL> if item in df . index : <EOL> df . at [ item , '<STR_LIT>' ] = index <EOL> df = df . sort_values ( by = '<STR_LIT>' , ascending = False ) <EOL> df = df . drop ( '<STR_LIT>' , axis = <NUM_LIT:1> ) <EOL> ax = df . plot ( kind = '<STR_LIT>' , stacked = True , legend = False , colormap = colormap , <EOL> ** kwargs ) <EOL> leg = au . legend_outside_ax ( ax , where = '<STR_LIT:right>' , artists = proxies , labels = names ) <EOL> ylab = ax . set_ylabel ( '<STR_LIT>' ) <EOL> xlab = ax . set_xlabel ( '<STR_LIT>' . format ( unit_label ) ) <EOL> return ax <EOL> def plot_transmission ( solution , tech = '<STR_LIT>' , carrier = '<STR_LIT>' , <EOL> labels = '<STR_LIT>' , <EOL> figsize = ( <NUM_LIT:15> , <NUM_LIT:15> ) , fontsize = <NUM_LIT:9> , <EOL> show_scale = True , <EOL> ax = None , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> import networkx as nx <EOL> def get_edge_capacity ( solution , a , b ) : <EOL> hrs = solution [ '<STR_LIT>' ] . to_pandas ( ) . sum ( ) <EOL> cap = solution [ '<STR_LIT>' ] . loc [ dict ( x = a , y = '<STR_LIT>' . format ( tech ) + b ) ] . value ( ) * hrs <EOL> return cap <EOL> zones = sorted ( list ( solution . coords [ '<STR_LIT:x>' ] . values ) ) <EOL> trans_tech = lambda x : '<STR_LIT>' . format ( tech , x ) <EOL> def get_annual_power_transmission ( zone ) : <EOL> try : <EOL> return solution [ '<STR_LIT:e>' ] . loc [ dict ( c = carrier , y = trans_tech ( zone ) ) ] . sum ( dim = '<STR_LIT:t>' ) . value ( ) <EOL> except KeyError : <EOL> return <NUM_LIT:0> <EOL> df = pd . DataFrame ( { zone : get_annual_power_transmission ( zone ) <EOL> for zone in zones } , index = zones ) <EOL> df [ df < <NUM_LIT:0> ] = <NUM_LIT:0> <EOL> G = nx . from_numpy_matrix ( df . as_matrix ( ) . T , create_using = nx . DiGraph ( ) ) <EOL> G = nx . relabel_nodes ( G , dict ( list ( zip ( list ( range ( len ( zones ) ) ) , zones ) ) ) ) <EOL> edge_transmission = { edge : int ( round ( df . at [ edge [ <NUM_LIT:1> ] , edge [ <NUM_LIT:0> ] ] / <NUM_LIT> ) ) <EOL> for edge in G . edges ( ) } <EOL> edge_use = { ( a , b ) : ( df . at [ a , b ] + df . at [ b , a ] ) <EOL> / get_edge_capacity ( solution , a , b ) <EOL> for ( a , b ) in G . edges ( ) } <EOL> if labels == '<STR_LIT>' : <EOL> edge_labels = { k : '<STR_LIT>' . format ( v ) for k , v in edge_use . items ( ) } <EOL> elif labels == '<STR_LIT>' : <EOL> edge_labels = edge_transmission <EOL> edge_colors = [ edge_use [ i ] for i in G . edges ( ) ] <EOL> ax , m = au . plot_graph_on_map ( solution . config_model , G = G , <EOL> edge_colors = edge_colors , <EOL> edge_labels = edge_labels , <EOL> figsize = figsize , fontsize = fontsize , <EOL> ax = ax , show_scale = show_scale , ** kwargs ) <EOL> return ax <EOL> def get_delivered_cost ( solution , cost_class = '<STR_LIT>' , carrier = '<STR_LIT>' , <EOL> count_unmet_demand = False , unit_multiplier = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> summary = solution . summary . to_pandas ( ) <EOL> meta = solution . metadata . to_pandas ( ) <EOL> carrier_subset = meta [ meta . carrier == carrier ] . index . tolist ( ) <EOL> if count_unmet_demand is False : <EOL> try : <EOL> carrier_subset . remove ( '<STR_LIT>' + carrier ) <EOL> except ValueError : <EOL> pass <EOL> cost = solution [ '<STR_LIT>' ] . loc [ dict ( k = cost_class , y = carrier_subset ) ] . to_pandas ( ) . sum ( ) . sum ( ) <EOL> met_demand = summary . at [ '<STR_LIT>' + carrier , '<STR_LIT>' ] <EOL> try : <EOL> unmet_demand = summary . at [ '<STR_LIT>' + carrier , '<STR_LIT>' ] <EOL> except KeyError : <EOL> unmet_demand = <NUM_LIT:0> <EOL> if count_unmet_demand is False : <EOL> demand = met_demand + unmet_demand <EOL> else : <EOL> demand = met_demand <EOL> return cost / demand * - <NUM_LIT:1> <EOL> def get_levelized_cost ( solution , cost_class = '<STR_LIT>' , carrier = '<STR_LIT>' , <EOL> group = None , locations = None , <EOL> unit_multiplier = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> if group : <EOL> members = solution . groups . to_pandas ( ) . at [ group , '<STR_LIT>' ] . split ( '<STR_LIT:|>' ) <EOL> else : <EOL> members = slice ( None ) <EOL> if locations is None : <EOL> locations = slice ( None ) <EOL> if isinstance ( locations , ( str , float , int ) ) : <EOL> locations = [ locations ] <EOL> cost = solution [ '<STR_LIT>' ] . loc [ dict ( k = cost_class , x = locations , y = members ) ] . sum ( dim = '<STR_LIT:x>' ) . to_pandas ( ) <EOL> ec_prod = solution [ '<STR_LIT>' ] . loc [ dict ( c = carrier , x = locations , y = members ) ] . sum ( dim = '<STR_LIT:x>' ) . to_pandas ( ) <EOL> return ( cost / ec_prod ) * unit_multiplier <EOL> def get_group_share ( solution , techs , group_type = '<STR_LIT>' , <EOL> var = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> summary = solution . summary <EOL> meta = solution . metadata . to_pandas ( ) <EOL> group = meta . query ( '<STR_LIT>' + group_type + '<STR_LIT:">' ) . index . tolist ( ) <EOL> if group_type == '<STR_LIT>' : <EOL> def transmission_basenames ( group ) : <EOL> return list ( set ( [ i . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] for i in group ] ) ) <EOL> techs = transmission_basenames ( techs ) <EOL> group = transmission_basenames ( group ) <EOL> supply_total = summary . loc [ group , var ] . sum ( ) <EOL> supply_group = summary . loc [ techs , var ] . sum ( ) <EOL> try : <EOL> return supply_group / supply_total <EOL> except ZeroDivisionError : <EOL> return np . nan <EOL> def get_unmet_demand_hours ( solution , carrier = '<STR_LIT>' , details = False ) : <EOL> """<STR_LIT>""" <EOL> unmet = ( solution [ '<STR_LIT:e>' ] <EOL> . loc [ dict ( c = carrier , y = '<STR_LIT>' + carrier ) ] <EOL> . sum ( dim = '<STR_LIT:x>' ) <EOL> . to_pandas ( ) ) <EOL> timesteps = len ( unmet [ unmet > <NUM_LIT:0> ] ) <EOL> hours = solution . time_res . to_pandas ( ) [ unmet > <NUM_LIT:0> ] . sum ( ) <EOL> if details : <EOL> return { '<STR_LIT>' : hours , '<STR_LIT>' : timesteps , <EOL> '<STR_LIT>' : unmet [ unmet > <NUM_LIT:0> ] . index } <EOL> else : <EOL> return hours <EOL> def areas_below_resolution ( solution , resolution ) : <EOL> """<STR_LIT>""" <EOL> time_res = solution . time_res . to_pandas ( ) <EOL> selected = time_res [ time_res < resolution ] <EOL> return list ( au . _get_ranges ( selected . index . tolist ( ) ) ) <EOL> def get_swi ( solution , shares_var = '<STR_LIT>' , exclude_patterns = [ '<STR_LIT>' ] ) : <EOL> """<STR_LIT>""" <EOL> techs = au . get_supply_groups ( solution ) <EOL> for pattern in exclude_patterns : <EOL> techs = [ t for t in techs if pattern not in t ] <EOL> swi = - <NUM_LIT:1> * sum ( ( p * np . log ( p ) ) <EOL> for p in [ solution . shares . to_pandas ( ) . at [ y , shares_var ] <EOL> for y in techs ] <EOL> if p > <NUM_LIT:0> ) <EOL> return swi <EOL> def get_hhi ( solution , shares_var = '<STR_LIT>' , exclude_patterns = [ '<STR_LIT>' ] ) : <EOL> """<STR_LIT>""" <EOL> techs = au . get_supply_groups ( solution ) <EOL> for pattern in exclude_patterns : <EOL> techs = [ t for t in techs if pattern not in t ] <EOL> hhi = sum ( ( solution . shares . to_pandas ( ) . at [ y , shares_var ] * <NUM_LIT> ) ** <NUM_LIT:2> <EOL> for y in techs ) <EOL> return hhi <EOL> def get_domestic_supply_index ( solution ) : <EOL> """<STR_LIT>""" <EOL> idx = solution . metadata . query ( '<STR_LIT>' ) . index . tolist ( ) <EOL> dom = ( solution . costs . loc [ dict ( k = '<STR_LIT>' , y = idx ) ] . sum ( ) . sum ( ) / <EOL> solution [ '<STR_LIT>' ] . loc [ dict ( c = '<STR_LIT>' ) ] . sum ( ) . sum ( ) ) <EOL> return dom <EOL> def map_results ( results , func , as_frame = False ) : <EOL> """<STR_LIT>""" <EOL> def map_func ( x ) : <EOL> try : <EOL> return func ( x ) <EOL> except Exception : <EOL> return np . nan <EOL> iterations = results . iterations . index <EOL> items = [ map_func ( results . solutions [ i ] ) for i in iterations ] <EOL> names = [ results . solutions [ i ] . config_run . name <EOL> if '<STR_LIT:name>' in results . solutions [ i ] . config_run <EOL> else i <EOL> for i in iterations ] <EOL> if as_frame : <EOL> return pd . DataFrame ( items , index = names ) <EOL> else : <EOL> return pd . Series ( items , index = names ) <EOL> class SolutionModel ( core . BaseModel ) : <EOL> def __init__ ( self , solution ) : <EOL> """<STR_LIT>""" <EOL> super ( ) . __init__ ( ) <EOL> self . config_model = solution . config_model <EOL> self . data = utils . AttrDict ( ) <EOL> for k in [ '<STR_LIT:r>' , '<STR_LIT>' ] : <EOL> if k in solution : <EOL> ds = solution [ k ] . to_dataset ( dim = '<STR_LIT>' . format ( k ) ) <EOL> for y in ds . data_vars : <EOL> self . data . set_key ( '<STR_LIT>' . format ( k , y ) , ds [ y ] . to_pandas ( ) ) <EOL> del solution [ k ] <EOL> self . solution = solution <EOL> self . _get_option = utils . option_getter ( self . config_model , solution ) <EOL> self . _get_cost = utils . cost_getter ( self . _get_option ) <EOL> self . _get_dep = utils . depreciation_getter ( self . _get_option ) <EOL> def get_option ( self , * args , ** kwargs ) : <EOL> return self . _get_option ( * args , ** kwargs ) <EOL> def get_cost ( self , * args , ** kwargs ) : <EOL> return self . _get_cost ( * args , ** kwargs ) <EOL> def get_depreciation ( self , * args , ** kwargs ) : <EOL> return self . _get_dep ( * args , ** kwargs ) <EOL> def recompute_investment_costs ( self , y , k = '<STR_LIT>' , <EOL> cost_adjustment = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> get_cost = self . get_cost <EOL> solution = self . solution <EOL> cost_con = pd . DataFrame ( ) <EOL> cost_con [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con [ '<STR_LIT>' ] = cost_con . sum ( axis = <NUM_LIT:1> ) <EOL> cost_con [ '<STR_LIT>' ] = ( cost_con [ '<STR_LIT>' ] <EOL> / solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] ) <EOL> cost_con . at [ '<STR_LIT>' , '<STR_LIT>' ] = cost_con [ '<STR_LIT>' ] . sum ( ) <EOL> total_per_e_cap = ( cost_con . at [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> / solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] . sum ( ) ) <EOL> cost_con . at [ '<STR_LIT>' , '<STR_LIT>' ] = total_per_e_cap <EOL> return cost_con * cost_adjustment <EOL> def recompute_operational_costs ( self , y , k = '<STR_LIT>' , carrier = '<STR_LIT>' , <EOL> cost_adjustment = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> get_cost = self . get_cost <EOL> get_depreciation = self . get_depreciation <EOL> solution = self . solution <EOL> production = solution [ '<STR_LIT>' ] . loc [ dict ( c = carrier , y = y ) ] <EOL> fuel = ( solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] . sum ( dim = '<STR_LIT:t>' ) <EOL> / self . get_option ( y + '<STR_LIT>' ) ) <EOL> fuel_rb = ( solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] . sum ( dim = '<STR_LIT:t>' ) <EOL> / self . get_option ( y + '<STR_LIT>' ) ) <EOL> icf = self . recompute_investment_costs <EOL> cost_con = icf ( y , k = k , cost_adjustment = cost_adjustment ) [ '<STR_LIT>' ] . sum ( ) <EOL> cost_op = pd . DataFrame ( ) <EOL> cost_op [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) * cost_con <EOL> * get_depreciation ( y , k ) <EOL> * ( solution [ '<STR_LIT>' ] . to_pandas ( ) . sum ( ) / <NUM_LIT> ) ) <EOL> cost_op [ '<STR_LIT>' ] = ( get_cost ( '<STR_LIT>' , y , k ) <EOL> * solution [ '<STR_LIT>' ] . loc [ dict ( y = y ) ] <EOL> * ( solution [ '<STR_LIT>' ] . to_pandas ( ) . sum ( ) / <NUM_LIT> ) ) <EOL> cost_op [ '<STR_LIT>' ] = get_cost ( '<STR_LIT>' , y , k ) * production <EOL> cost_op [ '<STR_LIT>' ] = get_cost ( '<STR_LIT>' , y , k ) * fuel <EOL> cost_op [ '<STR_LIT>' ] = get_cost ( '<STR_LIT>' , y , k ) * fuel_rb <EOL> cost_op [ '<STR_LIT>' ] = cost_op . sum ( axis = <NUM_LIT:1> ) <EOL> return cost_op <EOL> def recompute_levelized_costs ( self , y , k = '<STR_LIT>' , carrier = '<STR_LIT>' , <EOL> cost_adjustment = <NUM_LIT:1.0> ) : <EOL> """<STR_LIT>""" <EOL> get_depreciation = self . get_depreciation <EOL> solution = self . solution <EOL> icf = self . recompute_investment_costs <EOL> cost_con = icf ( y , k = k , cost_adjustment = cost_adjustment ) [ '<STR_LIT>' ] <EOL> ocf = self . recompute_operational_costs <EOL> cost_op = ocf ( y , k = k , carrier = carrier , <EOL> cost_adjustment = cost_adjustment ) [ '<STR_LIT>' ] <EOL> production = solution [ '<STR_LIT>' ] . loc [ dict ( c = carrier , y = y ) ] <EOL> lc = ( get_depreciation ( y , k ) * ( solution [ '<STR_LIT>' ] . to_pandas ( ) . sum ( ) / <NUM_LIT> ) <EOL> * cost_con + cost_op ) / production . to_pandas ( ) <EOL> lc . at [ '<STR_LIT>' ] = ( get_depreciation ( y , k ) <EOL> * ( solution [ '<STR_LIT>' ] . to_pandas ( ) . sum ( ) / <NUM_LIT> ) <EOL> * cost_con . total <EOL> + cost_op . sum ( ) ) / production . sum ( ) <EOL> return lc </s>
