#9249

Committed 19 Mar 2026 08:34PM UTC coverage: 24.649%. Remained the same

Build # #9249

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 41cd65108 into 4090d9eaa

Run Details

161 of 333 new or added lines in 46 files covered. (48.35%)

2821 existing lines in 36 files now uncovered.

12306 of 49925 relevant lines covered (24.65%)

0.25 hits per line

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

20.37

/seed/lib/progress_data/progress_data.py

"""
SEED Platform (TM), Copyright (c) Alliance for Energy Innovation, LLC, and other contributors.
See also https://github.com/SEED-platform/seed/blob/main/LICENSE.md
"""

import logging

from seed.decorators import get_prog_key
from seed.utils.cache import delete_cache, get_cache, set_cache

_log = logging.getLogger(__name__)


class ProgressData:
    def __init__(self, func_name, unique_id: str | int, init_data=None):
        self.func_name = func_name
        self.unique_id = unique_id
        self.key = get_prog_key(func_name, unique_id)
        self.total: int | None = None
        self.increment_by = None

        # Load in the initialized data, some of this may be overloaded based
        # on the contents in the cache
        self.initialize(init_data)

        # read the data from the cache, if there is any
        self.load()

    def initialize(self, init_data=None):
        if init_data:
            self.data = init_data
        else:
            self.data = {
                "status": "not-started",
                "status_message": "",
                "progress": 0,
                "progress_key": self.key,
                "unique_id": self.unique_id,
                "func_name": self.func_name,
                "message": None,
                "stacktrace": None,
                "summary": None,
            }
            self.total = None
            self.increment_by = None

        # set some member variables
        if "progress_key" in self.data:
            self.key = self.data["progress_key"]

        if "total" in self.data:
            self.total = self.data["total"]

        return self.save()

    def delete(self):
        """
        Delete the cache and reinitialize

        :return: dict, re-initialized data
        """
        delete_cache(self.key)

        return self.initialize()

    def finish_with_success(self, message=None):
        # update to get the latest results out of the cache
        self.load()

        self.data["status"] = "success"
        self.data["progress"] = 100
        self.data["message"] = message

        return self.save()

    def finish_with_warning(self, message=None):
        # update to get the latest results out of the cache
        self.load()

        _log.debug(f"Returning with warning from {self.key} with {message}")
        self.data["status"] = "warning"
        self.data["progress"] = 100
        self.data["message"] = message

        return self.save()

    def finish_with_error(self, message=None, stacktrace=None):
        # update to get the latest results out of the cache
        self.load()

        _log.debug(f"Returning with error from {self.key} with {message}")
        self.data["status"] = "error"
        self.data["progress"] = 100
        self.data["message"] = message
        self.data["stacktrace"] = stacktrace

        return self.save()

    @classmethod
    def from_key(cls, key):
        data = get_cache(key)
        if "func_name" in data and "unique_id" in data:
            return cls(func_name=data["func_name"], unique_id=data["unique_id"], init_data=data)
        else:
            raise Exception(f"Could not find key {key} in cache")

    def save(self):
        """Save the data to the cache"""
        # save some member variables
        self.data["total"] = self.total

        set_cache(self.key, self.data["status"], self.data)

        return get_cache(self.key)

    def load(self):
        """Read in the data from the cache"""

        # Merge the existing data with items from the cache, favor cache items
        self.data = dict(list(self.data.items()) + list(get_cache(self.key).items()))

        # set some member variables
        if self.data["progress_key"]:
            self.key = self.data["progress_key"]

        if self.data["total"]:
            self.total = self.data["total"]

    def step(self, status_message=None, new_summary=None):
        """Step the function by increment_value and save back to the cache"""
        # load the latest value out of the cache
        self.load()

        value = self.data["progress"]
        if value + self.increment_value() >= 100.0:
            value = 100.0
        else:
            value += self.increment_value()

        self.data["progress"] = value
        self.data["status"] = "parsing"
        if status_message is not None:
            self.data["status_message"] = status_message

        if new_summary is not None:
            self.data["summary"] = new_summary

        self.save()

        return self.result()

    def step_with_counter(self):
        """Step the function by increment_value and save back to the cache with a count"""
        # load the latest value out of the cache
        self.load()

        value = self.data["progress"]
        if value + self.increment_value() >= 100.0:
            value = 100.0
        else:
            value += self.increment_value()

        self.data["progress"] = value
        self.data["status"] = "running"
        self.data["completed_records"] = round(value / 100.0 * self.data["total_records"])
        self.data["status_message"] = f"{self.data['completed_records']:,} / {self.data['total_records']:,}"

        self.save()

        return self.result()

    def result(self):
        """
        Return the result from the cache

        :return: dict
        """
        return get_cache(self.key)

    def increment_value(self):
        """
        Return the value used to increment the progress. Currently, this always assumes that the
        size of the step is 1 / self.total count.

        :return: float, value to increment the step by
        """
        if self.total:
            return 1.0 / self.total * 100
        else:
            return 0

    def add_file_info(self, filename, info):
        """
        Add info for a file. Used when mapping xml data and errors/warnings are encountered.
        After mapping, the frontend checks for the 'file_info' attribute and displays
        the messages.
        """
        self.load()

        file_info = self.data.get("file_info", {})
        file_info[filename] = info
        self.data["file_info"] = file_info

        self.save()

    def update_summary(self, summary):
        self.data["summary"] = summary
        self.save()

    def summary(self):
        """
        Return the summary data of the progress key
        :return: dict
        """
        # read the latest from the cache
        self.load()
        return self.data["summary"]

