Version 1.1.1

Robustness overhaul

Author: johnmartins

.github/workflows/python-package.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v4

README.md

@@ -112,7 +112,7 @@ then it is possible to instantiate the DSMs with this as a keyword attribute:
 ```
 dsm = DSM(matrix=data, columns=column_names, instigator="row")
 ```
-The default is `instigator='column`.
+The default is `instigator='column'`.
 This can also be utilized when parsing a CSV-file, like this:
 
 ```

cpm/models.py

@@ -1,4 +1,5 @@
 from typing import Optional
+import numbers
 
 
 class DSM:
@@ -16,14 +17,41 @@ def __init__(self, matrix: list[list[Optional[float]]], columns: list[str], inst
         :param instigator: Can either be **column** or **row**. Determines directionality of interactions in DSM.
         By default, propagation travels from column to row
         """
-        self.matrix = matrix
+        self.matrix = DSM.clean_matrix(matrix)
         self.columns = columns
         self.node_network: dict[int, 'GraphNode'] = self.build_node_network(instigator)
         self.instigator = instigator
 
         if instigator not in ['row', 'column']:
             raise ValueError('instigator argument needs to be either "row" or "column".')
 
+        self.validate_matrix()
+
+    @staticmethod
+    def clean_matrix(matrix) -> list[list[float]]:
+        cleaned_matrix = []
+        for i, row in enumerate(matrix):
+            cleaned_matrix.append([])
+            for j, val in enumerate(row):
+                if val is None:
+                    val = 0
+                try:
+                    cleaned_value = float(val)
+                except ValueError:
+                    cleaned_value = 0
+
+                cleaned_matrix[i].append(cleaned_value)
+
+        return cleaned_matrix
+
+    def validate_matrix(self):
+        try:
+            assert len(self.matrix) == len(self.columns)
+            for row in self.matrix:
+                assert len(row) == len(self.columns)
+        except AssertionError:
+            raise ValueError('Matrix dimensions are inconsistent with provided columns.')
+
     def __str__(self):
         return f'{self.columns}\n{self.matrix}'
 
@@ -47,8 +75,8 @@ def build_node_network(self, instigator: str) -> dict[int, 'GraphNode']:
                 # Ignore diagonal
                 if i == j:
                     continue
-                # Ignore empty cells
-                if col == "" or col is None:
+                # Ignore empty connections
+                if col == "" or col is None or col == 0:
                     continue
 
                 numerical_value = 0.0
@@ -96,8 +124,13 @@ def set_level(self):
 
         return level
 
-    def get_probability(self):
+    def get_probability(self, stack=0):
 
+        # If this node is the single node in the chain, then the probability is none.
+        if len(self.next) == 0 and stack == 0:
+            return 0
+
+        # If final node in chain, set probability to 1 to complete the calculation
         if len(self.next) == 0:
             return 1
 
@@ -107,7 +140,7 @@ def get_probability(self):
             # Likelihood of propagating to this node
             from_this = self.node.neighbours[next_index]
             # Likelihood of that node being propagated to:
-            to_next = self.next[next_index].get_probability()
+            to_next = self.next[next_index].get_probability(stack=stack+1)
             prob = prob * (1 - from_this * to_next)
 
         return 1 - prob
@@ -161,6 +194,9 @@ def __init__(self, start_index: int, target_index: int, dsm_impact: DSM, dsm_lik
             start_index = target_index
             target_index = temp
 
+        if len(dsm_impact.matrix) != len(dsm_likelihood.matrix):
+            raise ValueError('Impact and Likelihood matrices need to have the same dimensions.')
+
         self.dsm_impact: DSM = dsm_impact
         self.dsm_likelihood: DSM = dsm_likelihood
         self.start_index: int = start_index

setup.py

@@ -1,7 +1,7 @@
 from setuptools import setup, find_packages
 
 # python setup.py bdist_wheel sdist
-VERSION = '1.1.0'
+VERSION = '1.1.1'
 DESCRIPTION = 'Tool for calculating risk of change propagation in a system.'
 
 with open("README.md", "r") as fh:

tests/test_input_validation.py

@@ -0,0 +1,54 @@
+import pytest
+from cpm.models import ChangePropagationTree, DSM
+from cpm.parse import parse_csv
+from cpm.utils import calculate_risk_matrix
+
+
+def test_throws_if_dsm_instigator_mismatch_1():
+    dsm_p = parse_csv('./tests/test-assets/dsm-cpx-probs.csv', instigator="row")
+    dsm_i = parse_csv('./tests/test-assets/dsm-cpx-imps.csv')
+
+    with pytest.raises(ValueError):
+        calculate_risk_matrix(dsm_i, dsm_p, search_depth=4)
+
+
+def test_throws_if_dsm_instigator_mismatch_2():
+    dsm_p = parse_csv('./tests/test-assets/dsm-cpx-probs.csv', instigator="row")
+    dsm_i = parse_csv('./tests/test-assets/dsm-cpx-imps.csv')
+
+    with pytest.raises(ValueError):
+        ChangePropagationTree(0, 4, dsm_i, dsm_p)
+
+
+def test_throws_if_dsm_incomplete():
+    mtx = [
+        [0.1, 0.2, 0.3],
+        [0.4, 0.5, 0.6],
+        [0.7, 0.8, 0.9],
+    ]
+    cols = ["a", "b", "c", "d"]
+    with pytest.raises(ValueError):
+        dsm = DSM(mtx, cols)
+
+
+def test_throws_if_dsm_size_mismatch():
+    cols_p = ["a", "b", "c"]
+    mtx_p = [
+        [0.1, 0.2, 0.3],
+        [0.4, 0.5, 0.6],
+        [0.7, 0.8, 0.9],
+    ]
+    cols_i = ["a", "b", "c", "d"]
+    mtx_i = [
+        [0.1, 0.2, 0.3, 0.4],
+        [0.4, 0.5, 0.6, 0.7],
+        [0.7, 0.8, 0.9, 0.10],
+        [0.11, 0.12, 0.13, 0.14]
+    ]
+
+    dsm_p = DSM(mtx_p, cols_p)
+    dsm_i = DSM(mtx_i, cols_i)
+
+    with pytest.raises(ValueError):
+        # If DSMs are of different size, then input validation should prevent execution.
+        ChangePropagationTree(0, 2, dsm_i, dsm_p)

tests/test_parser.py

@@ -14,10 +14,10 @@ def test_parse_dsm_header():
 def test_parse_dsm_matrix():
     dsm = parse_csv('./tests/test-assets/dsm-simple-symmetrical.csv')
 
-    should_be = [[None, 0.1, 0.2, 0.3],
-                 [0.1, None, 0.4, 0.5],
-                 [0.2, 0.4, None, 0.6],
-                 [0.3, 0.5, 0.6, None]]
+    should_be = [[0, 0.1, 0.2, 0.3],
+                 [0.1, 0, 0.4, 0.5],
+                 [0.2, 0.4, 0, 0.6],
+                 [0.3, 0.5, 0.6, 0]]
 
     for i, row in enumerate(should_be):
         for j, col in enumerate(row):

tests/test_propagation.py

@@ -1,4 +1,4 @@
-from cpm.models import ChangePropagationTree
+from cpm.models import ChangePropagationTree, DSM
 from cpm.parse import parse_csv
 from cpm.utils import calculate_risk_matrix
 
@@ -112,8 +112,44 @@ def test_probability_calculation():
 
     dsm_r = parse_csv('./tests/test-assets/dsm-cpx-answers-probs.csv')
 
-    for i, col in enumerate(dsm_r.columns):
-        for j, col in enumerate(dsm_r.columns):
-            if dsm_r.matrix[i][j] is None:
+    for i, col_i in enumerate(dsm_r.columns):
+        for j, col_j in enumerate(dsm_r.columns):
+            if i == j:
                 continue
-            assert abs(res_mtx[i][j] - dsm_r.matrix[i][j]) < 0.001
+            assert abs(res_mtx[i][j] - dsm_r.matrix[i][j]) < 0.001, f"Failed for index i={i} (row {col_i}), j={j} (col {col_j})"
+
+
+def test_dsm_input_robustness():
+    instigator = 'column'
+    cols = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']
+    # Purposefully poorly formatted input matrix
+    mtx_i = [['-', '0', 0.3, None, None, None, 0.7, 0],
+             [None, '-', 0, 0.4, 0.5, None, None, None],
+             ["0.1", 0, '-', '', '', '0.6', 0, 0],
+             [0, 0, 0, 'D', 0, 0, None, 0.8],
+             [0, 0, 0, 0, None, None, 0.7, 0],
+             ['0.1', '0.2', 0, 0, None, 'F', 0.7, 0],
+             [0, 0, 0, 0, 0, 0.6, 99, 0],
+             [0, 0, 0.3, 0.4, 0, 0, 0, 'H']]
+
+    dsm_i = DSM(mtx_i, cols, instigator)
+    # Purposefully poorly formatted input matrix
+    mtx_l = [[None, None, 0.1, None, None, None, 0.1, None],
+             [None, 'B', 0, 0.2, "0.2", 0, 0, None],
+             ["0.3", None, 'C', None, None, 0.3, None, 0],
+             [0, 0, 0, "D", 0, 0, 0, 0.4],
+             [0, 0, 0, 0, "E", 0, 0.5, 0],
+             [0.6, 0.6, None, None, None, "F", "0.6", None],
+             [None, None, None, None, None, 0.7, "G", 0],
+             [0, 0, 0.8, 0.8, 0, 0, 0, "H"]]
+
+    dsm_p = DSM(mtx_l, cols, instigator)
+
+    dsm_r = parse_csv('./tests/test-assets/dsm-cpx-answers-risks.csv')
+    res_mtx = calculate_risk_matrix(dsm_i, dsm_p, search_depth=4)
+
+    for i, col_i in enumerate(dsm_r.columns):
+        for j, col_j in enumerate(dsm_r.columns):
+            if i == j:
+                continue
+            assert abs(res_mtx[i][j] - dsm_r.matrix[i][j]) < 0.001, f"Failed for index i={i} (row {col_i}), j={j} (col {col_j})"