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 
 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
211
212
213
214
215
216
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=None,x="x",y="y",datetime="datetime"):
		"""
			A basic action list. 

			[
				{
					name: "indoor|masking|evacuation",
					params: { action params }
				}
			]
		"""
		if actionList is None:
			actionList = []
		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.path.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", lazy=False): 
		"""
			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()

		if not lazy:
			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=None, 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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def __init__(self,actionList=None,x="x",y="y",datetime="datetime"):
	"""
		A basic action list. 

		[
			{
				name: "indoor|masking|evacuation",
				params: { action params }
			}
		]
	"""
	if actionList is None:
		actionList = []
	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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def addActions(self,jsonStrOrFile): 
	"""
		add actions from a file. 

		The actions are under the "actions" key. 
	"""
	if isinstance(jsonStrOrFile,str): 
		if os.path.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 
142
143
144
145
146
147
148
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 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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', lazy=False)

Executes the pipeline.

Source code in hera/riskassessment/protectionpolicy/ProtectionPolicy.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def compute(self,data,C="C", lazy=False): 
	"""
		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()

	if not lazy:
		self._data.compute()
	return self.data

Effects

Injury models and calculators. Subclass names follow the Calculator<Name>, InjuryLevel<Name>, Injury<Name> convention because the factories resolve them with pydoc.locate.

Calculator

hera.riskassessment.agents.effects.Calculator

AbstractCalculator

Bases: object

Holds the abstract calculator.

Source code in hera/riskassessment/agents/effects/Calculator.py 
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
class AbstractCalculator(object):
	"""
		Holds the abstract calculator. 
	"""

	# The breathing rate with which the injury level was determined. 
	_injury_breathingRate = None 
	@property
	def injuryBreathingRate(self):
		"""Breathing rate used when the injury level was determined.

		Returns
		-------
		pint.Quantity
		"""
		return self._injury_breathingRate

	def __init__(self,breathingRate=10*ureg.L/ureg.min):
		"""Initialize with the reference breathing rate.

		Parameters
		----------
		breathingRate : pint.Quantity, optional
			Default is 10 L/min.
		"""
		self._injury_breathingRate=breathingRate


	def toJSON(self):
		"""Serialize the calculator to a JSON-compatible dictionary."""
		return dict(breathingRate=str(self.injuryBreathingRate))

	def __str__(self):
		"""Return a JSON string representation of the calculator."""
		return json.dumps(self.toJSON())
injuryBreathingRate property

Breathing rate used when the injury level was determined.

Returns:

Type Description
Quantity
__init__(breathingRate=10 * ureg.L / ureg.min)

Initialize with the reference breathing rate.

Parameters:

Name Type Description Default
breathingRate Quantity 
Default is 10 L/min.
10 * L / min
Source code in hera/riskassessment/agents/effects/Calculator.py 
28
29
30
31
32
33
34
35
36
def __init__(self,breathingRate=10*ureg.L/ureg.min):
	"""Initialize with the reference breathing rate.

	Parameters
	----------
	breathingRate : pint.Quantity, optional
		Default is 10 L/min.
	"""
	self._injury_breathingRate=breathingRate
toJSON()

Serialize the calculator to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/Calculator.py 
39
40
41
def toJSON(self):
	"""Serialize the calculator to a JSON-compatible dictionary."""
	return dict(breathingRate=str(self.injuryBreathingRate))
__str__()

Return a JSON string representation of the calculator.

Source code in hera/riskassessment/agents/effects/Calculator.py 
43
44
45
def __str__(self):
	"""Return a JSON string representation of the calculator."""
	return json.dumps(self.toJSON())

CalculatorHaber

Bases: AbstractCalculator

Implements the haber law (simple integral) on the concentrations.

Source code in hera/riskassessment/agents/effects/Calculator.py 
 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
class CalculatorHaber(AbstractCalculator): 
	"""
		Implements the haber law (simple integral) on the concentrations. 
	"""

	def __init__(self,breathingRate=10*ureg.L/ureg.min,**kwargs): 
		"""
			Calculates the Haber dosage. The integration is performed on the time axis. 

			:param: time: a string with the name of the axis to perform the integration on. 
		"""
		super().__init__(breathingRate=breathingRate)

	def calculate(self,concentrationField,breathingRate=10*ureg.L/ureg.min,time="datetime",field=None,inUnits=None):
		"""
			Calculates the dosage from a concentration field.

			Concentration field can be pandas.DataFrame or xarray.Dataset.
			If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

			Parameters
			----------
				concentrationField : pandas.DataFrame or xarray dataframe.
						The concentrations. Must have 'time' index.

						For pandas, we do not assume that the time is equispaced.

				breathingRate : unum, L/min
		    			The breathing rate of the population.

	  			time : str
	  					The name of the time coordinates. In pandas it is the name of the index.
	  					That is, the index will be accessed after a reset_index() function, and that field must be present.

	  			field : str
	  					Only in xarray. The name of the field to use from the xarray (default None).
	  					Ignored in pandas.
	  			inUnits: unum
	  					The units of the concentration.
	  					If None, use the default units.

	  					The default units of pandas are mg/m**3. For xarray it the units in the attributes.
	  		Return
	  		-------
	  				An xarray or pandas (depend on the input) with the toxic loads
	  				in each time step.


		"""
		breathingRate = unumToPint(breathingRate)
		breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
		if inUnits is None:
			inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
		if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
			inUnits = 1*inUnits

		CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

		if isinstance(concentrationField,xarray.Dataset):
			return concentrationField[field].cumsum(dim=time)*concentrationField.dt.m_as(ureg.min)*breathingRatio*CunitConversion
		elif isinstance(concentrationField,pandas.DataFrame):
			dt_min = concentrationField.reset_index()[time].diff().apply(lambda x: x.seconds)/60.
			return (concentrationField[:-1].fillna(0)*dt_min[1:]).cumsum()*breathingRatio * CunitConversion
		else:
			raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")

	def toJSON(self):
		"""Serialize the Haber calculator to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'haber'
		return ret
__init__(breathingRate=10 * ureg.L / ureg.min, **kwargs)

Calculates the Haber dosage. The integration is performed on the time axis.

:param: time: a string with the name of the axis to perform the integration on.

Source code in hera/riskassessment/agents/effects/Calculator.py 
52
53
54
55
56
57
58
def __init__(self,breathingRate=10*ureg.L/ureg.min,**kwargs): 
	"""
		Calculates the Haber dosage. The integration is performed on the time axis. 

		:param: time: a string with the name of the axis to perform the integration on. 
	"""
	super().__init__(breathingRate=breathingRate)
calculate(concentrationField, breathingRate=10 * ureg.L / ureg.min, time='datetime', field=None, inUnits=None)

Calculates the dosage from a concentration field.

Concentration field can be pandas.DataFrame or xarray.Dataset. If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

Return 
            An xarray or pandas (depend on the input) with the toxic loads
            in each time step.
Source code in hera/riskassessment/agents/effects/Calculator.py 
 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
def calculate(self,concentrationField,breathingRate=10*ureg.L/ureg.min,time="datetime",field=None,inUnits=None):
	"""
		Calculates the dosage from a concentration field.

		Concentration field can be pandas.DataFrame or xarray.Dataset.
		If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

		Parameters
		----------
			concentrationField : pandas.DataFrame or xarray dataframe.
					The concentrations. Must have 'time' index.

					For pandas, we do not assume that the time is equispaced.

			breathingRate : unum, L/min
	    			The breathing rate of the population.

  			time : str
  					The name of the time coordinates. In pandas it is the name of the index.
  					That is, the index will be accessed after a reset_index() function, and that field must be present.

  			field : str
  					Only in xarray. The name of the field to use from the xarray (default None).
  					Ignored in pandas.
  			inUnits: unum
  					The units of the concentration.
  					If None, use the default units.

  					The default units of pandas are mg/m**3. For xarray it the units in the attributes.
  		Return
  		-------
  				An xarray or pandas (depend on the input) with the toxic loads
  				in each time step.


	"""
	breathingRate = unumToPint(breathingRate)
	breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
	if inUnits is None:
		inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
	if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
		inUnits = 1*inUnits

	CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

	if isinstance(concentrationField,xarray.Dataset):
		return concentrationField[field].cumsum(dim=time)*concentrationField.dt.m_as(ureg.min)*breathingRatio*CunitConversion
	elif isinstance(concentrationField,pandas.DataFrame):
		dt_min = concentrationField.reset_index()[time].diff().apply(lambda x: x.seconds)/60.
		return (concentrationField[:-1].fillna(0)*dt_min[1:]).cumsum()*breathingRatio * CunitConversion
	else:
		raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")
toJSON()

Serialize the Haber calculator to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/Calculator.py 
113
114
115
116
117
def toJSON(self):
	"""Serialize the Haber calculator to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'haber'
	return ret

CalculatorTenBerge

Bases: AbstractCalculator

Implements the ten=berge law (int C^n dt) on the concentrations.

