Skip to content

Risk Assessment API

Toolkit for hazard modeling, agent effects, and protection policy evaluation.

RiskToolkit

hera.riskassessment.riskToolkit.RiskToolkit

Bases: abstractToolkit

Toolkit for agent-based risk assessment.

Manages hazardous agents (chemical, thermal, blast), their injury effect models, protection policies, and casualty estimation. Agents are stored as versioned data sources and can be loaded from JSON descriptors or from the database.

Source code in hera/riskassessment/riskToolkit.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class RiskToolkit(abstractToolkit):
    """
    Toolkit for agent-based risk assessment.

    Manages hazardous agents (chemical, thermal, blast), their injury
    effect models, protection policies, and casualty estimation.
    Agents are stored as versioned data sources and can be loaded
    from JSON descriptors or from the database.
    """
    _presentation = None
    _protectionPolicy = None
    _analysis = None

    @property
    def analysis(self):
        """
        Access the risk analysis layer.

        Returns
        -------
        analysis
            Object providing risk area calculation and LSM integration.
        """
        return self._analysis

    @property
    def ProtectionPolicy(self):
        """
        Access the ProtectionPolicy class for building protection action pipelines.

        Returns
        -------
        type
            The ProtectionPolicy class (not an instance).
        """
        return self._protectionPolicy

    @property
    def presentation(self):
        """
        Access the presentation layer for casualty visualizations.

        Returns
        -------
        casualtiesPlot
        """
        return self._presentation

    def __init__(self, projectName, filesDirectory=None, connectionName=None):
        """
        Initialize the RiskToolkit.

        Parameters
        ----------
        projectName : str
            The project name.
        filesDirectory : str, optional
            Directory for file outputs.
        connectionName : str, optional
            The DB connection name.
        """
        super().__init__(projectName=projectName, filesDirectory=filesDirectory, toolkitName="RiskAssessment", connectionName=connectionName)
        self._presentation = casualtiesPlot()
        self._protectionPolicy = ProtectionPolicy
        self._analysis = analysis(self)

    def getAgent(self, nameOrDesc, version=None):
        """
            Initialize the agents.

        :param nameOrDesc: str or JSON.
            Can be either the name of the agent (str) or
            the descriptor

            {
                "name" : [the name of the agent],
                "effectParameters" : {
                    TenBergeCoefficient and ect.
                },
                "effects": {
                    "effect name" : { effect data (+ injury levels) }


                }
            }


        :param projectName: str
                The name of the project in the local DB that will be searched for the agent.
        :return:
        """
        logger = get_classMethod_logger(self, "getAgent")
        if isinstance(nameOrDesc, str):
            descriptor = self.getDataSourceData(nameOrDesc, version=version)
            if descriptor is None:
                raise ValueError(f"Agent {nameOrDesc} is not found. Load it with hera-risk-agent load")

            agentDataSource = self.getDataSourceDocument(nameOrDesc, version=version)
            if 'name' not in descriptor:
                logger.warning("name not specified in the agent descriptor, using the datasource name as a default")
                descriptor['name'] = agentDataSource.desc['datasourceName']
            if 'version' not in descriptor:
                logger.warning("version not specified in the agent descriptor, using the datasource version as a default")
                descriptor['version'] = agentDataSource.desc['version']

        elif isinstance(nameOrDesc, dict):
            descriptor = nameOrDesc
        else:
            raise ValueError("nameOrDesc must be the agent name (str) or its JSON description (dict) ")

        return Agent(descriptor)


    def listAgentsNames(self):
        """
            Lists the agents that are currently loaded in the DB (both local and public).

        :return: list
            A list of agent names.

        """
        return [x.desc["datasourceName"] for x in self.getDataSourceDocumentsList()]

    def loadAgent(self, name, agentDescription, version,saveMode=TOOLKIT_SAVEMODE_FILEANDDB):
        """
			Adds the agent to the DB. Either to the public or to the local DB.
			Equivalent to loadData

        :param name: str
                Agent name
        :param agentDescription: dict
                The agent description

        :return:
                None
        """
        agentDescription['name'] = name
        agentDescription['version'] = version
        return self.loadData(agentDescription,saveMode=saveMode)

    def loadData(self, fileNameOrData, saveMode=TOOLKIT_SAVEMODE_FILEANDDB,**kwargs):
        """
            Abstract loading a data from file. Manages the parsing of the
            datafile.

			Equivalent to loadData

        Parameters
        ----------
        fileNameOrData: str
                If str , the datafile to load
                If other objects - convert the
        parser: str
                The name of the parser to use

        :param saveMode: str
                Can be either:

                    - TOOLKIT_SAVEMODE_NOSAVE   : Just load the data from file and return the datafile


                    - TOOLKIT_SAVEMODE_FILEANDDB : Loads the data from file and save to a file and store to the DB as a source.
                                                    Raise exception if the entry exists.

                    - TOOLKIT_SAVEMODE_FILEANDDB_REPLACE: Loads the data from file and save to a file and store to the DB as a source.
                                                    Replace the entry in the DB if it exists.

        """
        if isinstance(fileNameOrData, str):
            if os.path.isfile(fileNameOrData):
                with open(fileNameOrData,'r') as readFile:
                    agentDescription = json.load(readFile)
            else:
                agentDescription = json.loads(fileNameOrData)
        elif isinstance(fileNameOrData,dict):
            agentDescription = fileNameOrData
        else:
            raise ValueError("fileNameOrData must be a file, JSON str or JSON object (dict)")

        name = agentDescription['name']
        version = agentDescription.get('version',None)

        agentDoc = self.getDataSourceDocument(datasourceName=name, version=version)

        if agentDoc is None:
            self.addDataSource(name, resource=json.dumps(agentDescription), dataFormat=datatypes.JSON_DICT, **agentDescription)

        elif saveMode == TOOLKIT_SAVEMODE_FILEANDDB:
            raise ValueError(f"Agent {name} version {agentDoc.desc.get('version',None)} in the database.")
        else:
            agentDoc.resource = json.dumps(agentDescription)
            agentDoc.desc['version']  = version
            agentDoc.save()
        return nonDBMetadataFrame(agentDescription) if agentDoc is None else agentDoc

