pain

__init__.py

@@ -1,5 +1,5 @@
 import argparse
-from table_handler.serializer import TableSerializer
+from table_handler.serializer import Transformer, XslxReader
 from logger import logger 
 
 import instructions.countings_formater as countings_formater
@@ -40,10 +40,11 @@ def main():
         row_instructions = countings_formater.ROW_INSTRUCTION
 
     logger.info("Reading file, building serializer...")
-    serializer = TableSerializer(
-        args.input,
-        formatter.EXPECTED_INPUT_TABLE_SCHEMA,
-        formatter.EXPECTED_OUTPUT_TABLE_SCHEMA
+    data = XslxReader(args.input).serialize_table(formatter.EXPECTED_INPUT_TABLE_SCHEMA)
+    
+    serializer = Transformer(
+        data,
+        formatter.EXPECTED_OUTPUT_TABLE_SCHEMA or formatter.EXPECTED_INPUT_TABLE_SCHEMA,
     )
 
     logger.info("Starting transformation")

logger.py

@@ -1,3 +1,10 @@
 import logging
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
+
+def lock_logger(q):
+    while True:
+        msg = q.get()
+        if msg == "DONE":
+            break
+        logger.info(msg)

table_handler/serializer.py

@@ -3,7 +3,8 @@ from openpyxl import Workbook, load_workbook
 from dataclasses import dataclass
 from pprint import pprint as print
 
-import instructions.mappings as mappings
+from table_handler.utils import hex_to_rgb, delegate_tasks
+from logger import lock_logger
 
 from .commands.find_and_replace import FindAndReplace
 from .commands.regex_edit_col import RegexEditCol
@@ -11,16 +12,10 @@ from .commands.regex_move_to_col import RegexMoveToCol
 from .commands.copy import Copy
 from .commands.regex_group_valuebased_choose import RegexGroupValuebasedChoose
 from .commands.row_color_to_value import RowColorToCalue
-from logger import logger
 
 from .conditions.regex_cell import RegexCellCondition
 
-def hex_to_rgb(hex):
-    if type(hex) != str and str("Values must be of type <class 'str'>"):
-        # chatch wierd openxl things
-        hex = "#0"
-    hex=  "{:06}".format(str(hex).lstrip("#"))
-    return tuple(int(hex[i:i+2],16) for i in (0, 2, 4))
+from multiprocessing import Pool, Queue, Process, Manager, cpu_count 
 
 @dataclass
 class Entry:
@@ -30,22 +25,14 @@ class Entry:
 
 
 
-class TableSerializer:
-    def __init__(self, file: str, expected_input_schema: list[str], expected_output_schema: list[str]):
+class XslxReader:
+    def __init__(self, file):
         self.wb: Workbook = load_workbook(file, read_only=True)
         self.ws = self.wb.active
         if not self.ws:
             raise ValueError("No active worksheet found in the workbook.")
         
-        self.data = self._serialize_table(expected_input_schema)
-        self.serialized_output = {}
-        self.seperator = ", "
-        self.expected_output_schema = expected_output_schema or expected_input_schema
-        self.implicit_copy = True
-        self.output_header = True
-
-
-    def _serialize_table(self, expected_cols: list[str]) -> dict[int, list[Entry]]:
+    def serialize_table(self, expected_cols: list[str]) -> dict[int, list[Entry]]:
         data = {col: {} for col in expected_cols}
         map_expected_with_ws = {}
         for idx, cell in enumerate(self.ws[1]):
@@ -68,8 +55,19 @@ class TableSerializer:
                 )
         return data
 
+class Transformer:
+    def __init__(self, data, expected_output_schema: list[str], cumulated_instructions = {}):
+        self.data = data
+        self.fully_serialized_output = {}
+        self.seperator = ", "
+        self.expected_output_schema = expected_output_schema
+        self.implicit_copy = True
+        self.output_header = True
+        self.cumulated_instructions = cumulated_instructions
+
+    
     def write_output(self, destination_file: str):
-        if not self.serialized_output:
+        if not self.fully_serialized_output:
             raise ValueError("There is nothing to write, serialize first")
         
         # write header
@@ -81,7 +79,7 @@ class TableSerializer:
                 ows.cell(row=1, column=col_i, value=col_name)
 
         # write rows