Source code in hera/riskassessment/agents/effects/Calculator.py 
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
class CalculatorTenBerge(AbstractCalculator): 
	"""
		Implements the ten=berge law (int C^n dt) on the concentrations.
	"""

	n = None

	def __init__(self,tenbergeCoefficient,breathingRate=10*ureg.L/ureg.min,**kwargs): 
		"""
			:param: n: the ten-berge coefficient. 
			:param: time: the time variable. 
		"""
		super().__init__(breathingRate=breathingRate)
		self.n 	   = tenbergeCoefficient

	def calculate(self,concentrationField,field,breathingRate=10*ureg.L/ureg.min,time="datetime",inUnits=None):
		r"""
            Calculates the toxic load  from a concentration field.
            \begin{equation}
            		D(T) = \int_0^T C^n dt
            \end{equation}

            Concentration field can be pandas.DataFrame or xarray.Dataset.
            If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

            Parameters
            ----------
                concentrationField : pandas.DataFrame or xarray dataframe.
                        The concentrations. Must have 'time' index.

                        For pandas, we do not assume that the time is equispaced.

                breathingRate : unum, L/min
                        The breathing rate of the population.

                  time : str
                          The name of the time coordinates. In pandas it is the name of the index.
                          That is, the index will be accessed after a reset_index() function, and that field must be present.

                  field : str
                          Only in xarray. The name of the field to use from the xarray (default None).
                          Ignored in pandas.
                  inUnits: unum
                          The units of the concentration.
                          If None, use the default units.

                          The default units of pandas are mg/m**3. For xarray it the units in the attributes.
              Return
              -------
                      An xarray or pandas (depend on the input) with the toxic loads
                      in each time step.


        """
		breathingRate = unumToPint(breathingRate)
		breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
		if inUnits is None:
			inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
		if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
			inUnits = 1*inUnits
		CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

		if isinstance(concentrationField, xarray.Dataset):
			return ((concentrationField[field]*CunitConversion)**self.n).cumsum(dim=time)*concentrationField.dt.m_as(ureg.min)*breathingRatio
		elif isinstance(concentrationField,pandas.DataFrame):
			dt_min = concentrationField.reset_index()[time].diff().apply(lambda x: x.seconds) / 60.
			C_to_n = ((concentrationField[:-1].fillna(0)* CunitConversion )**self.n)
			# we have to ignore the index, so we transform one of the vectors to numpy.
			return (C_to_n * dt_min[1:].values.reshape(C_to_n.shape[0],1)).cumsum() * breathingRatio
		else:
			raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")

	def toJSON(self):
		"""Serialize the Ten Berge calculator to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'tenBerge'
		ret['n'] = str(self.n)
		return ret
__init__(tenbergeCoefficient, breathingRate=10 * ureg.L / ureg.min, **kwargs)

:param: n: the ten-berge coefficient. :param: time: the time variable.

Source code in hera/riskassessment/agents/effects/Calculator.py 
127
128
129
130
131
132
133
def __init__(self,tenbergeCoefficient,breathingRate=10*ureg.L/ureg.min,**kwargs): 
	"""
		:param: n: the ten-berge coefficient. 
		:param: time: the time variable. 
	"""
	super().__init__(breathingRate=breathingRate)
	self.n 	   = tenbergeCoefficient
calculate(concentrationField, field, breathingRate=10 * ureg.L / ureg.min, time='datetime', inUnits=None)

Calculates the toxic load from a concentration field. \begin{equation} D(T) = \int_0^T C^n dt \end{equation}

Concentration field can be pandas.DataFrame or xarray.Dataset. If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

Source code in hera/riskassessment/agents/effects/Calculator.py 
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
	def calculate(self,concentrationField,field,breathingRate=10*ureg.L/ureg.min,time="datetime",inUnits=None):
		r"""
            Calculates the toxic load  from a concentration field.
            \begin{equation}
            		D(T) = \int_0^T C^n dt
            \end{equation}

            Concentration field can be pandas.DataFrame or xarray.Dataset.
            If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

            Parameters
            ----------
                concentrationField : pandas.DataFrame or xarray dataframe.
                        The concentrations. Must have 'time' index.

                        For pandas, we do not assume that the time is equispaced.

                breathingRate : unum, L/min
                        The breathing rate of the population.

                  time : str
                          The name of the time coordinates. In pandas it is the name of the index.
                          That is, the index will be accessed after a reset_index() function, and that field must be present.

                  field : str
                          Only in xarray. The name of the field to use from the xarray (default None).
                          Ignored in pandas.
                  inUnits: unum
                          The units of the concentration.
                          If None, use the default units.

                          The default units of pandas are mg/m**3. For xarray it the units in the attributes.
              Return
              -------
                      An xarray or pandas (depend on the input) with the toxic loads
                      in each time step.


        """
		breathingRate = unumToPint(breathingRate)
		breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
		if inUnits is None:
			inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
		if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
			inUnits = 1*inUnits
		CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

		if isinstance(concentrationField, xarray.Dataset):
			return ((concentrationField[field]*CunitConversion)**self.n).cumsum(dim=time)*concentrationField.dt.m_as(ureg.min)*breathingRatio
		elif isinstance(concentrationField,pandas.DataFrame):
			dt_min = concentrationField.reset_index()[time].diff().apply(lambda x: x.seconds) / 60.
			C_to_n = ((concentrationField[:-1].fillna(0)* CunitConversion )**self.n)
			# we have to ignore the index, so we transform one of the vectors to numpy.
			return (C_to_n * dt_min[1:].values.reshape(C_to_n.shape[0],1)).cumsum() * breathingRatio
		else:
			raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")
toJSON()

Serialize the Ten Berge calculator to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/Calculator.py 
192
193
194
195
196
197
def toJSON(self):
	"""Serialize the Ten Berge calculator to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'tenBerge'
	ret['n'] = str(self.n)
	return ret

CalculatorMaxConcentration

Bases: AbstractCalculator

Calculator using the maximum rolling-average concentration.

Source code in hera/riskassessment/agents/effects/Calculator.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class CalculatorMaxConcentration(AbstractCalculator):
	"""Calculator using the maximum rolling-average concentration."""

	def __init__(self,sampling,breathingRate=10*ureg.L/ureg.min,**kwargs): 
		"""
			Implements the maximal concentration on the time axis.
		"""
		super().__init__(breathingRate=breathingRate)
		self._sampling = sampling

	def calculate(self,concentrationField,field,breathingRate=10*ureg.L/ureg.min,time="datetime",inUnits=None):
		"""
            Calculate the maximal concentrations

            Concentration field can be pandas.DataFrame or xarray.Dataset.
            If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

            Parameters
            ----------
                concentrationField : pandas.DataFrame or xarray dataframe.
                        The concentrations. Must have 'time' index.

                        For pandas, we do not assume that the time is equispaced.

                breathingRate : unum, L/min
                        The breathing rate of the population.

                  time : str
                          The name of the time coordinates. In pandas it is the name of the index.
                          That is, the index will be accessed after a reset_index() function, and that field must be present.

                  field : str
                          Only in xarray. The name of the field to use from the xarray (default None).
                          Ignored in pandas.
                  inUnits: unum
                          The units of the concentration.
                          If None, use the default units.

                          The default units of pandas are mg/m**3. For xarray it the units in the attributes.
              Return
              -------
                      An xarray or pandas (depend on the input) with the toxic loads
                      in each time step.


        """
		breathingRate = unumToPint(breathingRate)
		breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
		if inUnits is None:
			inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
		if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
			inUnits = 1*inUnits
		CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

		if isinstance(concentrationField, xarray.Dataset):
			itemstep = pandas.to_timedelta(self._sampling) / pandas.to_timedelta(
				str(concentrationField.attrs["dt"]).replace("[", "").replace("]", ""))
			samplingparam = {time : int(itemstep)}
			return concentrationField[field].chunk(chunks={time: int(3*itemstep)}).rolling(**samplingparam).mean().fillna(0).max(dim=time)*breathingRatio*CunitConversion
		elif isinstance(concentrationField,pandas.DataFrame):
			return concentrationField.rolling(window=self._sampling).mean().fillna(0).max()*breathingRatio*CunitConversion
		else:
			raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")


	def toJSON(self):
		"""Serialize the max-concentration calculator to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'maxConcentrations'
		return ret
__init__(sampling, breathingRate=10 * ureg.L / ureg.min, **kwargs)

Implements the maximal concentration on the time axis.

Source code in hera/riskassessment/agents/effects/Calculator.py 
202
203
204
205
206
207
def __init__(self,sampling,breathingRate=10*ureg.L/ureg.min,**kwargs): 
	"""
		Implements the maximal concentration on the time axis.
	"""
	super().__init__(breathingRate=breathingRate)
	self._sampling = sampling
calculate(concentrationField, field, breathingRate=10 * ureg.L / ureg.min, time='datetime', inUnits=None)

Calculate the maximal concentrations

Concentration field can be pandas.DataFrame or xarray.Dataset. If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

Source code in hera/riskassessment/agents/effects/Calculator.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
	def calculate(self,concentrationField,field,breathingRate=10*ureg.L/ureg.min,time="datetime",inUnits=None):
		"""
            Calculate the maximal concentrations

            Concentration field can be pandas.DataFrame or xarray.Dataset.
            If it is xarray.Dataset, then use the field parameter to calculate the toxic from the requested field.

            Parameters
            ----------
                concentrationField : pandas.DataFrame or xarray dataframe.
                        The concentrations. Must have 'time' index.

                        For pandas, we do not assume that the time is equispaced.

                breathingRate : unum, L/min
                        The breathing rate of the population.

                  time : str
                          The name of the time coordinates. In pandas it is the name of the index.
                          That is, the index will be accessed after a reset_index() function, and that field must be present.

                  field : str
                          Only in xarray. The name of the field to use from the xarray (default None).
                          Ignored in pandas.
                  inUnits: unum
                          The units of the concentration.
                          If None, use the default units.

                          The default units of pandas are mg/m**3. For xarray it the units in the attributes.
              Return
              -------
                      An xarray or pandas (depend on the input) with the toxic loads
                      in each time step.


        """
		breathingRate = unumToPint(breathingRate)
		breathingRatio = (breathingRate/self.injuryBreathingRate).magnitude
		if inUnits is None:
			inUnits = concentrationField.attrs[field] if hasattr(concentrationField, "attrs") else 1*(ureg.mg / ureg.m ** 3)
		if isinstance(inUnits, Unit): # pint by default can't do m_as to a Units type
			inUnits = 1*inUnits
		CunitConversion = inUnits.m_as(ureg.mg / ureg.m ** 3)

		if isinstance(concentrationField, xarray.Dataset):
			itemstep = pandas.to_timedelta(self._sampling) / pandas.to_timedelta(
				str(concentrationField.attrs["dt"]).replace("[", "").replace("]", ""))
			samplingparam = {time : int(itemstep)}
			return concentrationField[field].chunk(chunks={time: int(3*itemstep)}).rolling(**samplingparam).mean().fillna(0).max(dim=time)*breathingRatio*CunitConversion
		elif isinstance(concentrationField,pandas.DataFrame):
			return concentrationField.rolling(window=self._sampling).mean().fillna(0).max()*breathingRatio*CunitConversion
		else:
			raise ValueError("concentrationField is not a pandas.DataFrame or xarray.Dataset")
toJSON()

Serialize the max-concentration calculator to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/Calculator.py 
264
265
266
267
268
def toJSON(self):
	"""Serialize the max-concentration calculator to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'maxConcentrations'
	return ret