analysis property

Access the risk analysis layer.

Returns:

Type Description
analysis

Object providing risk area calculation and LSM integration.

ProtectionPolicy property

Access the ProtectionPolicy class for building protection action pipelines.

Returns:

Type Description
type

The ProtectionPolicy class (not an instance).

presentation property

Access the presentation layer for casualty visualizations.

Returns:

Type Description
casualtiesPlot

__init__(projectName, filesDirectory=None, connectionName=None)

Initialize the RiskToolkit.

Parameters:

Name Type Description Default
projectName str

The project name.

 required
filesDirectory str

Directory for file outputs.

 None
connectionName str

The DB connection name.

 None
Source code in hera/riskassessment/riskToolkit.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(self, projectName, filesDirectory=None, connectionName=None):
    """
    Initialize the RiskToolkit.

    Parameters
    ----------
    projectName : str
        The project name.
    filesDirectory : str, optional
        Directory for file outputs.
    connectionName : str, optional
        The DB connection name.
    """
    super().__init__(projectName=projectName, filesDirectory=filesDirectory, toolkitName="RiskAssessment", connectionName=connectionName)
    self._presentation = casualtiesPlot()
    self._protectionPolicy = ProtectionPolicy
    self._analysis = analysis(self)

getAgent(nameOrDesc, version=None) 

Initialize the agents.

:param nameOrDesc: str or JSON. Can be either the name of the agent (str) or the descriptor

{
    "name" : [the name of the agent],
    "effectParameters" : {
        TenBergeCoefficient and ect.
    },
    "effects": {
        "effect name" : { effect data (+ injury levels) }


    }
}

