db.results.restore_results() - Code Metrics - oemof/oemof-db - Measure and Improve Code Quality continuously with Scrutinizer

db.results.restore_results() D
last analyzed 2023-11-23 16:40 UTC

↳ Parent: db.results

Complexity

Conditions

Size

Total Lines	56
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	36
nop	2
dl	0
loc	56
rs	4.8
c	0
b	0
f	0

How to fix Long Method Complexity

"""
Module to store and restore input- and result-data from oemof into database.
Works with oemof version 2.0

Notes
-----
By now, table names are static - if you are not satisfied with resulting
table names, you have to change code yourself!

Examples
--------
The following code will setup your sqlalchemy session and
create all needed tables in database:

.. code-block:: python

    from sqlalchemy import orm, create_engine
    from oemof.db import results

    # You have to add valid SQLAlchemy URL-string first!
    engine = create_engine(db_url)
    SqlAlchemySession = orm.sessionmaker(bind=engine)
    results.Base.metadata.bind = engine
    results.Base.metadata.create_all()

The following code stores your data into DB:

.. code-block:: python

    sa_session = SqlAlchemySession()
    results.store_results(sa_session, input_dict, result_dict)
    sa_session.close()

The following code restores your data from DB:

.. code-block:: python

    sa_session = SqlAlchemySession()
    input_dict, result_dict = results.restore_results(sa_session, result_id)
    sa_session.close()
"""

from sqlalchemy import ARRAY, Column, ForeignKey, Integer, String
from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
import pandas

from oemof.solph.processing import convert_keys_to_strings

Base = declarative_base()


class OemofInputResult(Base):
    __tablename__ = 'oemof_input_result'

    input_result_id = Column(
        Integer,
        primary_key=True
    )
    input_id = Column(Integer, ForeignKey('oemof_data.data_id'))
    result_id = Column(Integer, ForeignKey('oemof_data.data_id'))
    input = relationship(
        "OemofData",
        backref="input",
        uselist=False,
        foreign_keys=[input_id],
    )
    result = relationship(
        "OemofData",
        backref="result",
        uselist=False,
        foreign_keys=[result_id],
    )


class OemofData(Base):
    __tablename__ = 'oemof_data'

    data_id = Column(
        Integer,
        primary_key=True
    )
    scalars = relationship("OemofScalar", cascade="delete")
    sequences = relationship("OemofSequence", cascade="delete")


class OemofScalar(Base):
    __tablename__ = 'oemof_scalar'

    scalar_id = Column(
        Integer,
        primary_key=True
    )
    data_id = Column(
        Integer, ForeignKey('oemof_data.data_id'))
    from_node = Column(
        String
    )
    to_node = Column(
        String
    )
    attribute = Column(
        String
    )
    value = Column(
        String
    )
    type = Column(
        String
    )


class OemofSequence(Base):
    __tablename__ = 'oemof_sequence'

    sequence_id = Column(
        Integer,
        primary_key=True
    )
    data_id = Column(
        Integer, ForeignKey('oemof_data.data_id'))
    from_node = Column(
        String
    )
    to_node = Column(
        String,
        nullable=True
    )
    attribute = Column(
        String
    )
    value = Column(
        ARRAY(DOUBLE_PRECISION)
    )
    type = Column(
        String
    )


def store_results(session, input_data, result_data):
    """
    Stores inputs and results from oemof into DB

    For each in entry in scalars and sequences of both input- and result-data
    an OemofScalar or OemofSequence is build and connected to an OemofData
    object representing either input or result data. At last, both OemofData
    objects are connected to OemofInputResult object and resulting index is
    returned.

    Parameters
    ----------
    session: sqlalchemy.session
        SQLAlchemy session build via sqlalchemy.orm.sessionmaker
    input_data: dict
        Output of oemof.outputlib.processing.param_results with nodes as str
        (use oemof.outputlib.processing.convert_keys_to_str if necessary)
    result_data: dict
        Output of oemof.outputlib.processing.param_results with nodes as str
        (use oemof.outputlib.processing.convert_keys_to_str if necessary)

    Returns
    -------
    int: Index of created OemofInputResult entry
    """
    # Check if nodes are strings:
    if not isinstance(next(iter(input_data)), str):
        input_data = convert_keys_to_strings(input_data)
    if not isinstance(next(iter(result_data)), str):
        result_data = convert_keys_to_strings(result_data)

    input_result = OemofInputResult()
    for input_result_attr, data in (
            ('input', input_data), ('result', result_data)):
        scalars = []
        sequences = []
        for (from_node, to_node), sc_sq_dict in data.items():
            for key, value in sc_sq_dict['scalars'].items():
                scalars.append(
                    OemofScalar(
                        from_node=from_node,
                        to_node=to_node,
                        attribute=key,
                        value=value,
                        type=type(value).__name__
                    )
                )
            session.add_all(scalars)
            for key, series in sc_sq_dict['sequences'].items():
                list_type = 'list'
                if isinstance(series, pandas.Series):
                    series = series.values.tolist()
                    list_type = 'series'
                sequences.append(
                    OemofSequence(
                        from_node=from_node,
                        to_node=to_node,
                        attribute=key,
                        value=series,
                        type=list_type
                    )
                )
            session.add_all(sequences)
        oemof_data = OemofData()
        oemof_data.scalars = scalars
        oemof_data.sequences = sequences
        setattr(input_result, input_result_attr, oemof_data)
    session.add(input_result)
    session.flush()
    result_id = input_result.input_result_id
    session.commit()
    return result_id