InjuryLevel

hera.riskassessment.agents.effects.InjuryLevel

InjuryLevel

Bases: object

Represents an injury level

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
 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
class InjuryLevel(object): 
	"""
		Represents an injury level 
	"""
	_name =None

	_parameters = None 

	_calculator = None

	_units 	    = None 

	@property
	def name(self):
		"""Name of this injury level.

		Returns
		-------
		str
		"""
		return self._name


	@property
	def units(self):
		"""Units associated with this injury level.

		Returns
		-------
		pint.Unit or None
		"""
		return self._units

	def __init__(self,name,units=None,**parameters):
		"""Initialize the injury level.

		Parameters
		----------
		name : str
			Name of the injury level.
		units : pint.Unit or None, optional
			Units for the threshold values.
		**parameters
			Additional keyword parameters (e.g. TL_50, sigma).
		"""
		self._name = name
		self._parameters = parameters
		self._units = units


	def calculateContours(self, toxicLoads, time="datetime", x="x", y="y"):
		"""
			Calculates the contours of the current injury from the toxic loads.

			Parameters
			----------
				toxicLoads : xarray.Dataset
						The 2D map of the toxicloads in time.


				time : str
						The name of the time column

				x : str
						The name of the x column.

				y : str
						The name of the y column.



			Return
			-------
			GeoPandas.GeoDataFrame with the fields:

				time |  injury name | total polygon  | diff polygon 				| percentInjury
           			 |       	    | the total area | a differance from the level above. 	| if necessary.


			The data returns in km because itm is in km. 

		"""
		# headless to avoid graph appearing in UI
		curBackend = plt.get_backend()
		plt.close('all')
		plt.switch_backend("pdf")

		# flag when iteration over time steps is needed
		if time in toxicLoads.dims: 
			if toxicLoads[time].size == 1:
				timeList = [pandas.to_datetime(toxicLoads[time].item())]
				timesel  = False 
			else: 
				timeList = toxicLoads[time]
				timesel = True
		elif hasattr(toxicLoads,time):
			timeList = [pandas.to_datetime(getattr(toxicLoads,time).item())]
			timesel  = False
		else: 
			timeList= [0] 
			timesel = False 

		retList = []

		for tt in timeList: 
			concField = toxicLoads.sel(**{time:tt}) if timesel else toxicLoads

			ret = self._getGeopandas(concField,x,y) #,**parameters)
			if not ret.empty:

				ret = ret[ret.area > 1]   # 1 m**2. 
				ret['contour'] = ret['contour'].buffer(0) # sometimes fixes invalid polygons.

				ret = ret.dissolve(by="Level").reset_index()

				if "cloud_math_direction" in toxicLoads.attrs:
					ret['contour'] = ret['contour'].rotate(-toxicLoads.attrs["cloud_math_direction"], origin=(0, 0))

				ret["severity"] = self.name
				ret = ret.rename(columns={"Level" : "ToxicLoad","contour":"TotalPolygon",time : "datetime"}) 
				ret = ret.set_geometry("TotalPolygon")
				ret[time] = pandas.to_datetime(tt.item()) if timesel else tt
				retList.append(ret)

		ret = None if len(retList)==0 else pandas.concat(retList,ignore_index=True)
		plt.close('all')
		plt.switch_backend(curBackend)

		return ret


	def _getGeopandas(self,concentrationField,x,y,**parameters):
		"""
			Return the correct geopandas of the  Injury level 
		"""
		raise NotImplementedError("abstract function")


	def getContoursFromLevels(self, concentrationField, x, y, level):
		conc_x = concentrationField[x]
		conc_y = concentrationField[y]
		squeezed = concentrationField.squeeze()
		if squeezed.dims != (y, x):
			if squeezed.dims != (x,y):
				raise Exception(f"Cannot extract contours from xarray with dimensions {squeezed.dims} when x is given as {x} and y as {y}")
			squeezed = squeezed.transpose()
		return plt.contour(conc_x,conc_y,squeezed,levels=numpy.atleast_1d(level))


	def toJSON(self):
		"""Serialize the injury level to a JSON-compatible dictionary."""
		return dict(name=self.name,
					units=str(self.units))


	def __str__(self):
		"""Return a JSON string representation of the injury level."""
		json.dumps(self.toJSON(),indent=4)
name property

Name of this injury level.

Returns:

Type Description
str
units property

Units associated with this injury level.

Returns:

Type Description
Unit or None
__init__(name, units=None, **parameters)

Initialize the injury level.

Parameters:

Name Type Description Default
name str 
Name of the injury level.
required
units Unit or None 
Units for the threshold values.
None
**parameters 
Additional keyword parameters (e.g. TL_50, sigma).
{}
Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(self,name,units=None,**parameters):
	"""Initialize the injury level.

	Parameters
	----------
	name : str
		Name of the injury level.
	units : pint.Unit or None, optional
		Units for the threshold values.
	**parameters
		Additional keyword parameters (e.g. TL_50, sigma).
	"""
	self._name = name
	self._parameters = parameters
	self._units = units
calculateContours(toxicLoads, time='datetime', x='x', y='y')

Calculates the contours of the current injury from the toxic loads.

Return

GeoPandas.GeoDataFrame with the fields:

    time |  injury name | total polygon  | diff polygon                             | percentInjury
     |                  | the total area | a differance from the level above.       | if necessary.