:param projectName: str The name of the project in the local DB that will be searched for the agent. :return:

Source code in hera/riskassessment/riskToolkit.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def getAgent(self, nameOrDesc, version=None):
    """
        Initialize the agents.

    :param nameOrDesc: str or JSON.
        Can be either the name of the agent (str) or
        the descriptor

        {
            "name" : [the name of the agent],
            "effectParameters" : {
                TenBergeCoefficient and ect.
            },
            "effects": {
                "effect name" : { effect data (+ injury levels) }


            }
        }


    :param projectName: str
            The name of the project in the local DB that will be searched for the agent.
    :return:
    """
    logger = get_classMethod_logger(self, "getAgent")
    if isinstance(nameOrDesc, str):
        descriptor = self.getDataSourceData(nameOrDesc, version=version)
        if descriptor is None:
            raise ValueError(f"Agent {nameOrDesc} is not found. Load it with hera-risk-agent load")

        agentDataSource = self.getDataSourceDocument(nameOrDesc, version=version)
        if 'name' not in descriptor:
            logger.warning("name not specified in the agent descriptor, using the datasource name as a default")
            descriptor['name'] = agentDataSource.desc['datasourceName']
        if 'version' not in descriptor:
            logger.warning("version not specified in the agent descriptor, using the datasource version as a default")
            descriptor['version'] = agentDataSource.desc['version']

    elif isinstance(nameOrDesc, dict):
        descriptor = nameOrDesc
    else:
        raise ValueError("nameOrDesc must be the agent name (str) or its JSON description (dict) ")

    return Agent(descriptor)

listAgentsNames() 

Lists the agents that are currently loaded in the DB (both local and public).

:return: list A list of agent names.

Source code in hera/riskassessment/riskToolkit.py 
127
128
129
130
131
132
133
134
135
def listAgentsNames(self):
    """
        Lists the agents that are currently loaded in the DB (both local and public).

    :return: list
        A list of agent names.

    """
    return [x.desc["datasourceName"] for x in self.getDataSourceDocumentsList()]

loadAgent(name, agentDescription, version, saveMode=TOOLKIT_SAVEMODE_FILEANDDB) 

            Adds the agent to the DB. Either to the public or to the local DB.
            Equivalent to loadData

:param name: str Agent name :param agentDescription: dict The agent description

:return: None

Source code in hera/riskassessment/riskToolkit.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
    def loadAgent(self, name, agentDescription, version,saveMode=TOOLKIT_SAVEMODE_FILEANDDB):
        """
			Adds the agent to the DB. Either to the public or to the local DB.
			Equivalent to loadData

        :param name: str
                Agent name
        :param agentDescription: dict
                The agent description

        :return:
                None
        """
        agentDescription['name'] = name
        agentDescription['version'] = version
        return self.loadData(agentDescription,saveMode=saveMode)

loadData(fileNameOrData, saveMode=TOOLKIT_SAVEMODE_FILEANDDB, **kwargs) 

Abstract loading a data from file. Manages the parsing of the
datafile.

            Equivalent to loadData

Parameters:

Name Type Description Default
fileNameOrData 
If str , the datafile to load
If other objects - convert the
required
parser 
The name of the parser to use
required
Source code in hera/riskassessment/riskToolkit.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
    def loadData(self, fileNameOrData, saveMode=TOOLKIT_SAVEMODE_FILEANDDB,**kwargs):
        """
            Abstract loading a data from file. Manages the parsing of the
            datafile.

			Equivalent to loadData

        Parameters
        ----------
        fileNameOrData: str
                If str , the datafile to load
                If other objects - convert the
        parser: str
                The name of the parser to use

        :param saveMode: str
                Can be either:

                    - TOOLKIT_SAVEMODE_NOSAVE   : Just load the data from file and return the datafile


                    - TOOLKIT_SAVEMODE_FILEANDDB : Loads the data from file and save to a file and store to the DB as a source.
                                                    Raise exception if the entry exists.

                    - TOOLKIT_SAVEMODE_FILEANDDB_REPLACE: Loads the data from file and save to a file and store to the DB as a source.
                                                    Replace the entry in the DB if it exists.

        """
        if isinstance(fileNameOrData, str):
            if os.path.isfile(fileNameOrData):
                with open(fileNameOrData,'r') as readFile:
                    agentDescription = json.load(readFile)
            else:
                agentDescription = json.loads(fileNameOrData)
        elif isinstance(fileNameOrData,dict):
            agentDescription = fileNameOrData
        else:
            raise ValueError("fileNameOrData must be a file, JSON str or JSON object (dict)")

        name = agentDescription['name']
        version = agentDescription.get('version',None)

        agentDoc = self.getDataSourceDocument(datasourceName=name, version=version)

        if agentDoc is None:
            self.addDataSource(name, resource=json.dumps(agentDescription), dataFormat=datatypes.JSON_DICT, **agentDescription)

        elif saveMode == TOOLKIT_SAVEMODE_FILEANDDB:
            raise ValueError(f"Agent {name} version {agentDoc.desc.get('version',None)} in the database.")
        else:
            agentDoc.resource = json.dumps(agentDescription)
            agentDoc.desc['version']  = version
            agentDoc.save()
        return nonDBMetadataFrame(agentDescription) if agentDoc is None else agentDoc

Agent

hera.riskassessment.agents.Agents.Agent

Represents a hazardous agent with its associated injury effect models.

An agent is initialized from a JSON descriptor that defines its name, effect parameters (e.g. Ten Berge coefficient), and one or more injury effects (e.g. inhalation, thermal). Each effect is accessible by name via dictionary-style access (agent["inhalation"]).

Source code in hera/riskassessment/agents/Agents.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class Agent:
	"""
	Represents a hazardous agent with its associated injury effect models.

	An agent is initialized from a JSON descriptor that defines its name,
	effect parameters (e.g. Ten Berge coefficient), and one or more injury
	effects (e.g. inhalation, thermal). Each effect is accessible by name
	via dictionary-style access (``agent["inhalation"]``).
	"""

	_effects = None

	_effectsParameters = None

	@property
	def effectNames(self):
		"""
		List of effect names defined for this agent.

		Returns
		-------
		list of str
		"""
		return [x for x in self._effects.keys()]

	def  __getitem__(self,name):
		"""
		Access an effect by name.

		Parameters
		----------
		name : str
			The effect name.

		Returns
		-------
		Injury
			The injury effect object.
		"""
		return self._effects[name]

	@property
	def physicalproperties(self):
		"""
		Physical properties of the agent (molecular weight, density, vapor pressure, etc.).

		Returns
		-------
		PhysicalPropeties
		"""
		return self._physicalproperties

	@property
	def fullDescription(self):
		"""
		The full JSON descriptor used to initialize this agent.

		Returns
		-------
		dict
		"""
		return self._agentconfig

	@property
	def effectproperties(self):
		"""
		The effect parameters dictionary (e.g. tenbergeCoefficient).

		Returns
		-------
		dict
		"""
		return self._effectParameters

	@property
	def tenbergeCoefficient(self):
		"""
		The Ten Berge coefficient for dose-response calculations.

		Returns
		-------
		float
		"""
		return self._effectParameters.get("tenbergeCoefficient",1)

	@tenbergeCoefficient.setter
	def tenbergeCoefficient(self,value):
		"""Set the Ten Berge coefficient and rebuild effects."""
		self._effectParameters["tenbergeCoefficient"] = float(value)
		for effectname,effectconfig in self._agentconfig["effects"].items():
			self._effects[effectname] = injuryfactory.getInjury(effectname,effectconfig,**self._effectParameters)


	@property
	def name(self):
		"""
		The name of the agent.

		Returns
		-------
		str
		"""
		return self._agentconfig['name']


	def __init__(self,descriptor):
		"""
			Constructor of the Agent project.

			Initializes an agent.


		Parameters
		-----------
		descriptor: JSON
			A JSON object (dict) that holds all the information on the agent.

			{
				"name" : [the name of the agent],
				"effectParameters" : {
					TenBergeCoefficient and ect.
				},
				"effects": {
					"effect name" : { effect data (+ injury levels) }


				}
			}

		"""
		self._agentconfig = descriptor
		self._effectParameters = self._agentconfig.get("effectParameters",{})

		self._effects = {}
		for effectname,effectconfig in self._agentconfig["effects"].items():
			self._effects[effectname] = injuryfactory.getInjury(effectname,effectconfig,**self._effectParameters)

		self.__dict__.update(self._effects)

		self._physicalproperties = PhysicalPropeties(self._agentconfig)


	def toJSON(self):
		"""
		Serialize the agent to a JSON-compatible dictionary.

		Returns
		-------
		dict
		"""
		ret = dict(name=self.name,
				   physicalProperties=self.physicalproperties.toJSON(),
				   effect={})

		for effect in self.effectNames:
			ret['effect'][effect] = self[effect].toJSON()

		return ret


	def __str__(self):
		"""Return a JSON string representation of the agent."""
		return json.dumps(self.toJSON(),indent=4)

