Change DSM instigator

Author: johnmartins

cpm/models.py

@@ -8,20 +8,22 @@ class DSM:
     one DSM for the likelihood of propagation, and one DSM for the impact of propagation.
     """
 
-    def __init__(self, matrix: list[list[Optional[float]]], columns: list[str]):
+    def __init__(self, matrix: list[list[Optional[float]]], columns: list[str], instigator='column'):
         """
         Construct a DSM using a matrix and column-header
         :param matrix: List matrix containing floats or empty cells
         :param columns: Matrix header
+        :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.columns = columns
-        self.node_network: dict[int, 'GraphNode'] = self.build_node_network()
+        self.node_network: dict[int, 'GraphNode'] = self.build_node_network(instigator)
 
     def __str__(self):
         return f'{self.columns}\n{self.matrix}'
 
-    def build_node_network(self) -> dict[int, 'GraphNode']:
+    def build_node_network(self, instigator: str) -> dict[int, 'GraphNode']:
         """
         Construct a node network using the DSM.
         This enables path finding to be run on the system.\n
@@ -41,7 +43,12 @@ def build_node_network(self) -> dict[int, 'GraphNode']:
                 if not col:
                     continue
                 # Add interaction to node network
-                network_dict[j].add_neighbour(network_dict[i], float(col))  # Assumes cell only contains a float.
+                if instigator == 'column':
+                    network_dict[j].add_neighbour(network_dict[i], float(col))  # Assumes cell only contains a float.
+                elif instigator == 'row':
+                    network_dict[i].add_neighbour(network_dict[j], float(col))
+                else:
+                    raise TypeError('Instigator needs to either be "column" or "row".')
 
         return network_dict
 

cpm/parse.py

@@ -2,10 +2,18 @@
 from cpm.models import DSM
 
 
-def parse_csv(filepath, delimiter='auto', encoding='utf8'):
+def parse_csv(filepath: str, delimiter: str = 'auto', encoding: str = 'utf-8', instigator: str = 'column'):
+    """
+    Parse CSV to DSM
+    :param filepath: Targeted CSV file
+    :param delimiter: CSV delimiter. Defaults to auto-detection.
+    :param encoding: text-encoding. Defaults to utf-8
+    :param instigator: Determines directionality of DSM. Defaults to columns instigating rows.
+    :return: DSM
+    """
 
     if delimiter == 'auto':
-        with open(filepath, 'r') as file:
+        with open(filepath, 'r', encoding=encoding) as file:
             delimiter = detect_delimiter(file.read())
 
     # Identify number of rows, and separate header row
@@ -37,12 +45,18 @@ def parse_csv(filepath, delimiter='auto', encoding='utf8'):
                     except ValueError:
                         data[i - 1].append(None)
 
-    dsm = DSM(matrix=data, columns=column_names)
+    dsm = DSM(matrix=data, columns=column_names, instigator=instigator)
 
     return dsm
 
 
 def detect_delimiter(text, look_ahead=1000):
+    """
+    Attempts to determine CSV delmiter based on a certain amount of sample characters
+    :param text: text from CSV file
+    :param look_ahead: number of samples from CSV-file used to guess the delimiter
+    :return:
+    """
     symbol_map = dict({
         ",": 0,
         ";": 0,

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_packages
 
-VERSION = '1.0.6'
+VERSION = '1.0.7'
 DESCRIPTION = 'Tool for calculating risk of change propagation in a system.'
 
 with open("README.md", "r") as fh:

tests/test_parser.py

@@ -41,3 +41,20 @@ def test_parse_dsm_network():
     assert len(b_neighbours) == 1
     assert b_neighbours[0] == 3
 
+
+def test_parse_dsm_network_instigator_row():
+    dsm = parse_csv('./tests/test-assets/dsm-network-test.csv', instigator='row')
+    assert dsm.node_network[3].name == 'D'
+    d_neighbours = list(dsm.node_network[3].neighbours.keys())
+    assert len(d_neighbours) == 1   # Should only have one connection
+    assert d_neighbours[0] == 1     # Has neighbour with B
+    b_neighbours = list(dsm.node_network[1].neighbours.keys())
+    assert len(b_neighbours) == 1   # Should only have one
+    assert b_neighbours[0] == 2     # B should be neighbour with C
+    c_neighbours = list(dsm.node_network[2].neighbours.keys())
+    assert len(c_neighbours) == 1
+    assert c_neighbours[0] == 0
+    a_neighbours = list(dsm.node_network[0].neighbours.keys())
+    assert len(a_neighbours) == 1
+    assert a_neighbours[0] == 3
+