The data returns in km because itm is in km.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
 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
	def calculateContours(self, toxicLoads, time="datetime", x="x", y="y"):
		"""
			Calculates the contours of the current injury from the toxic loads.

			Parameters
			----------
				toxicLoads : xarray.Dataset
						The 2D map of the toxicloads in time.


				time : str
						The name of the time column

				x : str
						The name of the x column.

				y : str
						The name of the y column.



			Return
			-------
			GeoPandas.GeoDataFrame with the fields:

				time |  injury name | total polygon  | diff polygon 				| percentInjury
           			 |       	    | the total area | a differance from the level above. 	| if necessary.


			The data returns in km because itm is in km. 

		"""
		# headless to avoid graph appearing in UI
		curBackend = plt.get_backend()
		plt.close('all')
		plt.switch_backend("pdf")

		# flag when iteration over time steps is needed
		if time in toxicLoads.dims: 
			if toxicLoads[time].size == 1:
				timeList = [pandas.to_datetime(toxicLoads[time].item())]
				timesel  = False 
			else: 
				timeList = toxicLoads[time]
				timesel = True
		elif hasattr(toxicLoads,time):
			timeList = [pandas.to_datetime(getattr(toxicLoads,time).item())]
			timesel  = False
		else: 
			timeList= [0] 
			timesel = False 

		retList = []

		for tt in timeList: 
			concField = toxicLoads.sel(**{time:tt}) if timesel else toxicLoads

			ret = self._getGeopandas(concField,x,y) #,**parameters)
			if not ret.empty:

				ret = ret[ret.area > 1]   # 1 m**2. 
				ret['contour'] = ret['contour'].buffer(0) # sometimes fixes invalid polygons.

				ret = ret.dissolve(by="Level").reset_index()

				if "cloud_math_direction" in toxicLoads.attrs:
					ret['contour'] = ret['contour'].rotate(-toxicLoads.attrs["cloud_math_direction"], origin=(0, 0))

				ret["severity"] = self.name
				ret = ret.rename(columns={"Level" : "ToxicLoad","contour":"TotalPolygon",time : "datetime"}) 
				ret = ret.set_geometry("TotalPolygon")
				ret[time] = pandas.to_datetime(tt.item()) if timesel else tt
				retList.append(ret)

		ret = None if len(retList)==0 else pandas.concat(retList,ignore_index=True)
		plt.close('all')
		plt.switch_backend(curBackend)

		return ret
toJSON()

Serialize the injury level to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
164
165
166
167
def toJSON(self):
	"""Serialize the injury level to a JSON-compatible dictionary."""
	return dict(name=self.name,
				units=str(self.units))
__str__()

Return a JSON string representation of the injury level.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
170
171
172
def __str__(self):
	"""Return a JSON string representation of the injury level."""
	json.dumps(self.toJSON(),indent=4)

InjuryLevelLognormal10DoseResponse

Bases: InjuryLevel

Calculates the % of the population that will be affected at a given ToxicLoad

This class also return the toxic load of a requested fraction.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
class InjuryLevelLognormal10DoseResponse(InjuryLevel): 
	"""
		Calculates the % of the population that will be affected at a given ToxicLoad 

		This class also return the toxic load of a requested fraction. 
	"""

	@property
	def TL_50(self):
		"""Toxic load at which 50% of the population is affected.

		Returns
		-------
		pint.Quantity
		"""
		return self._parameters["TL_50"]

	@property
	def sigma(self):
		"""Probit inverse (log-normal standard deviation).

		Returns
		-------
		float
		"""
		return self._parameters["sigma"]

	def __init__(self,name,**parameters):
		"""
			Initiates the injury level. 

			parameters are 
				- TL_50: The ToxicLoad in which 50% are effected. 
				- sigma: The probit inverse. 
		"""
		if "TL_50" not in parameters: 
			raise ValueError("Must supply TL_50") 
		else: 
			parameters["TL_50"] = tounit(parameters["TL_50"],dosage)
		if "sigma" not in parameters: 
			raise ValueError("Must supply sigma (1/probit)") 

		super(InjuryLevelLognormal10DoseResponse,self).__init__(name,**parameters)

	def getPercent(self,ToxicLoad):
		"""
	        Using lognormal distribution with base 10.

	        In that case define a new probable variable:

	        Y such that Y = X/a. (X is the original dose in base 10, Y is in natural base).
	        Then, X_avg/a =  Y_avg and  Sig_x/a = Sig_y.

	        Where a is the transformation between the bases:
	        10^(x) = exp(y) --> x = log10(e)*y --> a = log10(e).

	        Therefore, in order to use the lognormal with the natural base we divide
	        all the data (and the mean,std) with the 'a' coefficient.

	        Important Notes:
	        ---------------
	        1. See that the a factor in v/a and Effect[0]/a actually cancel.
	        2. lognorm calculates log(x/scale)/sigma --> log(x/scale)=  log10(x/scale)/a
	           But we use sigma/a ==> and we get that lognorm.cdf(x/a,sigma/a,scale/a) = lognorm.cdf(x,sigma/a,scale) = Phi(log10(x/scale)/a / (sigma/a) )
	                                  = Phi(log10(x/scale)/sigma ) which is what we got from the manuals (and see Ziv).

		"""
		a = numpy.log10(numpy.e)
		prob = lognorm.cdf(ToxicLoad/a,self.sigma/a,scale=self.TL_50.m_as(dosage)/a)
		return prob


	def getToxicLoad(self,Percent):
		"""Return the toxic load corresponding to a given injury fraction.

		Parameters
		----------
		Percent : float or array_like
			Desired injury fraction(s) (0-1).

		Returns
		-------
		float or numpy.ndarray
		"""
		a = numpy.log10(numpy.e)
		percent = lognorm.ppf(Percent,self.sigma/a,scale=self.TL_50.m_as(dosage)/a)*a
		return percent

	def _getGeopandas(self,concentrationField,x,y,**parameters):
		"""
			Return the correct geopandas of the  Injury level
			higher_severity
		"""
		defaultLevels = numpy.arange(0.05, 1, 0.05)
		percentList = parameters.get("percentInjury",defaultLevels)
		ToxicLoads = self.getToxicLoad(percentList)

		## We add the levels of higher toxicity to solve the problem
		# of a wide gape between the two severities (which led to double counting the casualties). 
		if self._parameters.get("higher_severity",None) is not None:
			HigherToxicLoads = self._parameters["higher_severity"].getToxicLoad(defaultLevels)
			ToxicLoads = numpy.unique(numpy.sort(numpy.concatenate([ToxicLoads,HigherToxicLoads])))

		CS = self.getContoursFromLevels(concentrationField, x, y, ToxicLoads)

		# Changes by ofir... not clear yet when the original fails....
		#concentrationField = concentrationField.to_dataframe().reset_index()
		#CS = plt.tricontour(concentrationField[x], concentrationField[y], concentrationField, levels=ToxicLoads) This line has bug because
		#	concentrationField is panas and we have to select the right column.

		if numpy.max(CS.levels) < numpy.min(ToxicLoads):
			ret = geopandas.GeoDataFrame()
		else:
			ret = toGeopandas(CS)
		return ret

	def toJSON(self):
		"""Serialize the lognormal dose-response level to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'logNormalBase10'
		ret['TL_50'] = str(self.TL_50)
		ret['sigma'] = str(self.sigma)
		return ret
TL_50 property

Toxic load at which 50% of the population is affected.

Returns:

Type Description
Quantity
sigma property

Probit inverse (log-normal standard deviation).

Returns:

Type Description
float
__init__(name, **parameters)

Initiates the injury level.

parameters are - TL_50: The ToxicLoad in which 50% are effected. - sigma: The probit inverse.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def __init__(self,name,**parameters):
	"""
		Initiates the injury level. 

		parameters are 
			- TL_50: The ToxicLoad in which 50% are effected. 
			- sigma: The probit inverse. 
	"""
	if "TL_50" not in parameters: 
		raise ValueError("Must supply TL_50") 
	else: 
		parameters["TL_50"] = tounit(parameters["TL_50"],dosage)
	if "sigma" not in parameters: 
		raise ValueError("Must supply sigma (1/probit)") 

	super(InjuryLevelLognormal10DoseResponse,self).__init__(name,**parameters)
getPercent(ToxicLoad)

Using lognormal distribution with base 10.

In that case define a new probable variable:

Y such that Y = X/a. (X is the original dose in base 10, Y is in natural base). Then, X_avg/a = Y_avg and Sig_x/a = Sig_y.

Where a is the transformation between the bases: 10^(x) = exp(y) --> x = log10(e)*y --> a = log10(e).

Therefore, in order to use the lognormal with the natural base we divide all the data (and the mean,std) with the 'a' coefficient.

Important Notes:
  1. See that the a factor in v/a and Effect[0]/a actually cancel.
  2. lognorm calculates log(x/scale)/sigma --> log(x/scale)= log10(x/scale)/a But we use sigma/a ==> and we get that lognorm.cdf(x/a,sigma/a,scale/a) = lognorm.cdf(x,sigma/a,scale) = Phi(log10(x/scale)/a / (sigma/a) ) = Phi(log10(x/scale)/sigma ) which is what we got from the manuals (and see Ziv).
Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def getPercent(self,ToxicLoad):
	"""
        Using lognormal distribution with base 10.

        In that case define a new probable variable:

        Y such that Y = X/a. (X is the original dose in base 10, Y is in natural base).
        Then, X_avg/a =  Y_avg and  Sig_x/a = Sig_y.

        Where a is the transformation between the bases:
        10^(x) = exp(y) --> x = log10(e)*y --> a = log10(e).

        Therefore, in order to use the lognormal with the natural base we divide
        all the data (and the mean,std) with the 'a' coefficient.

        Important Notes:
        ---------------
        1. See that the a factor in v/a and Effect[0]/a actually cancel.
        2. lognorm calculates log(x/scale)/sigma --> log(x/scale)=  log10(x/scale)/a
           But we use sigma/a ==> and we get that lognorm.cdf(x/a,sigma/a,scale/a) = lognorm.cdf(x,sigma/a,scale) = Phi(log10(x/scale)/a / (sigma/a) )
                                  = Phi(log10(x/scale)/sigma ) which is what we got from the manuals (and see Ziv).

	"""
	a = numpy.log10(numpy.e)
	prob = lognorm.cdf(ToxicLoad/a,self.sigma/a,scale=self.TL_50.m_as(dosage)/a)
	return prob
getToxicLoad(Percent)

Return the toxic load corresponding to a given injury fraction.

Parameters:

Name Type Description Default
Percent float or array_like 
Desired injury fraction(s) (0-1).
required

Returns:

Type Description
float or ndarray
Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def getToxicLoad(self,Percent):
	"""Return the toxic load corresponding to a given injury fraction.

	Parameters
	----------
	Percent : float or array_like
		Desired injury fraction(s) (0-1).

	Returns
	-------
	float or numpy.ndarray
	"""
	a = numpy.log10(numpy.e)
	percent = lognorm.ppf(Percent,self.sigma/a,scale=self.TL_50.m_as(dosage)/a)*a
	return percent
toJSON()

Serialize the lognormal dose-response level to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
295
296
297
298
299
300
301
def toJSON(self):
	"""Serialize the lognormal dose-response level to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'logNormalBase10'
	ret['TL_50'] = str(self.TL_50)
	ret['sigma'] = str(self.sigma)
	return ret

