Coverage for physioblocks / library / functions / watchers.py: 94%

1# SPDX-FileCopyrightText: Copyright INRIA 

2# 

3# SPDX-License-Identifier: LGPL-3.0-only 

4# 

5# Copyright INRIA 

6# 

7# This file is part of PhysioBlocks, a library mostly developed by the 

8# [Ananke project-team](https://team.inria.fr/ananke) at INRIA. 

9# 

10# Authors: 

11# - Colin Drieu 

12# - Dominique Chapelle 

13# - François Kimmig 

14# - Philippe Moireau 

15# 

16# PhysioBlocks is free software: you can redistribute it and/or modify it under the 

17# terms of the GNU Lesser General Public License as published by the Free Software 

18# Foundation, version 3 of the License. 

19# 

20# PhysioBlocks is distributed in the hope that it will be useful, but WITHOUT ANY 

21# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A 

22# PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. 

23# 

24# You should have received a copy of the GNU Lesser General Public License along with 

25# PhysioBlocks. If not, see <https://www.gnu.org/licenses/>. 

26 

27""" 

28Watcher are useful functions to use to set **Output Functions** 

29in the configuration. 

30""" 

31 

32from __future__ import annotations 

33 

34from collections.abc import Iterable 

35from dataclasses import dataclass, field 

36from typing import Any 

37 

38import numpy as np 

39 

40from physioblocks.base.operators import AbstractBaseOperators 

41from physioblocks.computing.models import Block 

42from physioblocks.computing.quantities import Quantity 

43from physioblocks.registers.type_register import register_type 

44from physioblocks.simulation import AbstractFunction 

45 

46# Quantity Watcher function type name 

47WATCH_QUANTITY_TYPE_ID = "watch_quantity" 

48 

49 

50@dataclass 

51@register_type(WATCH_QUANTITY_TYPE_ID) 

52class WatchQuantity(AbstractFunction, AbstractBaseOperators): 

53 """ 

54 Return the current value of the watch quantity. 

55 """ 

56 

57 quantity: Quantity[Any] 

58 """The quantities to watch""" 

59 

60 @property 

61 def operation_value(self) -> Any: 

62 """ 

63 Value used with base operators. 

64 

65 :return: the current evaluation of the function 

66 :rtype: Any 

67 """ 

68 return self.eval() 

69 

70 def eval(self) -> Any: 

71 """Just return the watched quantity current value 

72 

73 :return: the watched quantity current value 

74 :rtype: np.float64 | np.ndarray""" 

75 return self.quantity.current 

76 

77 

78# Sum Blocks Quantity function type name 

79SUM_BLOCKS_QUANTITY_TYPE_ID = "sum_blocks_quantity" 

80 

81 

82@dataclass 

83@register_type(SUM_BLOCKS_QUANTITY_TYPE_ID) 

84class SumBlocksQuantity(AbstractFunction, AbstractBaseOperators): 

85 """Sum the current values of quantities with the given name 

86 for each the blocks.""" 

87 

88 quantity_id: str 

89 """block local id of the quantity to sum""" 

90 

91 elements: Iterable[Block] = field(default_factory=list) 

92 """the blocks holding the quantity to sum""" 

93 

94 @property 

95 def operation_value(self) -> Any: 

96 """ 

97 Value used with base operators. 

98 

99 :return: the current evaluation of the function 

100 :rtype: Any 

101 """ 

102 return self.eval() 

103 

104 def eval(self) -> Any: 

105 """ 

106 Sum the quantities 

107 

108 :return: the sum of the quantities 

109 :rtype: Any 

110 """ 

111 return np.sum( 

112 [getattr(block, self.quantity_id).current for block in self.elements] 

113 )