Source code for smo.dynamical_models.core.Fields

'''
Created on May 10, 2015

@author:  Atanas Pavlov
@copyright: SysMo Ltd, Bulgaria
'''
from smo.web.exceptions import ConnectionError, FieldError

[docs]class Causality(object): """ Defines information direction for a variable """ Parameter = 0 CalculatedParameter = 1 Input = 2 Output = 3 Local = 4 Independent = 5 RealState = 6 #TimeDerivative = 7
[docs]class Variability(object): """ Defines the variability of a variable """ Constant = 0 Fixed = 1 Tunable = 2 Discrete = 3 Continuous = 4
[docs]class ModelField(object): """ Abstract base class for all field types """ # Tracks each time an instance is created. Used to retain order. creation_counter = 0 def __init__(self, label = None, description = None): """ :param str label: text :param str description: verbose description """ self.label = label self.description = description # Increase the creation counter, and save our local copy. self.creation_counter = ModelField.creation_counter ModelField.creation_counter += 1
[docs] def setName(self, name): """ This method is explicitly called to set the field name :param name: field name """ self.name = name if (self.label is None): self.label = self.name if (self.description is None): self.description = self.label
[docs]class Function(ModelField): """ Represents a calculation function in the model """ def __init__(self, inputs, outputs, **kwargs): """ :param inputs: input variable list :param outputs: output variable list """ super(Function, self).__init__(**kwargs) self.inputs = inputs self.outputs = outputs # TODO_: check causalities def __call__(self, funcDef): self.funcDef = funcDef return self
[docs]class Port(ModelField): """ Port is a group of model variables exposed to the external world, used to connect to other models """ def __init__(self, variables, subType = None, **kwargs): ''' :param variables: list of the port variables :param subtype: derived classes can define special subtypes (e.g. C-port or R-port) ''' super(Port, self).__init__(**kwargs) self.variables = variables self.subType = subType
[docs] def checkConnect(self, other): ''' Perform basic check whether this port can be connected to the other one. Currently it checks: * the two ports have the same number of variables :param other: other port that this one is connected to :type other: :class:`Port` ''' if (len(self.variables) != len(other.variables)): raise ConnectionError('Cannot connect ports with different number of variables')
[docs]class ScalarVariable(ModelField): """ A fundamental variable field for the model """ def __init__(self, causality, variability, **kwargs): """ :param causality: causality :param variabilty: variability """ super(ScalarVariable, self).__init__(**kwargs) self.causality = causality self.variability = variability @property def default(self): raise NotImplementedError("{} does not implement 'default' property".format(self.__class__))
[docs]class RealVariable(ScalarVariable): """ Scalar variable of type real """ def __init__(self, causality = Causality.Local, variability = Variability.Continuous, default = 0., **kwargs): super(RealVariable, self).__init__(causality, variability, **kwargs) self.defaultValue = float(default) @property def default(self): return self.defaultValue
[docs]class RealState(RealVariable): """ Real variable which is a continuous state. Its value cannot be set directly by the model, but rather its derivative is calculated and then passed to the integrator, which computes the value over time. """ def __init__(self, start, **kwargs): super(RealState, self).__init__(causality = Causality.RealState, variability = Variability.Continuous, **kwargs) self.start = start @property def default(self): return self.start
class IntegerVariable(ScalarVariable): """ Scalar variable of type integer """ def __init__(self, causality = Causality.Local, variability = Variability.Discrete, default = 0, **kwargs): super(IntegerVariable, self).__init__(causality, variability, **kwargs) self.defaultValue = int(default) @property def default(self): return self.defaultValue class StateEvent(ModelField): """ """ def __init__(self, locate, **kwargs): super(StateEvent, self).__init__(**kwargs) self.locate = locate def __call__(self, update): self.update = update return self
[docs]class SubModel(ModelField): """ Submodel as part of a larger model """ def __init__(self, klass, **kwargs): """ :param klass: submodel class (must be a subclass of :class:`DynamicalModel`) """ super(SubModel, self).__init__(**kwargs) self.klass = klass
[docs]class InstanceField(object): """ Abstract base class for a field in a :class:`DynamicalModel` instance, corresponding to a field in the model definition. """ def __init__(self, modelInstance, clsVar): """ :param modelInstance: dynamical model instance :param clsVar: class variable corresponding to this instance variable defined in the dynamical model """ self.modelInstance = modelInstance self.clsVar = clsVar @property def name(self): """""" return self.clsVar.name @property def qPath(self): """""" return self.modelInstance.qPath + [self.name] @property def qName(self): """""" return '.'.join(self.qPath)
[docs]class InstanceVariable(InstanceField): """ Variable in a model instance """ def __init__(self, **kwargs): super(InstanceVariable, self).__init__(**kwargs) self.connectedVars = [] def connect(self, other, complement = True): # Input variables if (self.clsVar.causality == Causality.Input): if (len(self.connectedVars) != 0): raise ConnectionError(self, other, 'Cannot connect input {.qName} to more than one variable'.format(self)) elif (other.clsVar.causality == Causality.Output or other.clsVar.causality == Causality.RealState): self.connectedVars.append(other) if (complement): other.connect(self, complement = False) else: raise ConnectionError(self, other, 'Can only connect Input variable to an Output variable or RealState') # Output variables elif (self.clsVar.causality == Causality.Output): if (other.clsVar.causality == Causality.Input): if(other not in self.connectedVars): self.connectedVars.append(other) if (complement): other.connect(self, complement = False) else: pass else: raise ConnectionError(self, other, 'Can only connect Output variable to an Input variable') # State variables elif (self.clsVar.causality == Causality.RealState): if (other.clsVar.causality == Causality.Input): if(other not in self.connectedVars): self.connectedVars.append(other) if (complement): other.connect(self, complement = False) else: pass else: raise ConnectionError(self, other, 'Can only connect RealState variable to an input variable') # Other else: raise ConnectionError(self, other, 'Connected variables must have causality Input, Output or RealState') def setValue(self, value): setattr(self.modelInstance, self.clsVar.name, value) def getValue(self): return getattr(self.modelInstance, self.clsVar.name)
[docs]class InstanceFunction(InstanceField): """ Function in a model instance """ def __init__(self, **kwargs): super(InstanceFunction, self).__init__(**kwargs) self.inputs = [] self.outputs = [] for inVar in self.clsVar.inputs: self.inputs.append(self.modelInstance.meta.dm_variables[inVar.name]) for outVar in self.clsVar.outputs: self.outputs.append(self.modelInstance.meta.dm_variables[outVar.name])
[docs]class InstancePort(InstanceField): """ Port in a model instance """ def __init__(self, **kwargs): super(InstancePort, self).__init__(**kwargs) self.variables = [] for var in self.clsVar.variables: self.variables.append(self.modelInstance.meta.dm_variables[var.name]) def connect(self, other): self.clsVar.checkConnect(other.clsVar) for (thisVar, otherVar) in zip(self.variables, other.variables): thisVar.connect(otherVar)
[docs]class DerivativeVector(object): """ Derivative vector in a model instance """ def __init__(self, model): object.__setattr__(self, 'model', model) object.__setattr__(self, 'dm_realStates', model.__class__.dm_realStates) for name in self.dm_realStates.keys(): object.__setattr__(self, name, 0.) def __setattr__(self, name, value): if (name in self.dm_realStates.keys()): object.__setattr__(self, name, value) else: raise FieldError('No state derivative with name {}'.format(name))