InjuryLevelThreshold

Bases: InjuryLevel

Injury level defined by a single concentration threshold.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class InjuryLevelThreshold(InjuryLevel):
	"""Injury level defined by a single concentration threshold."""

	@property
	def threshold(self):
		"""Concentration threshold value.

		Returns
		-------
		pint.Quantity
		"""
		return self._parameters["threshold"]

	def __init__(self,name,units=None,**parameters):
		"""Initialize a threshold-based injury level.

		Parameters
		----------
		name : str
			Name of the injury level.
		units : pint.Unit or None, optional
			Concentration units; defaults to mg/m^3.
		**parameters
			Must include ``threshold``.
		"""
		actualunit = ureg.mg/ureg.m**3 if units is None else units
		if "threshold" not in parameters: 
			raise ValueError("Cannot find the threshold")

		thr = ureg(parameters["threshold"])

		parameters["threshold"] = ureg(parameters["threshold"]).to(actualunit)

		super(InjuryLevelThreshold,self).__init__(name,units=actualunit,**parameters)

	def getPercent(self,ToxicLoad):
		"""Return 1 if toxic load exceeds threshold, else 0."""
		threshold = self.threshold
		ToxicLoad = numpy.atleast_1d(ToxicLoad)
		ret = numpy.array([ 1 if tounit(x,self.units) > threshold else 0 for x in ToxicLoad])
		return ret[0] if len(ToxicLoad)==1 else ret

	def _getGeopandas(self,concentrationField,x,y,**parameters):
		"""
			Return the correct geopandas of the  Injury level
		"""
		level = unumToPint(self.threshold).magnitude

		CS =  self.getContoursFromLevels(concentrationField, x, y, level)
		if numpy.max(CS.levels) < level:
			ret = geopandas.GeoDataFrame()
		else:
			ret = toGeopandas(CS)
		return ret

	def toJSON(self):
		"""Serialize the threshold injury level to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'threshold'
		ret['threshold'] = str(self.threshold)
		return ret
threshold property

Concentration threshold value.

Returns:

Type Description
Quantity
__init__(name, units=None, **parameters)

Initialize a threshold-based injury level.

Parameters:

Name Type Description Default
name str 
Name of the injury level.
required
units Unit or None 
Concentration units; defaults to mg/m^3.
None
**parameters 
Must include ``threshold``.
{}
Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def __init__(self,name,units=None,**parameters):
	"""Initialize a threshold-based injury level.

	Parameters
	----------
	name : str
		Name of the injury level.
	units : pint.Unit or None, optional
		Concentration units; defaults to mg/m^3.
	**parameters
		Must include ``threshold``.
	"""
	actualunit = ureg.mg/ureg.m**3 if units is None else units
	if "threshold" not in parameters: 
		raise ValueError("Cannot find the threshold")

	thr = ureg(parameters["threshold"])

	parameters["threshold"] = ureg(parameters["threshold"]).to(actualunit)

	super(InjuryLevelThreshold,self).__init__(name,units=actualunit,**parameters)
getPercent(ToxicLoad)

Return 1 if toxic load exceeds threshold, else 0.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
341
342
343
344
345
346
def getPercent(self,ToxicLoad):
	"""Return 1 if toxic load exceeds threshold, else 0."""
	threshold = self.threshold
	ToxicLoad = numpy.atleast_1d(ToxicLoad)
	ret = numpy.array([ 1 if tounit(x,self.units) > threshold else 0 for x in ToxicLoad])
	return ret[0] if len(ToxicLoad)==1 else ret
toJSON()

Serialize the threshold injury level to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
361
362
363
364
365
366
def toJSON(self):
	"""Serialize the threshold injury level to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'threshold'
	ret['threshold'] = str(self.threshold)
	return ret

InjuryLevelExponential

Bases: InjuryLevel

Injury level using an exponential dose-response model.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
class InjuryLevelExponential(InjuryLevel):
	"""Injury level using an exponential dose-response model."""

	@property
	def k(self):
		"""Exponential rate constant.

		Returns
		-------
		float
		"""
		return self._parameters["k"]

	def __init__(self,name,**parameters):
		"""Initialize an exponential injury level.

		Parameters
		----------
		name : str
			Name of the injury level.
		**parameters
			Must include ``k`` (the rate constant).
		"""

		parameters["k"] = float(parameters["k"])

		super().__init__(name,**parameters)

	def getPercent(self,ToxicLoad):
		"""Return the fraction affected using ``1 - exp(-k * ToxicLoad)``."""
		ret = 1-numpy.exp(-self.k*numpy.array(ToxicLoad))
		return ret 

	def _getGeopandas(self,concentrationField,x,y,**parameters):
		"""
			Return the correct geopandas of the  Injury level 
		"""
		level = self.k
		CS =  self.getContoursFromLevels(concentrationField, x, y, level)
		if numpy.max(CS.levels) < level: 
			ret = geopandas.GeoDataFrame()
		else:
			ret = toGeopandas(CS)

		return ret


	def toJSON(self):
		"""Serialize the exponential injury level to a JSON-compatible dictionary."""
		ret = super().toJSON()
		ret['type'] = 'exponential'
		ret['k'] = str(self.k)
		return ret
k property

Exponential rate constant.

Returns:

Type Description
float
__init__(name, **parameters)

Initialize an exponential injury level.

Parameters:

Name Type Description Default
name str 
Name of the injury level.
required
**parameters 
Must include ``k`` (the rate constant).
{}
Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
def __init__(self,name,**parameters):
	"""Initialize an exponential injury level.

	Parameters
	----------
	name : str
		Name of the injury level.
	**parameters
		Must include ``k`` (the rate constant).
	"""

	parameters["k"] = float(parameters["k"])

	super().__init__(name,**parameters)
getPercent(ToxicLoad)

Return the fraction affected using 1 - exp(-k * ToxicLoad).

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
398
399
400
401
def getPercent(self,ToxicLoad):
	"""Return the fraction affected using ``1 - exp(-k * ToxicLoad)``."""
	ret = 1-numpy.exp(-self.k*numpy.array(ToxicLoad))
	return ret
toJSON()

Serialize the exponential injury level to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/InjuryLevel.py 
417
418
419
420
421
422
def toJSON(self):
	"""Serialize the exponential injury level to a JSON-compatible dictionary."""
	ret = super().toJSON()
	ret['type'] = 'exponential'
	ret['k'] = str(self.k)
	return ret

Injury

hera.riskassessment.agents.effects.Injury

InjuryFactory

Bases: object

Factory for creating Injury instances from JSON configuration.

Resolves the injury type and calculator classes by name, instantiates them with the supplied parameters, and returns a fully configured Injury object.

Source code in hera/riskassessment/agents/effects/Injury.py 
 7
 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
class InjuryFactory(object):
	"""Factory for creating ``Injury`` instances from JSON configuration.

	Resolves the injury type and calculator classes by name, instantiates
	them with the supplied parameters, and returns a fully configured
	``Injury`` object.
	"""

	_name = None

	@property
	def name(self):
		"""Name of this injury factory.

		Returns
		-------
		str
		"""
		return self._name

	def getInjury(self,name,cfgJSON,**additionalparameters): 
		"""
			Creates the appropriate injury 

			Structure: 
			{
				"type" : <name> . Appends 'pyriskassessment.agents.effects.InjuryLevel.Injury' to the name.
				"calculator" :{ <calculatorName> : {"time" : "datetime"} }  Appends 'pyriskassessment.agents.effects.Calculator.Calculator' to the name.
				"parameters": { 
					... Injury level json. 
				}
			}

			additionalparameters: 
			=====================
				These are parameters that might be used by the calculator

				For example the tenberge coefficient. 
		"""		
		try:
			injuryType = cfgJSON["type"]
		except KeyError:
			raise ValueError("Injury type is not defined")

		try:
			calculatorTypeAndParams = cfgJSON["calculator"]
			calcType,calcParam = [x for x in calculatorTypeAndParams.items()][0]
		except KeyError: 
			raise ValueError("Calculator not defined")

		calculatorCLS = pydoc.locate("hera.riskassessment.agents.effects.Calculator.Calculator%s" % calcType)
		calculator    = calculatorCLS(**calcParam,**additionalparameters)

		injuryCLS = pydoc.locate("hera.riskassessment.agents.effects.Injury.Injury%s" % injuryType)

		if injuryCLS is None: 

			injuryExists  = ",".join([x[6:] for x in dir(pydoc.locate("hera.riskassessment.agents.effects.Injury")) if x.startswith("Injury")])

			raise NotImplementedError("The injury %s is not defined. Injuries found: %s  " % (injuryType,injuryExists))

		return injuryCLS(name,cfgJSON["parameters"],calculator,units=cfgJSON.get("units"))