def restore_results(session, input_result_id):
    """
    Restores input and result data from OemofInputResult from DB

    Parameters
    ----------
    session: sqlalchemy.session
        SQLAlchemy session build via sqlalchemy.orm.sessionmaker
    input_result_id: int
        Index of OemofInputResult object to restore

    Returns
    -------
    (dict, dict):
        Restored input- and result-data
    """
    def type_conversion(value_str, value_type):
        if value_type == 'str':
            return value_str
        elif value_type == 'float':
            return float(value_str)
        elif value_type == 'int':
            return int(value_str)
        elif value_type == 'bool':
            return bool(value_str)
        else:
            raise TypeError('Unknown conversion type "' + value_type + '"')

    # Find results:
    input_result = session.query(OemofInputResult).filter(
        OemofInputResult.input_result_id == input_result_id).first()
    if input_result is None:
        raise IndexError(
            'Could not find OemofInputResult with ID #' + str(input_result_id))

    input_data = {}
    result_data = {}
    for input_result_attr, data in (
            ('input', input_data), ('result', result_data)):
        ir_attr = getattr(input_result, input_result_attr)
        for scalar in ir_attr.scalars:
            nodes = (scalar.from_node, scalar.to_node)
            if nodes not in data:
                data[nodes] = {'scalars': {}, 'sequences': {}}
            data[nodes]['scalars'][scalar.attribute] = type_conversion(
                scalar.value, scalar.type)
        for sequence in ir_attr.sequences:
            nodes = (sequence.from_node, sequence.to_node)
            if nodes not in data:
                data[nodes] = {'scalars': {}, 'sequences': {}}
            if sequence.type == 'series':
                series = pandas.Series(sequence.value)
            else:
                series = sequence.value
            data[nodes]['sequences'][sequence.attribute] = series
    return input_data, result_data


