pingu_98/cara
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add names states for the Ventilation schemes, allowing their state to be persisted even if that scheme isn't currently active.

Browse source
This commit is contained in:
Phil Elson 2020-10-26 11:12:16 +01:00
parent a95181612b
commit 9037e68c31
4 changed files with 239 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
cara/apps.py
View file

@ -1,3 +1,4 @@
import dataclasses
import typing
import uuid
@ -239,8 +240,8 @@ class WidgetView:
def _build_ventilation(self, node):
ventilation_widgets = {
'Natural': self._build_window(node),
# 'HEPA': self._build_hepa(node)
'Natural': self._build_window(node._states['Natural']),
'HEPA': self._build_hepa(node._states['HEPA']),
}
for name, widget in ventilation_widgets.items():
widget.layout.visible = False
@ -252,10 +253,9 @@ class WidgetView:
def toggle_ventilation(value):
for name, widget in ventilation_widgets.items():
widget.layout.display = 'none'
# if value == 'Natural':
# node.dcs_update_from(models.PeriodicWindow())
# elif value == 'HEPA':
# node.dcs_update_from(models.PeriodicHEPA())
node.dcs_select(value)
widget = ventilation_widgets[value]
widget.layout.visible = True
widget.layout.display = 'block'
@ -293,9 +293,34 @@ baseline_model = models.Model(
)
class CARAStateBuilder(state.StateBuilder):
def build_type_Mask(self, _: dataclasses.Field):
return state.DataclassStatePredefined(
models.Mask,
choices=models.Mask.types,
)
def build_type_Ventilation(self, _: dataclasses.Field):
s = state.DataclassStateNamed(
states={
'Natural': self.build_generic(models.PeriodicWindow),
'HEPA': self.build_generic(models.PeriodicHEPA),
},
state_builder=self,
)
# Initialise the HEPA state
s._states['HEPA'].dcs_update_from(
models.PeriodicHEPA(120, 120, 500.)
)
return s
class ExpertApplication:
def __init__(self):
self.model_state = state.DataclassState(models.Model)
self.model_state = state.DataclassInstanceState(
models.Model,
state_builder=CARAStateBuilder(),
)
self.model_state.dcs_update_from(baseline_model)
self.view = WidgetView(self.model_state)

3
cara/models.py
View file

@ -70,6 +70,9 @@ class PeriodicHEPA(Ventilation):
def air_exchange(self, room: Room, time: float) -> float:
# Returns the rate at which air is being exchanged in the given room per cubic meter at a given time
period = self.period / 60.
duration = self.duration / 60.
# If the HEPA is off, no air is being exchanged
if time % self.period < (self.period - self.duration):
return 0

158
cara/state.py
View file

@ -35,10 +35,75 @@ class StateBuilder:
return self.build_generic
def build_generic(self, type_to_build: typing.Type):
return DataclassState(type_to_build, self)
return DataclassInstanceState(type_to_build, state_builder=self)
class DataclassState:
def __init__(self, state_builder=StateBuilder()):
with self._object_setattr():
self._state_builder = state_builder
@contextmanager
def _object_setattr(self):
"""
For the lifetime of this contextmanager, don't do anything other than
standard object.__setattr__ when setting attributes.
"""
object.__setattr__(self, '_use_base_setattr', True)
yield
object.__setattr__(self, '_use_base_setattr', False)
def dcs_instance(self):
"""
Return the instance that this state represents. The instance returned
is immutable, so it is advised to call this method each time that
you want the instance so that it reflects the most up-to-date state.
"""
pass
def dcs_observe(self, callback: typing.Callable):
"""
If any changes are made to the state, call the given callback.
"""
pass
@contextmanager
def dcs_state_transaction(self):
"""
For the lifetime of this context manager, do not fire observer
notifications. If any notifications would have been fired during the
lifetime of this context manager, then an event will be fired once
exiting the context.
"""
yield
def dcs_update_from(self, data: dataclass_instance):
"""
Update the state based on the values of the given dataclass instance.
"""
pass
def _dcs_set_value(self, attr_name, value):
"""
Set the state of the given attribute to the given value.
"""
def dcs_set_instance_type(self, instance_dataclass: Datamodel_T):
"""
Update the current instance of the state to this type.
Note: This currently wipes all downstream observers.
"""
class DataclassInstanceState(DataclassState):
"""
Represents the state of a frozen dataclass.
No type checking of the attributes is attempted.
@ -58,6 +123,8 @@ class DataclassState:
"""
def __init__(self, dataclass: Datamodel_T, state_builder=StateBuilder()):
super().__init__(state_builder=state_builder)
# Note that the constructor does *not* insert any data by default. It
# therefore doesn't build nested DataclassState instances when a dataclass contains another.
# For that, use the build classmethod.
@ -124,15 +191,9 @@ class DataclassState:
for observer in self._observers:
observer()
@contextmanager
def _object_setattr(self):
self._use_base_setattr = True
yield
self._use_base_setattr = False
def __getattr__(self, name):
try:
return super().__getattribute__(name)
return object.__getattribute__(self, name)
except AttributeError:
pass
if name in self._data:
@ -171,6 +232,7 @@ class DataclassState:
raise TypeError(f"The dataclass type provided ({instance_dataclass}) must be a subclass of the base ({self._base})")
self._instance_type = instance_dataclass
# TODO: It is possible to cut observer connections by clearing like this.
self._data.clear()
for field in dataclasses.fields(instance_dataclass):
if dataclasses.is_dataclass(field.type):
@ -204,7 +266,7 @@ class DataclassState:
return self._instance
class DataclassStatePredefined(DataclassState):
class DataclassStatePredefined(DataclassInstanceState):
"""
Only a pre-defined selection of states for the given type are allowed.
Selected by name (the keys in the dictionary).
@ -214,8 +276,12 @@ class DataclassStatePredefined(DataclassState):
state.dcs_select(name)
"""
def __init__(self, dataclass: Datamodel_T, choices: typing.Dict[typing.Hashable, dataclass_instance]):
super().__init__(dataclass=dataclass)
def __init__(self,
dataclass: Datamodel_T,
choices: typing.Dict[typing.Hashable, dataclass_instance],
**kwargs,
):
super().__init__(dataclass=dataclass, **kwargs)
with self._object_setattr():
self._choices = choices
@ -235,11 +301,75 @@ class DataclassStatePredefined(DataclassState):
def __repr__(self):
return f"<state for {self._instance_type.__name__}. '{self._selected}' selected>"
def _instance_kwargs(self):
raise NotImplementedError("Doesn't make much sense")
def _instance_state(self):
return dataclasses.asdict(self.dcs_instance())
def _instance_kwargs(self):
return dataclasses.asdict(self.dcs_instance())
class DataclassStateNamed(DataclassState):
"""
A collection of instances of the given type, switchable by name, but each
instance is still mutable.
"""
def __init__(self,
states: typing.Dict[typing.Hashable, DataclassState],
**kwargs
):
# TODO: This is effectively a container type. We shouldn't use the standard constructor for this.
enabled = list(states.keys())[0]
t = states[enabled]
super().__init__(**kwargs)
with self._object_setattr():
self._states = states.copy()
self._selected = None
# Pick the first choice until we know otherwise.
self.dcs_select(enabled)
def __getattr__(self, name):
try:
return object.__getattribute__(self, name)
except AttributeError:
pass
return getattr(self._selected_state(), name)
# if name in self._data:
# return self._data[name]
# elif name in self._instance_attrs():
# raise ValueError(f"State not yet set for {name}")
# else:
# raise AttributeError(f"Attribute {name} does not exist on {self._instance_type.__name__}")
def __setattr__(self, name, value):
if name in self.__dict__ or self.__dict__.get('_use_base_setattr', True):
return object.__setattr__(self, name, value)
setattr(self._selected_state(), name, value)
def dcs_select(self, name: typing.Hashable):
if name not in self._states:
raise ValueError(f'The choice {name} is not valid. Possible options are {", ".join(self._states)}')
self._selected = name
self._selected_state()._fire_observers()
def _selected_state(self):
return self._states[self._selected]
def dcs_instance(self):
return self._selected_state().dcs_instance()
def __repr__(self):
return f"<state for {self._instance_type.__name__}. Holding {len(self._states)} state(s). '{self._selected}' selected>"
def dcs_observe(self, callback: typing.Callable):
# Note there is no way to observe the selected state change currently.
# You can only watch for the individual selected states being changed.
for state in self._states.values():
state.dcs_observe(callback)
def dcs_update_from(self, data: dataclass_instance):
return self._selected_state().dcs_update_from(data)
def dcs_set_instance_type(self, instance_dataclass: Datamodel_T):
return self._selected_state().dcs_set_instance_type(instance_dataclass)

72
cara/tests/test_state.py
View file

@ -13,6 +13,11 @@ class DCSimple:
attr2: int
@dataclass
class DCAnother:
attr3: float
@dataclass
class DCSimpleSubclass(DCSimple):
attr3: float
@ -51,24 +56,24 @@ def dc_simple():
def test_DCS_construct():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
assert repr(s) == '<state for DCSimple(**{})>'
with pytest.raises(TypeError, match=r"A dataclass type must be provided, not an instance of one"):
state.DataclassState(DCSimple('', 1))
state.DataclassInstanceState(DCSimple('', 1))
with pytest.raises(TypeError, match="The given class is not a valid dataclass"):
state.DataclassState(None)
state.DataclassInstanceState(None)
def test_DCS_construct_nested():
s = state.DataclassState(DCNested)
s = state.DataclassInstanceState(DCNested)
assert repr(s) == "<state for DCNested(**{'simple': {}})>"
@pytest.mark.xfail
def test_DCS_subclass():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
s.dcs_set_instance_type(DCSimpleSubclass)
s.set('attr3', 3.14)
assert s._instance_kwargs() == {'attr3': 3.14}
@ -78,35 +83,35 @@ def test_DCS_subclass():
def test_DCS_setattr():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
s.attr1 = 'Hello world'
assert s._instance_kwargs() == {'attr1': 'Hello world'}
@pytest.mark.xfail
def test_DCS_type_check():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
with pytest.raises(TypeError):
# TODO: Should we make this fail? It involves type-checking / validation.
s.attr1 = 1
def test_DCS_update_from_instance():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
s.dcs_update_from(DCSimple('a1', 2))
assert s._instance_type == DCSimple
assert s._instance_kwargs() == {'attr1': 'a1', 'attr2': 2}
def test_DCS_update_from_instance_subclass():
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
s.dcs_update_from(DCSimpleSubclass('a1', 2, 3.14))
assert s._instance_type == DCSimpleSubclass
assert s._instance_kwargs() == {'attr1': 'a1', 'attr2': 2, 'attr3': 3.14}
def test_DCS_update_from_instance_nested():
s = state.DataclassState(DCNested)
s = state.DataclassInstanceState(DCNested)
nested = DCNested(DCSimpleSubclass('a1', 2, 3.14), [])
s.dcs_update_from(nested)
assert s.simple.dcs_instance() == nested.simple
@ -117,7 +122,7 @@ def test_observe_instance_nested():
top_level = Mock()
nested = Mock()
s = state.DataclassState(DCNested)
s = state.DataclassInstanceState(DCNested)
s.dcs_observe(top_level)
s.simple.dcs_observe(nested)
@ -154,8 +159,51 @@ def test_DCS_predefined():
assert s.dcs_instance() == opt2
def test_DCS_named():
opt1 = DCSimpleSubclass('a', 1, 3.14)
opt2 = DCAnother(4.2)
s = state.DataclassStateNamed({
# Entirely different types possible.
'option 1': state.DataclassInstanceState(DCSimple),
'option 2': state.DataclassInstanceState(DCAnother),
})
assert s._selected == 'option 1'
with pytest.raises(ValueError):
s.dcs_select('option 3')
with pytest.raises(TypeError):
# Not initialised all the values yet...
s.dcs_instance()
opt1_observer = Mock()
s.dcs_observe(opt1_observer)
s.dcs_update_from(opt1)
assert s.dcs_instance() == opt1
opt1_observer.assert_called_once_with()
opt1_observer.reset_mock()
with pytest.raises(TypeError):
s.dcs_update_from(opt2)
s.dcs_select('option 2')
s.dcs_update_from(opt2)
assert s.dcs_instance() == opt2
# We can't observe individual states directly.
opt1_observer.assert_called_once_with()
s.dcs_select('option 1')
assert s.dcs_instance() == opt1
# TODO: This should fail.
s.foo = 10
def test_DCS_non_dataclass_attrs():
val = DCClassVar('a', 1)
s = state.DataclassState(DCSimple)
s = state.DataclassInstanceState(DCSimple)
s.dcs_update_from(val)
s.dcs_instance() == val