name property

Name of this injury factory.

Returns:

Type Description
str
getInjury(name, cfgJSON, **additionalparameters)

Creates the appropriate injury

Structure: { "type" : . Appends 'pyriskassessment.agents.effects.InjuryLevel.Injury' to the name. "calculator" :{ : {"time" : "datetime"} } Appends 'pyriskassessment.agents.effects.Calculator.Calculator' to the name. "parameters": { ... Injury level json. } }

additionalparameters:
    These are parameters that might be used by the calculator

    For example the tenberge coefficient.
Source code in hera/riskassessment/agents/effects/Injury.py 
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
def getInjury(self,name,cfgJSON,**additionalparameters): 
	"""
		Creates the appropriate injury 

		Structure: 
		{
			"type" : <name> . Appends 'pyriskassessment.agents.effects.InjuryLevel.Injury' to the name.
			"calculator" :{ <calculatorName> : {"time" : "datetime"} }  Appends 'pyriskassessment.agents.effects.Calculator.Calculator' to the name.
			"parameters": { 
				... Injury level json. 
			}
		}

		additionalparameters: 
		=====================
			These are parameters that might be used by the calculator

			For example the tenberge coefficient. 
	"""		
	try:
		injuryType = cfgJSON["type"]
	except KeyError:
		raise ValueError("Injury type is not defined")

	try:
		calculatorTypeAndParams = cfgJSON["calculator"]
		calcType,calcParam = [x for x in calculatorTypeAndParams.items()][0]
	except KeyError: 
		raise ValueError("Calculator not defined")

	calculatorCLS = pydoc.locate("hera.riskassessment.agents.effects.Calculator.Calculator%s" % calcType)
	calculator    = calculatorCLS(**calcParam,**additionalparameters)

	injuryCLS = pydoc.locate("hera.riskassessment.agents.effects.Injury.Injury%s" % injuryType)

	if injuryCLS is None: 

		injuryExists  = ",".join([x[6:] for x in dir(pydoc.locate("hera.riskassessment.agents.effects.Injury")) if x.startswith("Injury")])

		raise NotImplementedError("The injury %s is not defined. Injuries found: %s  " % (injuryType,injuryExists))

	return injuryCLS(name,cfgJSON["parameters"],calculator,units=cfgJSON.get("units"))

Injury

Bases: object

Holds a list of injury levels

Source code in hera/riskassessment/agents/effects/Injury.py 
 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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
class Injury(object): 
	"""
                Holds a list of injury levels 
	"""

	_name = None
	_levelsmap = None   # just a map of name->injury.
	_levels = None 
	_calculator = None

	@property
	def calculator(self):
		"""The calculator used for toxic load computation.

		Returns
		-------
		AbstractCalculator
		"""
		return self._calculator

	@property
	def levels(self):
		"""Ordered list of injury levels.

		Returns
		-------
		list of InjuryLevel
		"""
		return self._levels

	@property
	def levelNames(self):
		"""Names of all injury levels.

		Returns
		-------
		list of str
		"""
		return [x.name for x in self._levels]

	def __getitem__(self,name):
		"""Access an injury level by name."""
		return self._levelsmap[name]

	def __init__(self,name,cfgJSON,calculator,units=None):
		"""
			Loads the relevant injury levels from the JSONcfg

			{

				"type" : <injuryLevel type>.  Appends 'pyriskassessment.casualties.InjuryLevel.InjuryLevel' to the name. 
				"levels" : [....],
			        parameters: { 
					     "InjuryName" : { injury parameters },
						.
						.			

			}

		"""
		self._name = name
		self._calculator = calculator

		injuryType = cfgJSON.get("type",None)
		if (injuryType is None):
			raise ValueError("InjuryLevel type is nor defined")

		injuryCLS = pydoc.locate("hera.riskassessment.agents.effects.InjuryLevel.InjuryLevel%s" % injuryType)
		self._levelsmap = {}
		self._levels = []
		levelNames = cfgJSON.get("levels")

		if units is not None:
			units = ureg(units)

		for lvl in levelNames:
			try:
				lvlparams = cfgJSON["parameters"][lvl].copy()
			except KeyError:
				raise ValueError("parameters for level %s not found !" % lvl)

			curindex = levelNames.index(lvl)
			lvlparams["higher_severity"] = None if curindex == 0 else self.levels[curindex-1]

			injry = injuryCLS(lvl, units=units, **lvlparams)
			self._levels.append(injry)
			self._levelsmap[lvl] = injry



	def getPercent(self,level, ToxicLoad):
		"""
			calculate the toxic load and subtract the percent from the percent above. 
		"""
		severityList = self.levelNames
		curindex     = severityList.index(level)

		if (curindex == 0): 
			val = self.levels[curindex].getPercent(ToxicLoad) 
		else: 
			val = self.levels[curindex].getPercent(ToxicLoad)  - self.levels[curindex-1].getPercent(ToxicLoad)
		return val

	def _postCalculate(self,retList,time):
		"""
			Apply some post calculations on the results.

			The post calculation depends on the injury type, and therefore implemented in the inherited classes.

			Parameters
			----------
				retList : list of pandas.
						The values of the calculation in different levels.

				time : str
						The name of the time columns.

		"""
		raise NotImplementedError("Abstract class")

	def _postCalculatePointWise(self, retList):
		"""
			apply some post calculations on the results

		"""
		pass

	def calculate(self, concentrationField, field, time="datetime", x="x", y="y", breathingRate=10 * ureg.L / ureg.min,
				  sel={},isel={}):
		"""Deprecated. Use ``calculateRegionOfInjured`` instead."""
		import warnings
		warnings.warn("This function is obselete. Use calculateRegionOfInjured instead")
		return self.calculateRegionOfInjured(concentrationField=concentrationField,
											  field=field,
											  time=time,
											  x=x,
											  y=y,
											  breathingRate=breathingRate,
											  sel=sel,isel=isel)

	def calculateRegionOfInjured(self,concentrationField,field,time="datetime",x="x",y="y",breathingRate=10*ureg.L / ureg.min,sel={},isel={}):
		"""
			Calculates the fraction of the population that was effected in each point, and returns its contour.
			The levels are determined by the injury type.

			Parameters
			----------
			concentrationField:
				an xarray with the concentrations at each time.

			time: str
				The name time dimension. if None then don't look up for time (assume it is without time).

			x:  str
				The name x dimension.

			y:  str
				The name of the y dimension.

			breathingRate:
					The breathing rate used. The default is man at Rest.

			parameters: kwargs
					Additional parameters that are needed for the calculations.


			Returns
			--------

            	A thresholdGeoDataFrame with the columns:

				time |  injury name | total polygon  | diff polygon                             |       addtional fields
					 |              | the total area | a differance from the level above.       |       if necessary.
		"""

		retList = [] 

		toxicLoads = self.calculateToxicLoads(concentrationField=concentrationField,
											  time=time,
											  breathingRate=breathingRate,
											  field=field).sel(**sel).isel(**isel).compute()
		for lvl in self.levels:
			data = lvl.calculateContours(toxicLoads=toxicLoads,time=time,x=x,y=y)
			if data is not None: 
				retList.append(data)

		ret = self._postCalculate(retList,time)
		return thresholdGeoDataFrame(ret)

	def calculateToxicLoads(self,concentrationField,time="datetime",breathingRate=10*ureg.L / ureg.min,field=None):
		"""
			Calculates the toxic loads of the concetration field.

			Paramters
			---------
					concentrationField : pandas or xarray.
							The concentrations to calculate.

							If xarray, then we assume that it is a 2D map with an additional 'time' coordinate.
							If pandas, then we assume that it is a 'time' in the index and each other column is a point to calculate.

					time : str
							The name of the time column (or the name of the index column if it is pandas).

					breathingRate: unum [L/min].
							The breathing rate used. The default is man at Rest is 10L/min.

					field : str
							If xarray, the name of the field to use.
							ignored if pandas

		"""
		return self.calculator.calculate(concentrationField, field, breathingRate=breathingRate, time=time)


	def calculatePointWiseFractionInjured(self,timeConcentration,time="datetime",breathingRate=10*ureg.L / ureg.min,field=None):
		"""
			Calculates the fraction of injury over time in each point.

			Can be used with pandas.DataFrame or  xarray.Dataset [not implemented yet].

			Parameters
			----------

				timeConcentration : pandas.DataFrame, or xarray.DataFrame.
							Holds the concentration in time.

							If xarray, also has a 'time' coordinate that will be calculated. [not implemented yet]

							If DataFrame:
								Each point is represeneted by a column, and the time is an index (with the name 'datetime').

								So the structure of the input is :
									  P1   P2   P3
								time
								00:00 0     0   0
								00:01 0.1   0.1 0.01
								00:02 0.4   0.1 0.01
								00:05 0.5   0.1 0.01

		    time  : str
		    			The name of the time column (or the name of the index).

		    breathingRate : unum, L/min
		    			The breathing rate of the population.

		    parameters: kwargs.
		    			Additional parameter to the calculator (for example ten-berge coefficient).

						selection parameters (xarray only):

							sel - select according to the coordinates (see sel funcion of the xarray).
							isel - select according to the coordinate index (see isel function of the xarray).

		"""




		# 2. For each injury level:
		#      Create a dataframe with the fields:
		#
		#					P1        injury
		#		datetime
		#		  0  	[fraction]        level 1 name
		#		  1  	[fraction]        level 1 name
		#					...

		if not isinstance(timeConcentration,pandas.DataFrame):
			raise ValueError("Still not implemented....")

		# 1. Calculate the toxic load for each point.
		toxicLoads = self.calculateToxicLoads(concentrationField=timeConcentration,
											  time=time,
											  breathingRate=breathingRate,
											  field=field)
		retList = []
		for lvl in self.levels:
			data = pandas.DataFrame()

			for device in toxicLoads:
				prct = toxicLoads[device].apply(lambda x: self.getPercent(lvl.name,x))
				data = prct.to_frame("injuryPercent").assign(deviceName=device,level=lvl.name)
				if data is not None:
					retList.append(data)

		ret = pandas.concat(retList)
		return ret


	def calculateThresholdPolygon(self,data,time):
		"""
			Calculates the diff of the polygon based on the toxic load. 
		"""
		ret = data.sort_values("ToxicLoad")
		polyList = []
		indexList = []
		for timeseries in ret.groupby(time):
			timedata = timeseries[1].sort_values("ToxicLoad",ascending=False)
			polyList.append(timedata.iloc[0]["TotalPolygon"])
			indexList.append(timedata.index[0]) 
			for curPolyIndex,prevPolyIndex in zip(timedata.index[1:],timedata.index[:-1]): 
				curPoly = ret.loc[curPolyIndex,"TotalPolygon"] 
				prevPoly = ret.loc[prevPolyIndex,"TotalPolygon"] 
				polyList.append(curPoly.difference(prevPoly))
				indexList.append(curPolyIndex)

		diff = pandas.DataFrame({"ThresholdPolygon" : polyList},index=indexList) 
		ret = data.merge(diff,left_index=True,right_index=True)		

		return ret 


	def toJSON(self):
		"""Serialize the injury to a JSON-compatible dictionary."""
		ret = dict()
		#ret['name']  = self._name
		ret['levels'] = dict([(lvl.name,lvl.toJSON()) for lvl in self.levels])
		ret['calculator'] = self._calculator.toJSON()
		return ret

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

