Added full matrix calculation and directory parsing

Author: johnmartins

cpm/parse.py

@@ -1,5 +1,34 @@
 from cpm.exceptions import *
 from cpm.models import DSM
+from os import listdir
+import re
+
+
+def parse_csv_dir(dir_path: str, pattern: str =  None,  delimiter: str = 'auto',
+                  encoding: str = 'utf-8', instigator: str = 'column') -> list[DSM]:
+    """
+    Parse a directory of CSVs. A pattern for what the filename needs to include can be used
+    as an inclusivity-filter.
+    :param path:
+    :param pattern:
+    :param delimiter:
+    :param encoding:
+    :param instigator:
+    :return:
+    """
+    dsm_array = []
+    p = None
+    if pattern is not None:
+        p = re.compile(pattern, re.DOTALL)
+
+    for filename in listdir(dir_path):
+        if p and p.match(filename) is None:
+            continue
+        filepath = dir_path + '/' + filename
+        print(filepath)
+        dsm_array.append(parse_csv(filepath))
+
+    return dsm_array
 
 
 def parse_csv(filepath: str, delimiter: str = 'auto', encoding: str = 'utf-8', instigator: str = 'column'):

cpm/utils.py

@@ -0,0 +1,27 @@
+from typing import Union
+from cpm.models import ChangePropagationTree, DSM
+
+
+def calculate_risk_matrix(dsm_impact: DSM, dsm_likelihood: DSM, search_depth=4) \
+        -> list[list[Union[float, str]]]:
+    """
+    Run Change Propagation algorithm on entire DSM, and generate a risk matrix.
+    :param dsm_impact:
+    :param dsm_likelihood:
+    :param search_depth:
+    :return:
+    """
+
+    cpm: list[list[Union[float, str]]] = []
+
+    for l_index, lcol in enumerate(dsm_likelihood.columns):
+        cpm.append([])
+
+        for i_index, icol in enumerate(dsm_impact.columns):
+            cpt = ChangePropagationTree(i_index, l_index,
+                                        dsm_impact=dsm_impact,
+                                        dsm_likelihood=dsm_likelihood)
+            cpt.propagate(search_depth=search_depth)
+            cpm[l_index].append(cpt.get_risk())
+
+    return cpm

tests/test_propagation.py

@@ -10,18 +10,18 @@ def test_risk_calculation():
     depth = 4
     res_mtx: list[list[float]] = []
 
-    for i, col in enumerate(dsm_p.columns):
+    for i, _ in enumerate(dsm_p.columns):
         res_mtx.append([])
-        for j, col in enumerate(dsm_p.columns):
+        for j, _ in enumerate(dsm_p.columns):
             cpt = ChangePropagationTree(start_index=j, target_index=i, dsm_impact=dsm_i, dsm_likelihood=dsm_p)
             cpt.propagate(search_depth=depth)
             r = cpt.get_risk()
             res_mtx[i].append(r)
 
     dsm_r = parse_csv('./tests/test-assets/dsm-cpx-answers-risks.csv')
 
-    for i, col in enumerate(dsm_r.columns):
-        for j, col in enumerate(dsm_r.columns):
+    for i, _ in enumerate(dsm_r.columns):
+        for j, _ in enumerate(dsm_r.columns):
             if dsm_r.matrix[i][j] is None:
                 continue
             assert abs(res_mtx[i][j] - dsm_r.matrix[i][j]) < 0.001

tests/test_utils.py

@@ -0,0 +1,26 @@
+
+from cpm.parse import parse_csv
+from cpm.utils import calculate_risk_matrix
+
+
+def test_matrix_risk_calculation():
+    dsm_p = parse_csv('./tests/test-assets/dsm-cpx-probs.csv')
+    dsm_i = parse_csv('./tests/test-assets/dsm-cpx-imps.csv')
+    dsm_answers = parse_csv('./tests/test-assets/dsm-cpx-answers-risks.csv')
+
+    depth = 4
+
+    dsm_risk = calculate_risk_matrix(dsm_i, dsm_p, search_depth=depth)
+
+    assert dsm_risk is not None
+    assert len(dsm_risk) > 0
+
+    for i, _ in enumerate(dsm_risk):
+        for j, _ in enumerate(dsm_risk):
+            if i == j:
+                continue
+
+            print(f'{dsm_answers.matrix[i][j]} {dsm_risk[i][j]}')
+            if dsm_answers.matrix[i][j] in [None, 0] and dsm_risk[i][j] in [None, 0]:
+                continue
+            assert abs(dsm_answers.matrix[i][j] - dsm_risk[i][j]) < 0.001