Merge pull request #10268 from gem/agg_values

Added a new output "Aggregated Exposure Values"

Author: micheles

debian/changelog

@@ -1,3 +1,6 @@
+  [Michele Simionato]
+  * Added a new output "Aggregated Exposure Values"
+
 python3-oq-engine (3.22.0-1~xenial01) xenial; urgency=low
 
   [Michele Simionato]

openquake/calculators/base.py

@@ -1553,7 +1553,7 @@ def save_agg_values(dstore, assetcol, lossnames, aggby, maxagg):
     if aggby:
         aggids, aggtags = assetcol.build_aggids(aggby, maxagg)
         logging.info('Storing %d aggregation keys', len(aggids))
-        agg_keys = [','.join(tags) for tags in aggtags]
+        agg_keys = ['\t'.join(tags) for tags in aggtags]
         dstore['agg_keys'] = numpy.array(agg_keys, hdf5.vstr)
         if 'assetcol' not in set(dstore):
             dstore['assetcol'] = assetcol

openquake/calculators/export/__init__.py

@@ -42,6 +42,7 @@
     'loss_curves-stats': 'Asset Loss Curves Statistics',
     'loss_maps-rlzs': 'Asset Loss Maps',
     'loss_maps-stats': 'Asset Loss Maps Statistics',
+    'agg_keys': 'Aggregated Exposure Values',
     'aggrisk': 'Aggregate Risk',
     'aggrisk-stats': 'Aggregate Risk Statistics',
     'agg_risk': 'Total Risk',

openquake/calculators/export/risk.py

@@ -594,6 +594,19 @@ def _fix(col):
     return col
 
 
+@export.add(('agg_keys', 'csv'))
+def export_agg_keys_csv(ekey, dstore):
+    """
+    :param ekey: export key, i.e. a pair (datastore key, fmt)
+    :param dstore: datastore object
+    """
+    df = extract(dstore, ekey[0] + '?')
+    writer = writers.CsvWriter(fmt=writers.FIVEDIGITS)
+    fname = dstore.export_path('%s.%s' % ekey)
+    writer.save(df, fname, comment=dstore.metadata)
+    return [fname]
+
+
 @export.add(('aggcurves', 'csv'))
 def export_aggcurves_csv(ekey, dstore):
     """

openquake/calculators/extract.py

@@ -769,7 +769,7 @@ def extract_agg_curves(dstore, what):
     tagvalues = [tagdict[t][0] for t in tagnames]
     if tagnames:
         lst = decode(dstore['agg_keys'][:])
-        agg_id = lst.index(','.join(tagvalues))
+        agg_id = lst.index('\t'.join(tagvalues))
     else:
         agg_id = 0  # total aggregation
     ep_fields = dstore.get_attr('aggcurves', 'ep_fields')
@@ -815,6 +815,26 @@ def extract_agg_curves(dstore, what):
     return ArrayWrapper(arr, dict(json=hdf5.dumps(attrs)))
 
 
+@extract.add('agg_keys')
+def extract_agg_keys(dstore, what):
+    """
+    Aggregate the exposure values (one for each loss type) by tag. Use it as
+    /extract/agg_keys?
+    """
+    aggby = dstore['oqparam'].aggregate_by[0]
+    keys = numpy.array([line.decode('utf8').split('\t')
+                        for line in dstore['agg_keys'][:]])
+    values = dstore['agg_values'][:-1]  # discard the total aggregation
+    ok = values['structural'] > 0
+    okvalues = values[ok]
+    dic = {}
+    for i, tag in enumerate(aggby):
+        dic[tag] = keys[ok, i]
+    for name in values.dtype.names:
+        dic[name] = okvalues[name]
+    return pandas.DataFrame(dic)
+
+
 @extract.add('agg_losses')
 def extract_agg_losses(dstore, what):
     """

openquake/calculators/tests/event_based_risk_test.py

@@ -99,6 +99,11 @@ def test_case_1_ins(self):
         # this is a case with insured losses and tags
         self.run_calc(case_1.__file__, 'job_ins.ini', concurrent_tasks='4')
 
+        # testing agg_keys
+        [fname] = export(('agg_keys', 'csv'), self.calc.datastore)
+        self.assertEqualFiles('expected/%s' % strip_calc_id(fname), fname,
+                              delta=1E-5)
+
         # testing the view agg_id
         agg_id = view('agg_id', self.calc.datastore)
         self.assertEqual(str(agg_id), '''\

openquake/calculators/views.py

@@ -1524,7 +1524,7 @@ def view_agg_id(token, dstore):
     Show the available aggregations
     """
     [aggby] = dstore['oqparam'].aggregate_by
-    keys = [key.decode('utf8').split(',') for key in dstore['agg_keys'][:]]
+    keys = [key.decode('utf8').split('\t') for key in dstore['agg_keys'][:]]
     keys = numpy.array(keys)  # shape (N, A)
     dic = {aggkey: keys[:, a] for a, aggkey in enumerate(aggby)}
     df = pandas.DataFrame(dic)

openquake/engine/tests/aristotle1/job.ini

@@ -10,3 +10,4 @@ maximum_distance = 600
 number_of_ground_motion_fields = 10
 asset_hazard_distance = 50
 tectonic_region_type = Active Shallow Crust
+aggregate_by = ID_1,OCCUPANCY

openquake/engine/tests/aristotle2/job.ini

@@ -10,3 +10,4 @@ maximum_distance = 400
 number_of_ground_motion_fields = 10
 asset_hazard_distance = 50
 tectonic_region_type = Tectonic_Type_A
+aggregate_by = ID_1,OCCUPANCY

openquake/engine/tests/aristotle3/job.ini

@@ -10,3 +10,4 @@ maximum_distance = 600
 number_of_ground_motion_fields = 10
 asset_hazard_distance = 50
 tectonic_region_type = Shallow Default
+aggregate_by = ID_1,OCCUPANCY