effectNames property

List of effect names defined for this agent.

Returns:

Type Description
list of str

physicalproperties property

Physical properties of the agent (molecular weight, density, vapor pressure, etc.).

Returns:

Type Description
PhysicalPropeties

fullDescription property

The full JSON descriptor used to initialize this agent.

Returns:

Type Description
dict

effectproperties property

The effect parameters dictionary (e.g. tenbergeCoefficient).

Returns:

Type Description
dict

tenbergeCoefficient property writable

The Ten Berge coefficient for dose-response calculations.

Returns:

Type Description
float

name property

The name of the agent.

Returns:

Type Description
str

__getitem__(name)

Access an effect by name.

Parameters:

Name Type Description Default
name str 
The effect name.
required

Returns:

Type Description
Injury

The injury effect object.
Source code in hera/riskassessment/agents/Agents.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def  __getitem__(self,name):
	"""
	Access an effect by name.

	Parameters
	----------
	name : str
		The effect name.

	Returns
	-------
	Injury
		The injury effect object.
	"""
	return self._effects[name]

__init__(descriptor) 

    Constructor of the Agent project.

    Initializes an agent.

Parameters:

Name Type Description Default
descriptor 
A JSON object (dict) that holds all the information on the agent.

{
        "name" : [the name of the agent],
        "effectParameters" : {
                TenBergeCoefficient and ect.
        },
        "effects": {
                "effect name" : { effect data (+ injury levels) }


        }
}
required
Source code in hera/riskassessment/agents/Agents.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def __init__(self,descriptor):
	"""
		Constructor of the Agent project.

		Initializes an agent.


	Parameters
	-----------
	descriptor: JSON
		A JSON object (dict) that holds all the information on the agent.

		{
			"name" : [the name of the agent],
			"effectParameters" : {
				TenBergeCoefficient and ect.
			},
			"effects": {
				"effect name" : { effect data (+ injury levels) }


			}
		}

	"""
	self._agentconfig = descriptor
	self._effectParameters = self._agentconfig.get("effectParameters",{})

	self._effects = {}
	for effectname,effectconfig in self._agentconfig["effects"].items():
		self._effects[effectname] = injuryfactory.getInjury(effectname,effectconfig,**self._effectParameters)

	self.__dict__.update(self._effects)

	self._physicalproperties = PhysicalPropeties(self._agentconfig)

toJSON()

Serialize the agent to a JSON-compatible dictionary.

Returns:

Type Description
dict
Source code in hera/riskassessment/agents/Agents.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def toJSON(self):
	"""
	Serialize the agent to a JSON-compatible dictionary.

	Returns
	-------
	dict
	"""
	ret = dict(name=self.name,
			   physicalProperties=self.physicalproperties.toJSON(),
			   effect={})

	for effect in self.effectNames:
		ret['effect'][effect] = self[effect].toJSON()

	return ret

__str__()

Return a JSON string representation of the agent.

Source code in hera/riskassessment/agents/Agents.py 
168
169
170
def __str__(self):
	"""Return a JSON string representation of the agent."""
	return json.dumps(self.toJSON(),indent=4)

ProtectionPolicy

hera.riskassessment.protectionpolicy.ProtectionPolicy.ProtectionPolicy

Bases: object

Calculates the expected concentration of the defined protection policy. The policy should include: - indoor time. - masking. - evacuation.

So a policy is a list of actions that take place: each action modifies the concentration field and is defined by a begining and end. The end can be infite (=None), like in evacuation.

The object should support chaining of actions:

ProtectionPolicy.indoor(alpha=0.2,begin=0,end=10).masking(factor=1000,begin=0,end=1000).compute(C,concentrationField="C") .masking(distribution="beta",mean=1000,sigma=10,begin=0,end=1000,bins=100).compute(C,concentrationField="C")

This will create attributes with the details of the execution pipeline. in each attribute we will have: actionID (sequential number) => { "params" : { paramname => value } "outputfields" : ["field"] }

outputfields is used for the subsequent protection (if there is multiple output).

Currently, only a single indoor time is defined (without exit time)

In the future, we should calculate the concentrations field for each action imposed on the concentrations of the previous one (i.e add a field).

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
class ProtectionPolicy(object): 
	"""
		Calculates the expected concentration of the defined protection policy. 
		The policy should include: 
			- indoor time. 
			- masking. 
			- evacuation. 

		So a policy is a list of actions that take place: 
		each action modifies the concentration field and is defined by a begining and end. 
		The end can be infite (=None), like in evacuation. 

		The object should support chaining of actions: 

		ProtectionPolicy.indoor(alpha=0.2,begin=0,end=10).masking(factor=1000,begin=0,end=1000).compute(C,concentrationField="C")
				  			         .masking(distribution="beta",mean=1000,sigma=10,begin=0,end=1000,bins=100).compute(C,concentrationField="C")

		This will create attributes with the details of the execution pipeline. 
		in each attribute we will have: 
				actionID (sequential number) =>  { "params" : { paramname => value }
                                                                   "outputfields" : ["field"] 
        							 } 

		outputfields is used for the subsequent protection (if there is multiple output). 

		Currently, only a single indoor time is defined (without exit time) 

		In the future, we should calculate the concentrations field for each action imposed on the concentrations of the previous 
 		one (i.e add a field). 
	"""
	_data = None
	_xname = None
	_yname = None
	_datetimename = None

	_actionList = None

	_finalname = None

	_params = None

	@property
	def params(self):
		"""
		dict
		    The parameters associated with this protection policy.
		"""
		return self._params

	@property
	def finalname(self):
		"""
		str
		    The name of the final concentration field in the dataset after
		    all protection actions have been applied.
		"""
		return self._finalname

	@property
	def xname(self):
		"""
		str
		    The name of the x-coordinate dimension in the dataset.
		"""
		return self._xname
	@property
	def yname(self):
		"""
		str
		    The name of the y-coordinate dimension in the dataset.
		"""
		return self._yname

	@property
	def datetimename(self):
		"""
		str
		    The name of the datetime dimension in the dataset.
		"""
		return self._datetimename

	@property
	def data(self):
		"""
		xarray.Dataset or None
		    The concentration dataset produced after calling ``compute``.
		    ``None`` before computation.
		"""
		return self._data

	def __init__(self,actionList=[],x="x",y="y",datetime="datetime"): 
		"""
			A basic action list. 

			[
				{
					name: "indoor|masking|evacuation",
					params: { action params }
				}
			]
		"""
		self._xname 		= x
		self._yname 		= y 
		self._datetimename 	= datetime
		self._actionList 	= []
		self._finalname  	= "C"
		self.addActions(dict(actions=numpy.atleast_1d(actionList)))


	def addActions(self,jsonStrOrFile): 
		"""
			add actions from a file. 

			The actions are under the "actions" key. 
		"""
		if isinstance(jsonStrOrFile,str): 
			if os.JSONpath.exists(jsonStrOrFile):
				with open(jsonStrOrFile,"r") as jsonFile: 
					jsonFile = json.load(jsonFile)
			else: 
				jsonFile = json.load(jsonStrOrFile)
		else:
			jsonFile = jsonStrOrFile

		for actionJSON in jsonFile["actions"]:
			self.addAction(actionJSON["name"],actionJSON["params"])


	def addAction(self,name,params): 
		"""
			Adds json with the params
		"""
		newaction = abstractAction.getAction(len(self._actionList)+1,self,name,params)
		self._actionList.append(newaction)
		return newaction

	def indoor(self,**kwargs): 
		"""
			Adds an indoor protection to the list of actions. 

			Must supply alpha or turnover. If indoor is restricted by time, 
			then must supply (begin,end) or (enter,stay). 

			:params: kwargs: 

			* alpha (1/[time])  - room alpha = 1/[turn over rate].
			* turnover ([time]) - room turn over rate. 
			* begin - the time in which the population enters indoor
			* end   - the time in which the population leaves indoor. 
			* enter - timedelta str after the begining of the simulation that the population enters indoor  
			* stay  - timedelta str for staying indoor. 
		"""
		if "alpha" not in kwargs and "turnover" not in kwargs: 
			raise ValueError("must supply either alpha or turnover")

		self.addAction("Indoor",kwargs)
		return self

	def masks(self, **kwargs):
		"""
			Adds a masking protection. 

			:params: kwargs: 

			* protectionFactor  - The protectionFactor of the masks. 
			* begin - the time in which the population enters indoor.
			* end   - the time in which the population leaves indoor. 
			* enter - timedelta str after the begining of the simulation that the population enters indoor.
			* stay  - timedelta str for staying indoor. 
		"""

		if "protectionFactor" not in kwargs: 
			raise ValueError("must supply protectionFactor")

		self.addAction("Masks",kwargs)

		return self

	def compute(self,data,C="C"): 
		"""
			Executes the pipeline. 
		"""
		self._data = xarray.Dataset()
		self._data.attrs = data.attrs
		self._data["outdoor_0"] = data[C]
		self._data[self.finalname] = data[C]
		self._data.attrs["0"] = { "type" : "outdoor"}
		for action in self._actionList: 
			action.compute()

		self._data.compute()
		return self.data

	@property
	def hdfkey(self):
		"""Combined HDF key for all actions in this policy.

		Returns
		-------
		str
		"""
		return "/".join([action.hdfkey for action in self._actionList])

params property

dict The parameters associated with this protection policy.

finalname property

str The name of the final concentration field in the dataset after all protection actions have been applied.

xname property

str The name of the x-coordinate dimension in the dataset.

yname property

str The name of the y-coordinate dimension in the dataset.

datetimename property

str The name of the datetime dimension in the dataset.

data property

xarray.Dataset or None The concentration dataset produced after calling compute. None before computation.

hdfkey property

Combined HDF key for all actions in this policy.

Returns:

Type Description
str

__init__(actionList=[], x='x', y='y', datetime='datetime')

A basic action list.

[ { name: "indoor|masking|evacuation", params: { action params } } ]

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __init__(self,actionList=[],x="x",y="y",datetime="datetime"): 
	"""
		A basic action list. 

		[
			{
				name: "indoor|masking|evacuation",
				params: { action params }
			}
		]
	"""
	self._xname 		= x
	self._yname 		= y 
	self._datetimename 	= datetime
	self._actionList 	= []
	self._finalname  	= "C"
	self.addActions(dict(actions=numpy.atleast_1d(actionList)))

addActions(jsonStrOrFile)

add actions from a file.

The actions are under the "actions" key.

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def addActions(self,jsonStrOrFile): 
	"""
		add actions from a file. 

		The actions are under the "actions" key. 
	"""
	if isinstance(jsonStrOrFile,str): 
		if os.JSONpath.exists(jsonStrOrFile):
			with open(jsonStrOrFile,"r") as jsonFile: 
				jsonFile = json.load(jsonFile)
		else: 
			jsonFile = json.load(jsonStrOrFile)
	else:
		jsonFile = jsonStrOrFile

	for actionJSON in jsonFile["actions"]:
		self.addAction(actionJSON["name"],actionJSON["params"])

addAction(name, params)

Adds json with the params

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
137
138
139
140
141
142
143
def addAction(self,name,params): 
	"""
		Adds json with the params
	"""
	newaction = abstractAction.getAction(len(self._actionList)+1,self,name,params)
	self._actionList.append(newaction)
	return newaction

indoor(**kwargs)

Adds an indoor protection to the list of actions.

Must supply alpha or turnover. If indoor is restricted by time, then must supply (begin,end) or (enter,stay).

:params: kwargs:

  • alpha (1/[time]) - room alpha = 1/[turn over rate].
  • turnover ([time]) - room turn over rate.
  • begin - the time in which the population enters indoor
  • end - the time in which the population leaves indoor.
  • enter - timedelta str after the begining of the simulation that the population enters indoor
  • stay - timedelta str for staying indoor.
Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def indoor(self,**kwargs): 
	"""
		Adds an indoor protection to the list of actions. 

		Must supply alpha or turnover. If indoor is restricted by time, 
		then must supply (begin,end) or (enter,stay). 

		:params: kwargs: 

		* alpha (1/[time])  - room alpha = 1/[turn over rate].
		* turnover ([time]) - room turn over rate. 
		* begin - the time in which the population enters indoor
		* end   - the time in which the population leaves indoor. 
		* enter - timedelta str after the begining of the simulation that the population enters indoor  
		* stay  - timedelta str for staying indoor. 
	"""
	if "alpha" not in kwargs and "turnover" not in kwargs: 
		raise ValueError("must supply either alpha or turnover")

	self.addAction("Indoor",kwargs)
	return self

masks(**kwargs)

Adds a masking protection.

:params: kwargs:

  • protectionFactor - The protectionFactor of the masks.
  • begin - the time in which the population enters indoor.
  • end - the time in which the population leaves indoor.
  • enter - timedelta str after the begining of the simulation that the population enters indoor.
  • stay - timedelta str for staying indoor.
Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def masks(self, **kwargs):
	"""
		Adds a masking protection. 

		:params: kwargs: 

		* protectionFactor  - The protectionFactor of the masks. 
		* begin - the time in which the population enters indoor.
		* end   - the time in which the population leaves indoor. 
		* enter - timedelta str after the begining of the simulation that the population enters indoor.
		* stay  - timedelta str for staying indoor. 
	"""

	if "protectionFactor" not in kwargs: 
		raise ValueError("must supply protectionFactor")

	self.addAction("Masks",kwargs)

	return self

compute(data, C='C')

Executes the pipeline.

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def compute(self,data,C="C"): 
	"""
		Executes the pipeline. 
	"""
	self._data = xarray.Dataset()
	self._data.attrs = data.attrs
	self._data["outdoor_0"] = data[C]
	self._data[self.finalname] = data[C]
	self._data.attrs["0"] = { "type" : "outdoor"}
	for action in self._actionList: 
		action.compute()

	self._data.compute()
	return self.data