Merge branch '2024-03-24_dmg_process_loc'

zsarnoczay · Mar 28, 2024 · 6b1d6bc · 6b1d6bc
2 parents c1f3d3a + 64ef134
commit 6b1d6bc
Show file tree

Hide file tree

Showing 2 changed files with 71 additions and 16 deletions.
diff --git a/pelicun/model/damage_model.py b/pelicun/model/damage_model.py
@@ -1145,11 +1145,6 @@ def _perform_dmg_task(self, task, ds_sample):
 
                 # clear damage state information
                 elif target_event == 'NA':
-                    if match_locations:
-                        raise ValueError(
-                            'Invalid damage task configuration. Cannot match '
-                            'locations when the target event is set to NA.'
-                        )
                     ds_target = -1
                     # -1 stands for nan (ints don'ts support nan)
 
@@ -1223,13 +1218,19 @@ def _perform_dmg_event_loc(
 
             # affected columns
             if target_cmp == 'ALL':
-                raise ValueError('Cannot combine `-LOC` with `ALL` keywords')
-            column_selection = np.where(
-                np.logical_and(
-                    ds_sample.columns.get_level_values('cmp') == target_cmp,
-                    ds_sample.columns.get_level_values('loc') == loc,
-                )
-            )[0]
+                column_selection = np.where(
+                    np.logical_and(
+                        ds_sample.columns.get_level_values('cmp') != source_cmp,
+                        ds_sample.columns.get_level_values('loc') == loc,
+                    )
+                )[0]
+            else:
+                column_selection = np.where(
+                    np.logical_and(
+                        ds_sample.columns.get_level_values('cmp') == target_cmp,
+                        ds_sample.columns.get_level_values('loc') == loc,
+                    )
+                )[0]
             ds_sample.iloc[row_selection, column_selection] = ds_target
 
     def _get_pg_batches(self, block_batch_size):

diff --git a/pelicun/tests/test_model.py b/pelicun/tests/test_model.py
@@ -1374,10 +1374,6 @@ def test__evaluate_damage_state_and_prepare_dmg_quantities(
 
     def test__perform_dmg_task(self, assessment_instance):
 
-        x = assessment.Assessment()
-        x.log.verbose = False
-        assessment_instance = x
-
         damage_model = assessment_instance.damage
 
         #
@@ -1522,6 +1518,64 @@ def test__perform_dmg_task(self, assessment_instance):
             ('CMP.B', '1', '1', '1'): {0: 1, 1: 0, 2: 0},
         }
 
+        #
+        # NA keyword combined with `-LOC`
+        #
+
+        ds_sample = pd.DataFrame(
+            {
+                ('CMP.A', '1', '1', '0'): [0, 0, 0],
+                ('CMP.A', '2', '1', '0'): [0, 0, 0],
+                ('CMP.B', '1', '1', '0'): [0, 0, 1],
+                ('CMP.B', '2', '1', '0'): [1, 0, 0],
+            },
+            dtype='int32',
+        )
+        ds_sample.columns.names = ['cmp', 'loc', 'dir', 'uid']
+
+        dmg_process = {"1_CMP.B-LOC": {"DS1": "CMP.A_NA"}}
+        for task in dmg_process.items():
+            damage_model._perform_dmg_task(task, ds_sample)
+        after = ds_sample
+
+        assert after.to_dict() == {
+            ('CMP.A', '1', '1', '0'): {0: 0, 1: 0, 2: -1},
+            ('CMP.A', '2', '1', '0'): {0: -1, 1: 0, 2: 0},
+            ('CMP.B', '1', '1', '0'): {0: 0, 1: 0, 2: 1},
+            ('CMP.B', '2', '1', '0'): {0: 1, 1: 0, 2: 0},
+        }
+
+        #
+        # NA keyword combined with `-LOC` and `ALL`
+        #
+
+        ds_sample = pd.DataFrame(
+            {
+                ('CMP.A', '1', '1', '0'): [0, 0, 1],
+                ('CMP.A', '2', '1', '0'): [1, 0, 0],
+                ('CMP.B', '1', '1', '0'): [0, 0, 0],
+                ('CMP.B', '2', '1', '0'): [0, 0, 0],
+                ('CMP.C', '1', '1', '0'): [0, 0, 0],
+                ('CMP.C', '2', '1', '0'): [0, 0, 0],
+            },
+            dtype='int32',
+        )
+        ds_sample.columns.names = ['cmp', 'loc', 'dir', 'uid']
+
+        dmg_process = {"1_CMP.A-LOC": {"DS1": "ALL_NA"}}
+        for task in dmg_process.items():
+            damage_model._perform_dmg_task(task, ds_sample)
+        after = ds_sample
+
+        assert after.to_dict() == {
+            ('CMP.A', '1', '1', '0'): {0: 0, 1: 0, 2: 1},
+            ('CMP.A', '2', '1', '0'): {0: 1, 1: 0, 2: 0},
+            ('CMP.B', '1', '1', '0'): {0: 0, 1: 0, 2: -1},
+            ('CMP.B', '2', '1', '0'): {0: -1, 1: 0, 2: 0},
+            ('CMP.C', '1', '1', '0'): {0: 0, 1: 0, 2: -1},
+            ('CMP.C', '2', '1', '0'): {0: -1, 1: 0, 2: 0},
+        }
+
     def test__get_pg_batches_1(self, assessment_instance):
         damage_model = assessment_instance.damage
         asset_model = assessment_instance.asset