7709518362

Committed 30 Jan 2024 10:11AM UTC coverage: 77.332% (-0.2%) from 77.518%

Build # 7709518362

Build Type

push

github

Committed by

thouska

Commit Message

Merge branch 'master' of https://github.com/thouska/spotpy

Run Details

4162 of 5382 relevant lines covered (77.33%)

3.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.24

/src/spotpy/database/sql.py

import sqlite3
import sys

import numpy as np

from .base import database


class PickalableSWIG:
    def __setstate__(self, state):
        self.__init__(*state["args"])

    def __getstate__(self):
        return {"args": self.args}


class PickalableSQL3Connect(sqlite3.Connection, PickalableSWIG):
    def __init__(self, *args, **kwargs):
        self.args = args
        sqlite3.Connection.__init__(self, *args, **kwargs)


class PickalableSQL3Cursor(sqlite3.Cursor, PickalableSWIG):
    def __init__(self, *args, **kwargs):
        self.args = args
        sqlite3.Cursor.__init__(self, *args, **kwargs)


class sql(database):
    """
    This class saves the process in the working storage. It can be used if
    safety matters.
    """

    def __init__(self, *args, **kwargs):
        import os

        # init base class
        super(sql, self).__init__(*args, **kwargs)
        # Create a open file, which needs to be closed after the sampling
        try:
            os.remove(self.dbname + ".db")
        except:
            pass

        self.db = PickalableSQL3Connect(self.dbname + ".db")
        self.db_cursor = PickalableSQL3Cursor(self.db)
        # Create Table
        #        self.db_cursor.execute('''CREATE TABLE IF NOT EXISTS  '''+self.dbname+'''
        #                     (like1 real, parx real, pary real, simulation1 real, chain int)''')
        self.db_cursor.execute(
            """CREATE TABLE IF NOT EXISTS  """
            + self.dbname
            + """
                     ("""
            + " real ,".join(self.header)
            + """)"""
        )

    def save(self, objectivefunction, parameterlist, simulations=None, chains=1):
        coll = (
            self.dim_dict["like"](objectivefunction)
            + self.dim_dict["par"](parameterlist)
            + self.dim_dict["simulation"](simulations)
            + [chains]
        )
        # Apply rounding of floats
        coll = map(self.db_precision, coll)
        self.db_cursor.execute(
            "INSERT INTO "
            + self.dbname
            + " VALUES ("
            + '"'
            + str('","'.join(map(str, coll)))
            + '"'
            + ")"
        )

        self.db.commit()

    def finalize(self):
        self.db.close()

    def getdata(self):
        self.db = PickalableSQL3Connect(self.dbname + ".db")
        self.db_cursor = PickalableSQL3Cursor(self.db)

        headers = [
            (row[1], "<f8")
            for row in self.db_cursor.execute("PRAGMA table_info(" + self.dbname + ");")
        ]

        back = np.array(
            [row for row in self.db_cursor.execute("SELECT * FROM " + self.dbname)],
            dtype=headers,
        )

        self.db.close()
        return back

1	import sqlite3	4✔
2	import sys	4✔
3
4	import numpy as np	4✔
5
6	from .base import database	4✔
7
8
9	class PickalableSWIG:	4✔
10	def __setstate__(self, state):	4✔
11	self.__init__(*state["args"])	×
12
13	def __getstate__(self):	4✔
14	return {"args": self.args}	×
15
16
17	class PickalableSQL3Connect(sqlite3.Connection, PickalableSWIG):	4✔
18	def __init__(self, args, *kwargs):	4✔
19	self.args = args	4✔
20	sqlite3.Connection.__init__(self, args, *kwargs)	4✔
21
22
23	class PickalableSQL3Cursor(sqlite3.Cursor, PickalableSWIG):	4✔
24	def __init__(self, args, *kwargs):	4✔
25	self.args = args	4✔
26	sqlite3.Cursor.__init__(self, args, *kwargs)	4✔
27
28
29	class sql(database):	4✔
30	"""
31	This class saves the process in the working storage. It can be used if
32	safety matters.
33	"""
34
35	def __init__(self, args, *kwargs):	4✔
36	import os	4✔
37
38	# init base class
39	super(sql, self).__init__(args, *kwargs)	4✔
40	# Create a open file, which needs to be closed after the sampling
41	try:	4✔
42	os.remove(self.dbname + ".db")	4✔
43	except:	4✔
44	pass	4✔
45
46	self.db = PickalableSQL3Connect(self.dbname + ".db")	4✔
47	self.db_cursor = PickalableSQL3Cursor(self.db)	4✔
48	# Create Table
49	# self.db_cursor.execute('''CREATE TABLE IF NOT EXISTS '''+self.dbname+'''
50	# (like1 real, parx real, pary real, simulation1 real, chain int)''')
51	self.db_cursor.execute(	4✔
52	"""CREATE TABLE IF NOT EXISTS """
53	+ self.dbname
54	+ """
55	("""
56	+ " real ,".join(self.header)
57	+ """)"""
58	)
59
60	def save(self, objectivefunction, parameterlist, simulations=None, chains=1):	4✔
61	coll = (	4✔
62	self.dim_dict["like"](objectivefunction)
63	+ self.dim_dict["par"](parameterlist)
64	+ self.dim_dict["simulation"](simulations)
65	+ [chains]
66	)
67	# Apply rounding of floats
68	coll = map(self.db_precision, coll)	4✔
69	self.db_cursor.execute(	4✔
70	"INSERT INTO "
71	+ self.dbname
72	+ " VALUES ("
73	+ '"'
74	+ str('","'.join(map(str, coll)))
75	+ '"'
76	+ ")"
77	)
78
79	self.db.commit()	4✔
80
81	def finalize(self):	4✔
82	self.db.close()	4✔
83
84	def getdata(self):	4✔
85	self.db = PickalableSQL3Connect(self.dbname + ".db")	4✔
86	self.db_cursor = PickalableSQL3Cursor(self.db)	4✔
87
88	headers = [	4✔
89	(row[1], "<f8")
90	for row in self.db_cursor.execute("PRAGMA table_info(" + self.dbname + ");")
91	]
92
93	back = np.array(	4✔
94	[row for row in self.db_cursor.execute("SELECT * FROM " + self.dbname)],
95	dtype=headers,
96	)
97
98	self.db.close()	4✔
99	return back	4✔

thouska / spotpy / 7709518362

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous