Last active
January 26, 2021 13:49
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Utility functions for the analysis of oemof results.""" | |
import pandas as pd | |
import graphviz | |
from oemof.solph.network import Sink, Source, Bus, Transformer | |
from oemof.solph.components import GenericStorage | |
# Define shapes for the component types | |
SHAPES = {Source: 'trapezium', | |
Sink: 'invtrapezium', | |
Bus: 'ellipse', | |
Transformer: 'octagon', | |
GenericStorage: 'cylinder'} | |
def generate_graph(energysystem): | |
"""Generate graphviz graph from energysystem.""" | |
dot = graphviz.Digraph(format='png') | |
for node in energysystem.nodes: | |
dot.node(node.label, shape=SHAPES.get(type(node), 'rectangle')) | |
for node in energysystem.nodes: | |
for output in node.outputs: | |
dot.edge(node.label, output.label) | |
return dot | |
def get_flows(results): | |
""" | |
Extract flows from results dictionary. | |
To access the data you might want to use the xs function, i.e. | |
>>> flows = get_flows(results) | |
>>> flows.xs('component0', axis=1, level='source') | |
>>> flows.xs('component0', axis=1, level='destination') | |
:param results: Results from oemof optimization | |
""" | |
flows = {(source_node.label, destination_node.label): result['sequences']['flow'] | |
for (source_node, destination_node), result in results.items() | |
if destination_node is not None} | |
return pd.concat(flows.values(), axis=1, names=['source', 'destination'], keys=flows.keys()) | |
def get_variables(results): | |
""" | |
Extract variables from results dictionary. | |
To access the data you might want to use the xs function, i.e. | |
>>> vars = get_variables(results) | |
>>> vars.xs('component0', axis=1, level='component') | |
>>> vars.xs('variable0', axis=1, level='variable_name') | |
:param results: Results from oemof optimization | |
""" | |
variables = {source_node.label: result['sequences'] | |
for (source_node, destination_node), result in results.items() | |
if destination_node is None} | |
return pd.concat(variables.values(), axis=1, names=['component',], keys=variables.keys()) | |
def get_flows_for_component(flows, component): | |
"""Get flows in and out of component.""" | |
inflows = flows.xs(component, axis=1, level='destination') \ | |
if component in flows.columns.get_level_values('destination') \ | |
else pd.DataFrame(index=flows) | |
outflows = flows.xs(component, axis=1, level='source') \ | |
if component in flows.columns.get_level_values('source') \ | |
else pd.DataFrame(index=flows) | |
return pd.concat([inflows.rename('from_{0}'.format, axis=1), | |
outflows.rename('to_{0}'.format, axis=1)], | |
axis=1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment