9538747516

Committed 16 Jun 2024 08:36PM UTC coverage: 94.253% (-0.2%) from 94.444%

Build # 9538747516

Build Type

Pull #3

github

Committed by

WardLT

Commit Message

Update to Py3.10

It's what I've come to use for most projects,
and Py3.9 will be gone by the end of ROVI

Pull Request Pull Request #3: Progpy-inspired base class using Pydantic

Run Details

82 of 87 new or added lines in 2 files covered. (94.25%)

82 of 87 relevant lines covered (94.25%)

0.94 hits per line

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

92.65

/asoh/models/base.py

"""Base classes which define the state of a storage system,
the control signals applied to it, the outputs observable from it,
and the mathematical model which links state, control, and outputs together."""

import numpy as np
from scipy.integrate import solve_ivp
from pydantic import BaseModel, Field, model_validator


# TODO (wardlt): Consider an implementation where we store the parameters as a single numpy array, then provide helper classes for the parameters.
#  Maybe add some kind of `compile` which generates an object which provides such an interface (progpy does something like that)
class InstanceState(BaseModel, arbitrary_types_allowed=True):
    """Defines the state of a particular instance of a model

    Creating a InstanceState
    ------------------------

    Define a new instance state by adding the attributes which define the state
    of a particular model to a subclass of ``InstanceState``. List the names of
    attributes which always vary with time as the :attr:`state_params`.
    """

    health_params: tuple[str, ...] = Field(description='List of parameters which are being treated as state of health'
                                                       ' for this particular instance of a storage system', default_factory=tuple)

    covariance: np.ndarray = Field(None, description='Covariance matrix between all parameters being fit, including "health" and "state"')

    state_params: tuple[str, ...] = ...
    """Parameter which are always treated as time dependent"""

    @model_validator(mode='after')
    def _set_covariance(self):
        n_params = len(self.full_params)
        if self.covariance is None:
            self.covariance = np.eye(n_params)
        if self.covariance.shape != (n_params, n_params):
            raise ValueError(f'Expected ({n_params}, {n_params}) covariance matrix. Found {self.covariance.shape}')

    def _assemble_array(self, params: tuple[str, ...]) -> np.ndarray:
        """Assemble a numpy array from the instances within this class

        Args:
            params: Names of parameters to store
        Returns:
            A numpy array of the specified parameters
        """
        output = []
        for s in params:
            x = getattr(self, s)
            if isinstance(x, (float, int)):
                output.append(x)
            else:
                output.extend(x)
        return np.array(output)

    # TODO (wardlt): Generate names for parameters that are tuples/lists
    @property
    def full_params(self) -> tuple[str, ...]:
        """All parameters being adjusted by the model"""
        return self.state_params + self.health_params

    @property
    def state(self) -> np.ndarray:
        """Only the state of variables"""
        return self._assemble_array(self.state_params)

    @property
    def soh(self) -> np.ndarray:
        """Only the state of health variables"""
        return self._assemble_array(self.health_params)

    @property
    def full_state(self) -> np.ndarray:
        return self._assemble_array(self.state_params + self.health_params)

    def _update_params(self, x, params):
        """Update the parameters of parts of the state given the list of names and their new values"""
        param_iter = iter(x)
        for s in params:
            x = getattr(self, s)
            if isinstance(x, (float, int)):
                setattr(self, s, next(param_iter))
            else:
                p = [next(param_iter) for _ in x]
                setattr(self, s, p)

    def update_state(self, new_state: np.ndarray | list[float]):
        """Update this state to the values in the vector

        Args:
            new_state: New parameters
        """

        self._update_params(new_state, self.state_params)

    def update_soh(self, new_state: np.ndarray | list[float]):
        """Update the state of health values given a list of values

        Args:
            new_state: New parameters
        """

        self._update_params(new_state, self.health_params)

    def update_full_state(self, new_state: np.ndarray | list[float]):
        """Update the state and health parameters given a list of values

        Args:
            new_state: New parameters
        """

        self._update_params(new_state, self.full_params)


class ControlState(BaseModel):
    """The control of a battery system

    Add new fields to subclassess of ``ControlState`` for more complex systems
    """

    current: float = Field(description='Current applied to the storage system. Units: A')

    def to_numpy(self) -> np.ndarray:
        """Control inputs as a numpy vector"""
        output = [value for value in self.model_fields.values()]
        return np.array(output)


class Outputs(BaseModel):
    """Output model for observables from a battery system

    Add new fields to subclasses of ``ControlState`` for more complex systems
    """


# TODO (wardlt): Use generic classes? That might cement the relationship between a model and its associated input types
# TODO (warldt): Can we combine State and HealthModel? The State could contain parameters about how to perform updates
class HealthModel:
    """A model for a storage system which describes its operating state and health.

    Using a Health Model
    --------------------

    The health model implements tools which simulate using a storage system under specific control systems.
    Either call the :meth:`dx` function to estimate the rate of change of state parameters over time,
    or call :meth:`update` to simulate the change in state over a certain time period.

    Types of Parameters
    -------------------

    A storage system is described by many parameters differentiated by whether they remain static
    or change during operation.

    Static parameters define the design elements of a system that are not expected to degrade.

    Dynamic parameters could be those which change rapidly with operating conditions,
    like the state of charge, or those which change slowly with time, like the internal resistances.
    The quickly-varying parameters are called the **state** of the battery.
    The slowly-varying parameters are called the **state-of-health** of the battery.

    Implementing a Health Model
    ---------------------------

    First create the states which define your model as subclasses of the
    :class:`InstanceState`, :class:`ControlState`, and :class:`OutputModel`.

    We assume all models express dynamic systems which vary continuously with time.
    Implement the derivatives of each model parameter as a function of time
    as the :meth:`dx` function.

    Define the output function as :meth:`output` and how many outputs to expect as :attr:`num_outputs`
    """

    num_outputs: int = ...
    """Number of outputs from the output function"""

    def dx(self, state: InstanceState, control: ControlState) -> np.ndarray:
        """Compute the derivatives of each state variable with respect to time

        Args:
            state: State of the battery system
            control: Control signal applied to the system

        Returns:
            The derivative of each parameter with respect to time in units of "X per second"
        """
        raise NotImplementedError()

    def update(self, state: InstanceState, control: ControlState, total_time: float):
        """Update the state under the influence of a control variable for a certain amount of time

        Args:
            state: Starting state
            control: Control signal
            total_time: Amount of time to propagate
        """

        # Set up the time propagation function
        def _update_fun(t, y):
            state.update_state(y)
            return self.dx(state, control)

        result = solve_ivp(
            fun=_update_fun,
            y0=state.state,
            t_span=(0, total_time)
        )
        state.update_state(result.y[:, -1])

    def output(self, state: InstanceState, control: ControlState) -> Outputs:
        """Compute the observed outputs of a system given the current state and control

        Args:
            state: State of the battery system
            control: Control signal applied to the system

        Returns:
            Each observable of the battery system
        """
        raise NotImplementedError()

1	"""Base classes which define the state of a storage system,
2	the control signals applied to it, the outputs observable from it,
3	and the mathematical model which links state, control, and outputs together."""
4
5	import numpy as np	1✔
6	from scipy.integrate import solve_ivp	1✔
7	from pydantic import BaseModel, Field, model_validator	1✔
8
9
10	# TODO (wardlt): Consider an implementation where we store the parameters as a single numpy array, then provide helper classes for the parameters.
11	# Maybe add some kind of `compile` which generates an object which provides such an interface (progpy does something like that)
12	class InstanceState(BaseModel, arbitrary_types_allowed=True):	1✔
13	"""Defines the state of a particular instance of a model
14
15	Creating a InstanceState
16	------------------------
17
18	Define a new instance state by adding the attributes which define the state
19	of a particular model to a subclass of ``InstanceState``. List the names of
20	attributes which always vary with time as the :attr:`state_params`.
21	"""
22
23	health_params: tuple[str, ...] = Field(description='List of parameters which are being treated as state of health'	1✔
24	' for this particular instance of a storage system', default_factory=tuple)
25
26	covariance: np.ndarray = Field(None, description='Covariance matrix between all parameters being fit, including "health" and "state"')	1✔
27
28	state_params: tuple[str, ...] = ...	1✔
29	"""Parameter which are always treated as time dependent"""	1✔
30
31	@model_validator(mode='after')	1✔
32	def _set_covariance(self):	1✔
33	n_params = len(self.full_params)	1✔
34	if self.covariance is None:	1✔
35	self.covariance = np.eye(n_params)	1✔
36	if self.covariance.shape != (n_params, n_params):	1✔
NEW 37	raise ValueError(f'Expected ({n_params}, {n_params}) covariance matrix. Found {self.covariance.shape}')	×
38
39	def _assemble_array(self, params: tuple[str, ...]) -> np.ndarray:	1✔
40	"""Assemble a numpy array from the instances within this class
41
42	Args:
43	params: Names of parameters to store
44	Returns:
45	A numpy array of the specified parameters
46	"""
47	output = []	1✔
48	for s in params:	1✔
49	x = getattr(self, s)	1✔
50	if isinstance(x, (float, int)):	1✔
51	output.append(x)	1✔
52	else:
53	output.extend(x)	1✔
54	return np.array(output)	1✔
55
56	# TODO (wardlt): Generate names for parameters that are tuples/lists
57	@property	1✔
58	def full_params(self) -> tuple[str, ...]:	1✔
59	"""All parameters being adjusted by the model"""
60	return self.state_params + self.health_params	1✔
61
62	@property	1✔
63	def state(self) -> np.ndarray:	1✔
64	"""Only the state of variables"""
65	return self._assemble_array(self.state_params)	1✔
66
67	@property	1✔
68	def soh(self) -> np.ndarray:	1✔
69	"""Only the state of health variables"""
70	return self._assemble_array(self.health_params)	1✔
71
72	@property	1✔
73	def full_state(self) -> np.ndarray:	1✔
74	return self._assemble_array(self.state_params + self.health_params)	1✔
75
76	def _update_params(self, x, params):	1✔
77	"""Update the parameters of parts of the state given the list of names and their new values"""
78	param_iter = iter(x)	1✔
79	for s in params:	1✔
80	x = getattr(self, s)	1✔
81	if isinstance(x, (float, int)):	1✔
82	setattr(self, s, next(param_iter))	1✔
83	else:
84	p = [next(param_iter) for _ in x]	1✔
85	setattr(self, s, p)	1✔
86
87	def update_state(self, new_state: np.ndarray \| list[float]):	1✔
88	"""Update this state to the values in the vector
89
90	Args:
91	new_state: New parameters
92	"""
93
94	self._update_params(new_state, self.state_params)	1✔
95
96	def update_soh(self, new_state: np.ndarray \| list[float]):	1✔
97	"""Update the state of health values given a list of values
98
99	Args:
100	new_state: New parameters
101	"""
102
103	self._update_params(new_state, self.health_params)	1✔
104
105	def update_full_state(self, new_state: np.ndarray \| list[float]):	1✔
106	"""Update the state and health parameters given a list of values
107
108	Args:
109	new_state: New parameters
110	"""
111
112	self._update_params(new_state, self.full_params)	1✔
113
114
115	class ControlState(BaseModel):	1✔
116	"""The control of a battery system
117
118	Add new fields to subclassess of ``ControlState`` for more complex systems
119	"""
120
121	current: float = Field(description='Current applied to the storage system. Units: A')	1✔
122
123	def to_numpy(self) -> np.ndarray:	1✔
124	"""Control inputs as a numpy vector"""
NEW 125	output = [value for value in self.model_fields.values()]	×
NEW 126	return np.array(output)	×
127
128
129	class Outputs(BaseModel):	1✔
130	"""Output model for observables from a battery system
131
132	Add new fields to subclasses of ``ControlState`` for more complex systems
133	"""
134
135
136	# TODO (wardlt): Use generic classes? That might cement the relationship between a model and its associated input types
137	# TODO (warldt): Can we combine State and HealthModel? The State could contain parameters about how to perform updates
138	class HealthModel:	1✔
139	"""A model for a storage system which describes its operating state and health.
140
141	Using a Health Model
142	--------------------
143
144	The health model implements tools which simulate using a storage system under specific control systems.
145	Either call the :meth:`dx` function to estimate the rate of change of state parameters over time,
146	or call :meth:`update` to simulate the change in state over a certain time period.
147
148	Types of Parameters
149	-------------------
150
151	A storage system is described by many parameters differentiated by whether they remain static
152	or change during operation.
153
154	Static parameters define the design elements of a system that are not expected to degrade.
155
156	Dynamic parameters could be those which change rapidly with operating conditions,
157	like the state of charge, or those which change slowly with time, like the internal resistances.
158	The quickly-varying parameters are called the state of the battery.
159	The slowly-varying parameters are called the state-of-health of the battery.
160
161	Implementing a Health Model
162	---------------------------
163
164	First create the states which define your model as subclasses of the
165	:class:`InstanceState`, :class:`ControlState`, and :class:`OutputModel`.
166
167	We assume all models express dynamic systems which vary continuously with time.
168	Implement the derivatives of each model parameter as a function of time
169	as the :meth:`dx` function.
170
171	Define the output function as :meth:`output` and how many outputs to expect as :attr:`num_outputs`
172	"""
173
174	num_outputs: int = ...	1✔
175	"""Number of outputs from the output function"""	1✔
176
177	def dx(self, state: InstanceState, control: ControlState) -> np.ndarray:	1✔
178	"""Compute the derivatives of each state variable with respect to time
179
180	Args:
181	state: State of the battery system
182	control: Control signal applied to the system
183
184	Returns:
185	The derivative of each parameter with respect to time in units of "X per second"
186	"""
NEW 187	raise NotImplementedError()	×
188
189	def update(self, state: InstanceState, control: ControlState, total_time: float):	1✔
190	"""Update the state under the influence of a control variable for a certain amount of time
191
192	Args:
193	state: Starting state
194	control: Control signal
195	total_time: Amount of time to propagate
196	"""
197
198	# Set up the time propagation function
199	def _update_fun(t, y):	1✔
200	state.update_state(y)	1✔
201	return self.dx(state, control)	1✔
202
203	result = solve_ivp(	1✔
204	fun=_update_fun,
205	y0=state.state,
206	t_span=(0, total_time)
207	)
208	state.update_state(result.y[:, -1])	1✔
209
210	def output(self, state: InstanceState, control: ControlState) -> Outputs:	1✔
211	"""Compute the observed outputs of a system given the current state and control
212
213	Args:
214	state: State of the battery system
215	control: Control signal applied to the system
216
217	Returns:
218	Each observable of the battery system
219	"""
NEW 220	raise NotImplementedError()	×

ROVI-org / auto-soh / 9538747516

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