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

93.33

/src/spotpy/database/csv.py

import io
import sys
import time

import numpy as np

from .base import database


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

    def __init__(self, *args, **kwargs):
        # init base class
        super(csv, self).__init__(*args, **kwargs)
        # store init item only if dbinit
        if kwargs.get("dbappend", False) is False:
            print("* Database file '{}.csv' created.".format(self.dbname))
            # Create a open file, which needs to be closed after the sampling
            mode = "w"
            self.db = io.open(self.dbname + ".csv", mode)
            # write header line
            self.db.write(str(",".join(self.header) + "\n"))
        else:
            print("* Appending to database file '{}.csv'.".format(self.dbname))
            # Continues writing file
            mode = "a"
            self.db = io.open(self.dbname + ".csv", mode)

    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.write(",".join(map(str, coll)) + "\n")

        acttime = time.time()
        # Force writing to disc at least every two seconds
        if acttime - self.last_flush >= 2:
            self.db.flush()
            self.last_flush = time.time()

    def finalize(self):
        self.db.flush()  # Just to make sure that everything is written in file
        self.db.close()

    def getdata(self):
        data = np.genfromtxt(self.dbname + ".csv", delimiter=",", names=True)
        return data

1	import io	4✔
2	import sys	4✔
3	import time	4✔
4
5	import numpy as np	4✔
6
7	from .base import database	4✔
8
9
10	class csv(database):	4✔
11	"""
12	This class saves the process in the working storage. It can be used if
13	safety matters.
14	"""
15
16	def __init__(self, args, *kwargs):	4✔
17	# init base class
18	super(csv, self).__init__(args, *kwargs)	4✔
19	# store init item only if dbinit
20	if kwargs.get("dbappend", False) is False:	4✔
21	print("* Database file '{}.csv' created.".format(self.dbname))	4✔
22	# Create a open file, which needs to be closed after the sampling
23	mode = "w"	4✔
24	self.db = io.open(self.dbname + ".csv", mode)	4✔
25	# write header line
26	self.db.write(str(",".join(self.header) + "\n"))	4✔
27	else:
28	print("* Appending to database file '{}.csv'.".format(self.dbname))	4✔
29	# Continues writing file
30	mode = "a"	4✔
31	self.db = io.open(self.dbname + ".csv", mode)	4✔
32
33	def save(self, objectivefunction, parameterlist, simulations=None, chains=1):	4✔
34	coll = (	4✔
35	self.dim_dict["like"](objectivefunction)
36	+ self.dim_dict["par"](parameterlist)
37	+ self.dim_dict["simulation"](simulations)
38	+ [chains]
39	)
40	# Apply rounding of floats
41	coll = map(self.db_precision, coll)	4✔
42	self.db.write(",".join(map(str, coll)) + "\n")	4✔
43
44	acttime = time.time()	4✔
45	# Force writing to disc at least every two seconds
46	if acttime - self.last_flush >= 2:	4✔
47	self.db.flush()	×
48	self.last_flush = time.time()	×
49
50	def finalize(self):	4✔
51	self.db.flush() # Just to make sure that everything is written in file	4✔
52	self.db.close()	4✔
53
54	def getdata(self):	4✔
55	data = np.genfromtxt(self.dbname + ".csv", delimiter=",", names=True)	4✔
56	return data	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