1			"""
2			Module to store and restore input- and result-data from oemof into database.
3			Works with oemof version 2.0
4
5			Notes
6			-----
7			By now, table names are static - if you are not satisfied with resulting
8			table names, you have to change code yourself!
9
10			Examples
11			--------
12			The following code will setup your sqlalchemy session and
13			create all needed tables in database:
14
15			.. code-block:: python
16
17			from sqlalchemy import orm, create_engine
18			from oemof.db import results
19
20			# You have to add valid SQLAlchemy URL-string first!
21			engine = create_engine(db_url)
22			SqlAlchemySession = orm.sessionmaker(bind=engine)
23			results.Base.metadata.bind = engine
24			results.Base.metadata.create_all()
25
26			The following code stores your data into DB:
27
28			.. code-block:: python
29
30			sa_session = SqlAlchemySession()
31			results.store_results(sa_session, input_dict, result_dict)
32			sa_session.close()
33
34			The following code restores your data from DB:
35
36			.. code-block:: python
37
38			sa_session = SqlAlchemySession()
39			input_dict, result_dict = results.restore_results(sa_session, result_id)
40			sa_session.close()
41			"""
42
43			from sqlalchemy import ARRAY, Column, ForeignKey, Integer, String
44			from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION
45			from sqlalchemy.ext.declarative import declarative_base
46			from sqlalchemy.orm import relationship
47			import pandas
48
49			from oemof.solph.processing import convert_keys_to_strings
50
51			Base = declarative_base()
52
53
54			class OemofInputResult(Base):
55			__tablename__ = 'oemof_input_result'
56
57			input_result_id = Column(
58			Integer,
59			primary_key=True
60			)
61			input_id = Column(Integer, ForeignKey('oemof_data.data_id'))
62			result_id = Column(Integer, ForeignKey('oemof_data.data_id'))
63			input = relationship(
64			"OemofData",
65			backref="input",
66			uselist=False,
67			foreign_keys=[input_id],
68			)
69			result = relationship(
70			"OemofData",
71			backref="result",
72			uselist=False,
73			foreign_keys=[result_id],
74			)
75
76
77			class OemofData(Base):
78			__tablename__ = 'oemof_data'
79
80			data_id = Column(
81			Integer,
82			primary_key=True
83			)
84			scalars = relationship("OemofScalar", cascade="delete")
85			sequences = relationship("OemofSequence", cascade="delete")
86
87
88			class OemofScalar(Base):
89			__tablename__ = 'oemof_scalar'
90
91			scalar_id = Column(
92			Integer,
93			primary_key=True
94			)
95			data_id = Column(
96			Integer, ForeignKey('oemof_data.data_id'))
97			from_node = Column(
98			String
99			)
100			to_node = Column(
101			String
102			)
103			attribute = Column(
104			String
105			)
106			value = Column(
107			String
108			)
109			type = Column(
110			String
111			)
112
113
114			class OemofSequence(Base):
115			__tablename__ = 'oemof_sequence'
116
117			sequence_id = Column(
118			Integer,
119			primary_key=True
120			)
121			data_id = Column(
122			Integer, ForeignKey('oemof_data.data_id'))
123			from_node = Column(
124			String
125			)
126			to_node = Column(
127			String,
128			nullable=True
129			)
130			attribute = Column(
131			String
132			)
133			value = Column(
134			ARRAY(DOUBLE_PRECISION)
135			)
136			type = Column(
137			String
138			)
139
140
141			def store_results(session, input_data, result_data):
142			"""
143			Stores inputs and results from oemof into DB
144
145			For each in entry in scalars and sequences of both input- and result-data
146			an OemofScalar or OemofSequence is build and connected to an OemofData
147			object representing either input or result data. At last, both OemofData
148			objects are connected to OemofInputResult object and resulting index is
149			returned.
150
151			Parameters
152			----------
153			session: sqlalchemy.session
154			SQLAlchemy session build via sqlalchemy.orm.sessionmaker
155			input_data: dict
156			Output of oemof.outputlib.processing.param_results with nodes as str
157			(use oemof.outputlib.processing.convert_keys_to_str if necessary)
158			result_data: dict
159			Output of oemof.outputlib.processing.param_results with nodes as str
160			(use oemof.outputlib.processing.convert_keys_to_str if necessary)
161
162			Returns
163			-------
164			int: Index of created OemofInputResult entry
165			"""
166			# Check if nodes are strings:
167			if not isinstance(next(iter(input_data)), str):
168			input_data = convert_keys_to_strings(input_data)
169			if not isinstance(next(iter(result_data)), str):
170			result_data = convert_keys_to_strings(result_data)
171
172			input_result = OemofInputResult()
173			for input_result_attr, data in (
174			('input', input_data), ('result', result_data)):
175			scalars = []
176			sequences = []
177			for (from_node, to_node), sc_sq_dict in data.items():
178			for key, value in sc_sq_dict['scalars'].items():
179			scalars.append(
180			OemofScalar(
181			from_node=from_node,
182			to_node=to_node,
183			attribute=key,
184			value=value,
185			type=type(value).__name__
186			)
187			)
188			session.add_all(scalars)
189			for key, series in sc_sq_dict['sequences'].items():
190			list_type = 'list'
191			if isinstance(series, pandas.Series):
192			series = series.values.tolist()
193			list_type = 'series'
194			sequences.append(
195			OemofSequence(
196			from_node=from_node,
197			to_node=to_node,
198			attribute=key,
199			value=series,
200			type=list_type
201			)
202			)
203			session.add_all(sequences)
204			oemof_data = OemofData()
205			oemof_data.scalars = scalars
206			oemof_data.sequences = sequences
207			setattr(input_result, input_result_attr, oemof_data)
208			session.add(input_result)
209			session.flush()
210			result_id = input_result.input_result_id
211			session.commit()
212			return result_id
213
214
215			def restore_results(session, input_result_id):
216			"""
217			Restores input and result data from OemofInputResult from DB
218
219			Parameters
220			----------
221			session: sqlalchemy.session
222			SQLAlchemy session build via sqlalchemy.orm.sessionmaker
223			input_result_id: int
224			Index of OemofInputResult object to restore
225
226			Returns
227			-------
228			(dict, dict):
229			Restored input- and result-data
230			"""
231			def type_conversion(value_str, value_type):
232			if value_type == 'str':
233			return value_str
234			elif value_type == 'float':
235			return float(value_str)
236			elif value_type == 'int':
237			return int(value_str)
238			elif value_type == 'bool':
239			return bool(value_str)
240			else:
241			raise TypeError('Unknown conversion type "' + value_type + '"')
242
243			# Find results:
244			input_result = session.query(OemofInputResult).filter(
245			OemofInputResult.input_result_id == input_result_id).first()
246			if input_result is None:
247			raise IndexError(
248			'Could not find OemofInputResult with ID #' + str(input_result_id))
249
250			input_data = {}
251			result_data = {}
252			for input_result_attr, data in (
253			('input', input_data), ('result', result_data)):
254			ir_attr = getattr(input_result, input_result_attr)
255			for scalar in ir_attr.scalars:
256			nodes = (scalar.from_node, scalar.to_node)
257			if nodes not in data:
258			data[nodes] = {'scalars': {}, 'sequences': {}}
259			data[nodes]['scalars'][scalar.attribute] = type_conversion(
260			scalar.value, scalar.type)
261			for sequence in ir_attr.sequences:
262			nodes = (sequence.from_node, sequence.to_node)
263			if nodes not in data:
264			data[nodes] = {'scalars': {}, 'sequences': {}}
265			if sequence.type == 'series':
266			series = pandas.Series(sequence.value)
267			else:
268			series = sequence.value
269			data[nodes]['sequences'][sequence.attribute] = series
270			return input_data, result_data
271

oemof / oemof-db

db.results.restore_results() D last analyzed 2023-11-23 16:40 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

db.results.restore_results() D
last analyzed 2023-11-23 16:40 UTC