The calculator used for toxic load computation.

Returns:

Type Description
AbstractCalculator
levels property

Ordered list of injury levels.

Returns:

Type Description
list of InjuryLevel
levelNames property

Names of all injury levels.

Returns:

Type Description
list of str
__getitem__(name)

Access an injury level by name.

Source code in hera/riskassessment/agents/effects/Injury.py 
110
111
112
def __getitem__(self,name):
	"""Access an injury level by name."""
	return self._levelsmap[name]
__init__(name, cfgJSON, calculator, units=None)

Loads the relevant injury levels from the JSONcfg

{

    "type" : <injuryLevel type>.  Appends 'pyriskassessment.casualties.InjuryLevel.InjuryLevel' to the name. 
    "levels" : [....],
    parameters: { 
                 "InjuryName" : { injury parameters },
                    .
                    .

}

Source code in hera/riskassessment/agents/effects/Injury.py 
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
def __init__(self,name,cfgJSON,calculator,units=None):
	"""
		Loads the relevant injury levels from the JSONcfg

		{

			"type" : <injuryLevel type>.  Appends 'pyriskassessment.casualties.InjuryLevel.InjuryLevel' to the name. 
			"levels" : [....],
		        parameters: { 
				     "InjuryName" : { injury parameters },
					.
					.			

		}

	"""
	self._name = name
	self._calculator = calculator

	injuryType = cfgJSON.get("type",None)
	if (injuryType is None):
		raise ValueError("InjuryLevel type is nor defined")

	injuryCLS = pydoc.locate("hera.riskassessment.agents.effects.InjuryLevel.InjuryLevel%s" % injuryType)
	self._levelsmap = {}
	self._levels = []
	levelNames = cfgJSON.get("levels")

	if units is not None:
		units = ureg(units)

	for lvl in levelNames:
		try:
			lvlparams = cfgJSON["parameters"][lvl].copy()
		except KeyError:
			raise ValueError("parameters for level %s not found !" % lvl)

		curindex = levelNames.index(lvl)
		lvlparams["higher_severity"] = None if curindex == 0 else self.levels[curindex-1]

		injry = injuryCLS(lvl, units=units, **lvlparams)
		self._levels.append(injry)
		self._levelsmap[lvl] = injry
getPercent(level, ToxicLoad)

calculate the toxic load and subtract the percent from the percent above.

Source code in hera/riskassessment/agents/effects/Injury.py 
160
161
162
163
164
165
166
167
168
169
170
171
def getPercent(self,level, ToxicLoad):
	"""
		calculate the toxic load and subtract the percent from the percent above. 
	"""
	severityList = self.levelNames
	curindex     = severityList.index(level)

	if (curindex == 0): 
		val = self.levels[curindex].getPercent(ToxicLoad) 
	else: 
		val = self.levels[curindex].getPercent(ToxicLoad)  - self.levels[curindex-1].getPercent(ToxicLoad)
	return val
calculate(concentrationField, field, time='datetime', x='x', y='y', breathingRate=10 * ureg.L / ureg.min, sel={}, isel={})

Deprecated. Use calculateRegionOfInjured instead.

Source code in hera/riskassessment/agents/effects/Injury.py 
197
198
199
200
201
202
203
204
205
206
207
208
def calculate(self, concentrationField, field, time="datetime", x="x", y="y", breathingRate=10 * ureg.L / ureg.min,
			  sel={},isel={}):
	"""Deprecated. Use ``calculateRegionOfInjured`` instead."""
	import warnings
	warnings.warn("This function is obselete. Use calculateRegionOfInjured instead")
	return self.calculateRegionOfInjured(concentrationField=concentrationField,
										  field=field,
										  time=time,
										  x=x,
										  y=y,
										  breathingRate=breathingRate,
										  sel=sel,isel=isel)
calculateRegionOfInjured(concentrationField, field, time='datetime', x='x', y='y', breathingRate=10 * ureg.L / ureg.min, sel={}, isel={}) 
    Calculates the fraction of the population that was effected in each point, and returns its contour.
    The levels are determined by the injury type.

    Parameters

    concentrationField:
            an xarray with the concentrations at each time.

    time: str
            The name time dimension. if None then don't look up for time (assume it is without time).

    x:  str
            The name x dimension.

    y:  str
            The name of the y dimension.

    breathingRate:
                    The breathing rate used. The default is man at Rest.

    parameters: kwargs
                    Additional parameters that are needed for the calculations.

    Returns

A thresholdGeoDataFrame with the columns:

            time |  injury name | total polygon  | diff polygon                             |       addtional fields
                     |              | the total area | a differance from the level above.       |       if necessary.
Source code in hera/riskassessment/agents/effects/Injury.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
	def calculateRegionOfInjured(self,concentrationField,field,time="datetime",x="x",y="y",breathingRate=10*ureg.L / ureg.min,sel={},isel={}):
		"""
			Calculates the fraction of the population that was effected in each point, and returns its contour.
			The levels are determined by the injury type.

			Parameters
			----------
			concentrationField:
				an xarray with the concentrations at each time.

			time: str
				The name time dimension. if None then don't look up for time (assume it is without time).

			x:  str
				The name x dimension.

			y:  str
				The name of the y dimension.

			breathingRate:
					The breathing rate used. The default is man at Rest.

			parameters: kwargs
					Additional parameters that are needed for the calculations.


			Returns
			--------

            	A thresholdGeoDataFrame with the columns:

				time |  injury name | total polygon  | diff polygon                             |       addtional fields
					 |              | the total area | a differance from the level above.       |       if necessary.
		"""

		retList = [] 

		toxicLoads = self.calculateToxicLoads(concentrationField=concentrationField,
											  time=time,
											  breathingRate=breathingRate,
											  field=field).sel(**sel).isel(**isel).compute()
		for lvl in self.levels:
			data = lvl.calculateContours(toxicLoads=toxicLoads,time=time,x=x,y=y)
			if data is not None: 
				retList.append(data)

		ret = self._postCalculate(retList,time)
		return thresholdGeoDataFrame(ret)
calculateToxicLoads(concentrationField, time='datetime', breathingRate=10 * ureg.L / ureg.min, field=None)

Calculates the toxic loads of the concetration field.

Paramters 
            concentrationField : pandas or xarray.
                            The concentrations to calculate.

                            If xarray, then we assume that it is a 2D map with an additional 'time' coordinate.
                            If pandas, then we assume that it is a 'time' in the index and each other column is a point to calculate.

            time : str
                            The name of the time column (or the name of the index column if it is pandas).

            breathingRate: unum [L/min].
                            The breathing rate used. The default is man at Rest is 10L/min.

            field : str
                            If xarray, the name of the field to use.
                            ignored if pandas
Source code in hera/riskassessment/agents/effects/Injury.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def calculateToxicLoads(self,concentrationField,time="datetime",breathingRate=10*ureg.L / ureg.min,field=None):
	"""
		Calculates the toxic loads of the concetration field.

		Paramters
		---------
				concentrationField : pandas or xarray.
						The concentrations to calculate.

						If xarray, then we assume that it is a 2D map with an additional 'time' coordinate.
						If pandas, then we assume that it is a 'time' in the index and each other column is a point to calculate.

				time : str
						The name of the time column (or the name of the index column if it is pandas).

				breathingRate: unum [L/min].
						The breathing rate used. The default is man at Rest is 10L/min.

				field : str
						If xarray, the name of the field to use.
						ignored if pandas

	"""
	return self.calculator.calculate(concentrationField, field, breathingRate=breathingRate, time=time)
calculatePointWiseFractionInjured(timeConcentration, time='datetime', breathingRate=10 * ureg.L / ureg.min, field=None) 
Calculates the fraction of injury over time in each point.

Can be used with pandas.DataFrame or  xarray.Dataset [not implemented yet].

Parameters

        timeConcentration : pandas.DataFrame, or xarray.DataFrame.
                                Holds the concentration in time.

                                If xarray, also has a 'time' coordinate that will be calculated. [not implemented yet]

                                If DataFrame:
                                        Each point is represeneted by a column, and the time is an index (with the name 'datetime').

                                        So the structure of the input is :
                                                  P1   P2   P3
                                        time
                                        00:00 0     0   0
                                        00:01 0.1   0.1 0.01
                                        00:02 0.4   0.1 0.01
                                        00:05 0.5   0.1 0.01

time : str The name of the time column (or the name of the index).

breathingRate : unum, L/min The breathing rate of the population.

parameters: kwargs. Additional parameter to the calculator (for example ten-berge coefficient).

                        selection parameters (xarray only):

                                sel - select according to the coordinates (see sel funcion of the xarray).
                                isel - select according to the coordinate index (see isel function of the xarray).
Source code in hera/riskassessment/agents/effects/Injury.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def calculatePointWiseFractionInjured(self,timeConcentration,time="datetime",breathingRate=10*ureg.L / ureg.min,field=None):
	"""
		Calculates the fraction of injury over time in each point.

		Can be used with pandas.DataFrame or  xarray.Dataset [not implemented yet].

		Parameters
		----------

			timeConcentration : pandas.DataFrame, or xarray.DataFrame.
						Holds the concentration in time.

						If xarray, also has a 'time' coordinate that will be calculated. [not implemented yet]

						If DataFrame:
							Each point is represeneted by a column, and the time is an index (with the name 'datetime').

							So the structure of the input is :
								  P1   P2   P3
							time
							00:00 0     0   0
							00:01 0.1   0.1 0.01
							00:02 0.4   0.1 0.01
							00:05 0.5   0.1 0.01

	    time  : str
	    			The name of the time column (or the name of the index).

	    breathingRate : unum, L/min
	    			The breathing rate of the population.

	    parameters: kwargs.
	    			Additional parameter to the calculator (for example ten-berge coefficient).

					selection parameters (xarray only):

						sel - select according to the coordinates (see sel funcion of the xarray).
						isel - select according to the coordinate index (see isel function of the xarray).

	"""




	# 2. For each injury level:
	#      Create a dataframe with the fields:
	#
	#					P1        injury
	#		datetime
	#		  0  	[fraction]        level 1 name
	#		  1  	[fraction]        level 1 name
	#					...

	if not isinstance(timeConcentration,pandas.DataFrame):
		raise ValueError("Still not implemented....")

	# 1. Calculate the toxic load for each point.
	toxicLoads = self.calculateToxicLoads(concentrationField=timeConcentration,
										  time=time,
										  breathingRate=breathingRate,
										  field=field)
	retList = []
	for lvl in self.levels:
		data = pandas.DataFrame()

		for device in toxicLoads:
			prct = toxicLoads[device].apply(lambda x: self.getPercent(lvl.name,x))
			data = prct.to_frame("injuryPercent").assign(deviceName=device,level=lvl.name)
			if data is not None:
				retList.append(data)

	ret = pandas.concat(retList)
	return ret
calculateThresholdPolygon(data, time)

Calculates the diff of the polygon based on the toxic load.

Source code in hera/riskassessment/agents/effects/Injury.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def calculateThresholdPolygon(self,data,time):
	"""
		Calculates the diff of the polygon based on the toxic load. 
	"""
	ret = data.sort_values("ToxicLoad")
	polyList = []
	indexList = []
	for timeseries in ret.groupby(time):
		timedata = timeseries[1].sort_values("ToxicLoad",ascending=False)
		polyList.append(timedata.iloc[0]["TotalPolygon"])
		indexList.append(timedata.index[0]) 
		for curPolyIndex,prevPolyIndex in zip(timedata.index[1:],timedata.index[:-1]): 
			curPoly = ret.loc[curPolyIndex,"TotalPolygon"] 
			prevPoly = ret.loc[prevPolyIndex,"TotalPolygon"] 
			polyList.append(curPoly.difference(prevPoly))
			indexList.append(curPolyIndex)

	diff = pandas.DataFrame({"ThresholdPolygon" : polyList},index=indexList) 
	ret = data.merge(diff,left_index=True,right_index=True)		

	return ret
toJSON()

Serialize the injury to a JSON-compatible dictionary.

Source code in hera/riskassessment/agents/effects/Injury.py 
383
384
385
386
387
388
389
def toJSON(self):
	"""Serialize the injury to a JSON-compatible dictionary."""
	ret = dict()
	#ret['name']  = self._name
	ret['levels'] = dict([(lvl.name,lvl.toJSON()) for lvl in self.levels])
	ret['calculator'] = self._calculator.toJSON()
	return ret
__str__()

Return a JSON string representation of the injury.

Source code in hera/riskassessment/agents/effects/Injury.py 
391
392
393
def __str__(self):
	"""Return a JSON string representation of the injury."""
	json.dumps(self.toJSON(),indent=4)

InjuryLognormal10

Bases: Injury

Injury using base-10 lognormal dose-response levels.

Source code in hera/riskassessment/agents/effects/Injury.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
class InjuryLognormal10(Injury):
	"""Injury using base-10 lognormal dose-response levels."""

	def _postCalculate(self,retList,time):
		"""
                        Fill in the actual % that was effected. 

		"""
		modList = []
		for vals in retList: 
			vals["percentEffected"] = vals.apply(lambda x: self.getPercent(x['severity'],x['ToxicLoad']),axis=1)
			vals = self.calculateThresholdPolygon(vals,time)
			modList.append(vals)

		return pandas.concat(modList,ignore_index=True)

InjuryThreshold

Bases: Injury

Injury using threshold-based levels.

Source code in hera/riskassessment/agents/effects/Injury.py 
415
416
417
418
419
420
421
422
423
424
425
426
427
428
class InjuryThreshold(Injury):
	"""Injury using threshold-based levels."""

	def _postCalculate(self,retList,time):
		"""
                        Calculate the percent Effected and calculate the differential polygons. 
		"""
		if len(retList) > 0: 
			ret = pandas.concat(retList,ignore_index=True)
			ret = self.calculateThresholdPolygon(ret,time)
			ret['percentEffected'] = 1.
		else: 
			ret = None
		return ret

InjuryExponential

Bases: Injury

Injury using exponential dose-response levels.

Source code in hera/riskassessment/agents/effects/Injury.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
class InjuryExponential(Injury):
	"""Injury using exponential dose-response levels."""

	def _postCalculate(self,retList,time):
		"""
                        Calculate the percent Effected and calculate the differential polygons. 
		"""
		modList = []
		for vals in retList: 
			vals["percentEffected"] = vals.apply(lambda x: self.getPercent(x['severity'],x['ToxicLoad']),axis=1)
			vals = self.calculateThresholdPolygon(vals,time)
			modList.append(vals)

		return pandas.concat(modList,ignore_index=True)