1	"""
2	SEED Platform (TM), Copyright (c) Alliance for Energy Innovation, LLC, and other contributors.
3	See also https://github.com/SEED-platform/seed/blob/main/LICENSE.md
4	"""
5
6	import logging	1✔
7
8	from seed.decorators import get_prog_key	1✔
9	from seed.utils.cache import delete_cache, get_cache, set_cache	1✔
10
11	_log = logging.getLogger(__name__)	1✔
12
13
14	class ProgressData:	1✔
15	def __init__(self, func_name, unique_id: str \| int, init_data=None):	1✔
16	self.func_name = func_name	×
17	self.unique_id = unique_id	×
18	self.key = get_prog_key(func_name, unique_id)	×
NEW 19	self.total: int \| None = None	×
UNCOV 20	self.increment_by = None	×
21
22	# Load in the initialized data, some of this may be overloaded based
23	# on the contents in the cache
UNCOV 24	self.initialize(init_data)	×
25
26	# read the data from the cache, if there is any
UNCOV 27	self.load()	×
28
29	def initialize(self, init_data=None):	1✔
30	if init_data:	×
UNCOV 31	self.data = init_data	×
32	else:
UNCOV 33	self.data = {	×
34	"status": "not-started",
35	"status_message": "",
36	"progress": 0,
37	"progress_key": self.key,
38	"unique_id": self.unique_id,
39	"func_name": self.func_name,
40	"message": None,
41	"stacktrace": None,
42	"summary": None,
43	}
44	self.total = None	×
UNCOV 45	self.increment_by = None	×
46
47	# set some member variables
48	if "progress_key" in self.data:	×
UNCOV 49	self.key = self.data["progress_key"]	×
50
51	if "total" in self.data:	×
UNCOV 52	self.total = self.data["total"]	×
53
UNCOV 54	return self.save()	×
55
56	def delete(self):	1✔
57	"""
58	Delete the cache and reinitialize
59
60	:return: dict, re-initialized data
61	"""
UNCOV 62	delete_cache(self.key)	×
63
UNCOV 64	return self.initialize()	×
65
66	def finish_with_success(self, message=None):	1✔
67	# update to get the latest results out of the cache
UNCOV 68	self.load()	×
69
70	self.data["status"] = "success"	×
71	self.data["progress"] = 100	×
UNCOV 72	self.data["message"] = message	×
73
UNCOV 74	return self.save()	×
75
76	def finish_with_warning(self, message=None):	1✔
77	# update to get the latest results out of the cache
UNCOV 78	self.load()	×
79
80	_log.debug(f"Returning with warning from {self.key} with {message}")	×
81	self.data["status"] = "warning"	×
82	self.data["progress"] = 100	×
UNCOV 83	self.data["message"] = message	×
84
UNCOV 85	return self.save()	×
86
87	def finish_with_error(self, message=None, stacktrace=None):	1✔
88	# update to get the latest results out of the cache
UNCOV 89	self.load()	×
90
91	_log.debug(f"Returning with error from {self.key} with {message}")	×
92	self.data["status"] = "error"	×
93	self.data["progress"] = 100	×
94	self.data["message"] = message	×
UNCOV 95	self.data["stacktrace"] = stacktrace	×
96
UNCOV 97	return self.save()	×
98
99	@classmethod	1✔
100	def from_key(cls, key):	1✔
101	data = get_cache(key)	×
102	if "func_name" in data and "unique_id" in data:	×
UNCOV 103	return cls(func_name=data["func_name"], unique_id=data["unique_id"], init_data=data)	×
104	else:
UNCOV 105	raise Exception(f"Could not find key {key} in cache")	×
106
107	def save(self):	1✔
108	"""Save the data to the cache"""
109	# save some member variables
UNCOV 110	self.data["total"] = self.total	×
111
UNCOV 112	set_cache(self.key, self.data["status"], self.data)	×
113
UNCOV 114	return get_cache(self.key)	×
115
116	def load(self):	1✔
117	"""Read in the data from the cache"""
118
119	# Merge the existing data with items from the cache, favor cache items
UNCOV 120	self.data = dict(list(self.data.items()) + list(get_cache(self.key).items()))	×
121
122	# set some member variables
123	if self.data["progress_key"]:	×
UNCOV 124	self.key = self.data["progress_key"]	×
125
126	if self.data["total"]:	×
UNCOV 127	self.total = self.data["total"]	×
128
129	def step(self, status_message=None, new_summary=None):	1✔
130	"""Step the function by increment_value and save back to the cache"""
131	# load the latest value out of the cache
UNCOV 132	self.load()	×
133
134	value = self.data["progress"]	×
135	if value + self.increment_value() >= 100.0:	×
UNCOV 136	value = 100.0	×
137	else:
UNCOV 138	value += self.increment_value()	×
139
140	self.data["progress"] = value	×
141	self.data["status"] = "parsing"	×
142	if status_message is not None:	×
UNCOV 143	self.data["status_message"] = status_message	×
144
145	if new_summary is not None:	×
UNCOV 146	self.data["summary"] = new_summary	×
147
UNCOV 148	self.save()	×
149
UNCOV 150	return self.result()	×
151
152	def step_with_counter(self):	1✔
153	"""Step the function by increment_value and save back to the cache with a count"""
154	# load the latest value out of the cache
UNCOV 155	self.load()	×
156
157	value = self.data["progress"]	×
158	if value + self.increment_value() >= 100.0:	×
UNCOV 159	value = 100.0	×
160	else:
UNCOV 161	value += self.increment_value()	×
162
163	self.data["progress"] = value	×
164	self.data["status"] = "running"	×
165	self.data["completed_records"] = round(value / 100.0 * self.data["total_records"])	×
UNCOV 166	self.data["status_message"] = f"{self.data['completed_records']:,} / {self.data['total_records']:,}"	×
167
UNCOV 168	self.save()	×
169
UNCOV 170	return self.result()	×
171
172	def result(self):	1✔
173	"""
174	Return the result from the cache
175
176	:return: dict
177	"""
UNCOV 178	return get_cache(self.key)	×
179
180	def increment_value(self):	1✔
181	"""
182	Return the value used to increment the progress. Currently, this always assumes that the
183	size of the step is 1 / self.total count.
184
185	:return: float, value to increment the step by
186	"""
187	if self.total:	×
UNCOV 188	return 1.0 / self.total * 100	×
189	else:
UNCOV 190	return 0	×
191
192	def add_file_info(self, filename, info):	1✔
193	"""
194	Add info for a file. Used when mapping xml data and errors/warnings are encountered.
195	After mapping, the frontend checks for the 'file_info' attribute and displays
196	the messages.
197	"""
UNCOV 198	self.load()	×
199
200	file_info = self.data.get("file_info", {})	×
201	file_info[filename] = info	×
UNCOV 202	self.data["file_info"] = file_info	×
203
UNCOV 204	self.save()	×
205
206	def update_summary(self, summary):	1✔
207	self.data["summary"] = summary	×
UNCOV 208	self.save()	×
209
210	def summary(self):	1✔
211	"""
212	Return the summary data of the progress key
213	:return: dict
214	"""
215	# read the latest from the cache
216	self.load()	×
UNCOV 217	return self.data["summary"]	×

SEED-platform / seed / #9249

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