-        for row_i, (row_id, row_data) in enumerate(self.serialized_output.items(), start=2 if self.output_header else 1):
+        for row_i, (row_id, row_data) in enumerate(self.fully_serialized_output.items(), start=2 if self.output_header else 1):
             for col_i, col_name in enumerate(self.expected_output_schema, start=1):
                 ows.cell(row=row_i, column=col_i, value=row_data.get(col_name))
         
@@ -134,24 +132,15 @@ class TableSerializer:
             working_row[source_cell] = self._merge_change_to_cel(working_row.get(source_cell), res)
         return working_row
 
-
-    @property
-    def vertical_len(self):
-        return max(len(data_row) for data_row in self.data.values())
-
-    def get_row_as_dict(self, rid: int):
-        return {col: rows.get(rid) for col, rows in self.data.items()}
-
-    def run(self, col_instructions: dict[tuple[str], dict], row_instructions: dict[tuple[str], dict]):
-        self.serialized_output = {}
-        amount_working_rows = self.vertical_len +2
-        for rid in range(2, amount_working_rows):
-            logger.info(f"Working on row {rid}/{amount_working_rows} ")
+    def _run_row_resolution(self, range, worker: str = "X", q = None):
+        self.fully_serialized_output = {}
+        end = list(range)[-1]
+        for rid in range:
+            if rid % 100 == 0 and q:
+                q.put(f"Worker {worker}:Working on row {rid}/{end}")
             working_row = {}
             source_row = self.get_row_as_dict(rid)
-            for source_cols, instructions in list(col_instructions.items()) + [
-                    ((None,), instr) for instr in row_instructions
-                ]:
+            for source_cols, instructions in self.cumulated_instructions:
                 for instruct in instructions if type(instructions) == list else [instructions]:
                     for col in source_cols:
                         if not self._check_conditions(instruct, source_row):
@@ -163,4 +152,38 @@ class TableSerializer:
                 if col in working_row:
                     continue
                 working_row[col] = source_row.get(col).value
-            self.serialized_output[rid] = working_row
+            self.fully_serialized_output[rid] = working_row
+        return self.fully_serialized_output
+
+    @property
+    def vertical_len(self):
+        return max(len(data_row) for data_row in self.data.values())
+
+    def get_row_as_dict(self, rid: int):
+        return {col: rows.get(rid) for col, rows in self.data.items()}
+
+    def run(self, col_instructions: dict[tuple[str], dict], row_instructions: dict[tuple[str], dict]):
+        self.cumulated_instructions = list(col_instructions.items()) + [
+                    ((None,), instr) for instr in row_instructions
+                ]
+        amount_working_rows = self.vertical_len + 2
+
+        with Manager() as manager:
+            q = manager.Queue()
+
+            chnucs = [[elem, worker_name, q] for worker_name, elem in enumerate(
+                delegate_tasks(range(2, amount_working_rows), cpu_count(), 1000)
+            )]
+
+            log_proc = Process(target=lock_logger, args=(q,))
+            log_proc.start()
+
+            with Pool(len(chnucs)) as p:
+                result = p.starmap(
+                    Transformer(self.data, self.expected_output_schema, self.cumulated_instructions)._run_row_resolution,
+                    chnucs
+                )
+            q.put("DONE")
+            log_proc.join(1)
+        self.fully_serialized_output = {k: v for d in result for k, v in d.items()}
+        return

table_handler/utils.py

@@ -0,0 +1,20 @@
+def hex_to_rgb(hex):
+    if type(hex) != str and str("Values must be of type <class 'str'>"):
+        # chatch wierd openxl things
+        hex = "#0"
+    hex=  "{:06}".format(str(hex).lstrip("#"))
+    return tuple(int(hex[i:i+2],16) for i in (0, 2, 4))
+
+def delegate_tasks(tasks, threats, min_tasks_per_threat):
+    n = len(tasks)
+
+    # determine usable cores
+    usable_cores = min(threats, n // min_tasks_per_threat)
+
+    if usable_cores == 0:
+        usable_cores = 1
+
+    # split tasks evenly
+    chunk_size = (n + usable_cores - 1) // usable_cores
+    chunks = [tasks[i:i + chunk_size] for i in range(0, n, chunk_size)]
+    return chunks