ringe should work now
This commit is contained in:
parent
3891c9c0bf
commit
aef138533c
12
.vscode/launch.json
vendored
12
.vscode/launch.json
vendored
@ -4,12 +4,20 @@
|
||||
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
|
||||
"version": "0.2.0",
|
||||
"configurations": [
|
||||
|
||||
{
|
||||
"name": "Python Debugger: Current File",
|
||||
"name": "Rings",
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/__init__.py",
|
||||
"args": ["-i", "data/ringe_short.xlsx", "-o", "ringe_short_out.xlsx", "-t", "ring"],
|
||||
"console": "integratedTerminal"
|
||||
},
|
||||
{
|
||||
"name": "Countings",
|
||||
"type": "debugpy",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/__init__.py",
|
||||
"args": ["-i", "data/zählungen_short.xlsx", "-o", "zählungen_short_out.xlsx", "-t", "counting"],
|
||||
"console": "integratedTerminal"
|
||||
}
|
||||
]
|
||||
|
||||
70
__init__.py
70
__init__.py
@ -1,17 +1,61 @@
|
||||
|
||||
import argparse
|
||||
from table_handler.serializer import TableSerializer
|
||||
from pprint import pprint as print
|
||||
from logger import logger
|
||||
|
||||
import instructions.countings_formater as countings_formater
|
||||
import instructions.ring_formater as ring_formater
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
parser = argparse.ArgumentParser(description="Serialize Excel tables.")
|
||||
|
||||
parser.add_argument(
|
||||
"-i", "--input",
|
||||
required=True,
|
||||
help="Input Excel file"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"-o", "--output",
|
||||
required=True,
|
||||
help="Output Excel file"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"-t", "--type",
|
||||
required=True,
|
||||
choices=["ring", "counting"],
|
||||
help="Type of table to process"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Select formatter based on type
|
||||
if args.type == "ring":
|
||||
formatter = ring_formater
|
||||
row_instructions = ring_formater.ROW_INSTRUCTIONS
|
||||
else:
|
||||
formatter = countings_formater
|
||||
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
|
||||
)
|
||||
|
||||
logger.info("Starting transformation")
|
||||
serializer.run(
|
||||
formatter.COL_INSTRUCTIONS,
|
||||
row_instructions
|
||||
)
|
||||
|
||||
logger.info("Writing transformations")
|
||||
serializer.write_output(args.output)
|
||||
|
||||
logger.info("Done")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
import instructions.countings_formater as countings_formater
|
||||
|
||||
file = "data/zählungen_short.xlsx"
|
||||
serializer = TableSerializer(file, countings_formater.EXPECTED_INPUT_TABLE_SCHEMA, countings_formater.EXPECTED_OUTPUT_TABLE_SCHEMA)
|
||||
serializer.run(
|
||||
countings_formater.COL_INSTRUCTIONS,
|
||||
countings_formater.ROW_INSTRUCTION
|
||||
)
|
||||
print(serializer.serialized_output)
|
||||
serializer.write_output("zählungen_test_out.xlsx")
|
||||
main()
|
||||
BIN
__pycache__/logger.cpython-312.pyc
Normal file
BIN
__pycache__/logger.cpython-312.pyc
Normal file
Binary file not shown.
BIN
data/ringe_short.xlsx
Normal file
BIN
data/ringe_short.xlsx
Normal file
Binary file not shown.
Binary file not shown.
BIN
instructions/__pycache__/ring_formater.cpython-312.pyc
Normal file
BIN
instructions/__pycache__/ring_formater.cpython-312.pyc
Normal file
Binary file not shown.
@ -1,9 +1,9 @@
|
||||
from mappings import *
|
||||
from .mappings import *
|
||||
|
||||
EXPECTED_INPUT_TABLE_SCHEMA = [
|
||||
"Ringnummer",
|
||||
"Art",
|
||||
"Geschlecht"
|
||||
"Geschlecht",
|
||||
"Alter",
|
||||
"Reproduktionsstatus",
|
||||
"Unterarmlänge",
|
||||
@ -20,51 +20,50 @@ EXPECTED_INPUT_TABLE_SCHEMA = [
|
||||
EXPECTED_OUTPUT_TABLE_SCHEMA = EXPECTED_INPUT_TABLE_SCHEMA
|
||||
|
||||
COL_INSTRUCTIONS = {
|
||||
["Geschlecht"]: {
|
||||
"name": "REGEX_MOVE_TO_COL",
|
||||
"command": {
|
||||
"with_input_iteration": flatten_dict(REPRODUCTION), # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
|
||||
("Geschlecht",): [
|
||||
{
|
||||
"command": "FIND_AND_REPLACE",
|
||||
"map": REPRODUCTION
|
||||
},
|
||||
{
|
||||
"command": "FIND_AND_REPLACE",
|
||||
"map": AGE
|
||||
},
|
||||
{
|
||||
"command": "REGEX_MOVE_TO_COL",
|
||||
"with_input_iteration": list(REPRODUCTION.keys()), # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 0 %}}),?(?: .*+)*$",
|
||||
"to": "Reproduktionsstatus"
|
||||
}
|
||||
},
|
||||
["Geschlecht"]: {
|
||||
"name": "REGEX_MOVE_TO_COL",
|
||||
"command": {
|
||||
"with_input_iteration": flatten_dict(AGE), # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
|
||||
{
|
||||
"command": "REGEX_MOVE_TO_COL",
|
||||
"with_input_iteration": list(AGE.keys()), # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 0 %}}),?(?: .*+)*$",
|
||||
"to": "Alter"
|
||||
}
|
||||
|
||||
},
|
||||
["Geschlecht"]: {
|
||||
{
|
||||
# Special case ma -> Geschlecht a | Alter a
|
||||
"name": "REGEX_MOVE_TO_COL",
|
||||
"command": {
|
||||
"command": "REGEX_MOVE_TO_COL",
|
||||
"match": r"^(?:[^ ]* +)*m(a),?(?: .*+)*$",
|
||||
"to": "Alter"
|
||||
}
|
||||
},
|
||||
["Ringnummer"]: {
|
||||
"name": "REGEX_EDIT_COL",
|
||||
"command": {
|
||||
}],
|
||||
("Ringnummer",): {
|
||||
"command": "REGEX_EDIT_COL",
|
||||
"matching_group_rearange": True,
|
||||
"match": r"^[^a-zA-Z\d]*([a-zA-Z]*)[^a-zA-Z\d]*(\d+)$",
|
||||
"to": "{{% 1 %}}{{% 2 %}}"
|
||||
}
|
||||
"replace_with": r"{{% 0 %}}{{% 1 %}}"
|
||||
},
|
||||
["Art"]: {
|
||||
"name": "FIND_AND_REPLACE",
|
||||
"command": {
|
||||
("Art",): {
|
||||
"command": "FIND_AND_REPLACE",
|
||||
"map": SPICIES
|
||||
}
|
||||
},
|
||||
["Ort"]: {
|
||||
"name": "REGEX_MOVE_TO_COL",
|
||||
"command": {
|
||||
"with_input_iteration": flatten_dict(CATCHED_BY), # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 1 %}}),?(?: .*+)*$",
|
||||
("Ort",): {
|
||||
"command": "REGEX_MOVE_TO_COL",
|
||||
"with_input_iteration": CATCHED_BY, # Moves matching group {{% 1 %}} to "to"
|
||||
"match": r"^(?:[^ ]* +)*({{% 0 %}}),?(?: .*+)*$",
|
||||
"to": "Fangart"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -72,12 +71,13 @@ COL_INSTRUCTIONS = {
|
||||
ROW_INSTRUCTIONS = [
|
||||
{
|
||||
"command": "ROW_COLOR_TO_BOOL", # Match when found as text color in any col of the row
|
||||
"color": ["#ff0000", "#0000ff"],
|
||||
"fg_colors": [(255,0,0), (0,0,255)],
|
||||
"approximate": True,
|
||||
"column": "Wiederfunde",
|
||||
"value": "TRUE"
|
||||
},
|
||||
{
|
||||
"command": "ROW_COPY_REMAINING"
|
||||
"threshold": "ALL", # ALL, ANY, MAJORITY of row must be collored
|
||||
"target": "Wiederfunde",
|
||||
"value": "TRUE",
|
||||
}
|
||||
#{
|
||||
# "command": "ROW_COPY_REMAINING"
|
||||
#}
|
||||
]
|
||||
3
logger.py
Normal file
3
logger.py
Normal file
@ -0,0 +1,3 @@
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -1,13 +1,20 @@
|
||||
import re
|
||||
from copy import deepcopy
|
||||
|
||||
class FindAndReplace:
|
||||
def __init__(self, map: dict[str, dict[str]]):
|
||||
def __init__(self, map: dict[str, list[str]], case_sensitive: bool = False):
|
||||
self.map = map
|
||||
self.case_sensitive = case_sensitive
|
||||
|
||||
def execute_on_cell(self, entry) -> str:
|
||||
res = deepcopy(entry.value)
|
||||
|
||||
for species, alternatives in self.map.items():
|
||||
for old_version in alternatives:
|
||||
if old_version in entry.value:
|
||||
if self.case_sensitive:
|
||||
res = res.replace(old_version, species)
|
||||
else:
|
||||
pattern = re.compile(re.escape(old_version), re.IGNORECASE)
|
||||
res = pattern.sub(species, res)
|
||||
|
||||
return res
|
||||
@ -1,11 +1,26 @@
|
||||
import re
|
||||
import regex
|
||||
|
||||
class RegexEditCol:
|
||||
def __init__(self, match, replace_with):
|
||||
def __init__(self, match, replace_with, matching_group_rearange: bool = False):
|
||||
self.match = match
|
||||
self.replace_with = replace_with
|
||||
self.matching_group_rearange = matching_group_rearange
|
||||
|
||||
def _reorder(self, value: str):
|
||||
matches = regex.match(self.match, value)
|
||||
res = None
|
||||
if matches:
|
||||
for i, match in enumerate(matches.groups([])):
|
||||
res = (res if res else self.replace_with).replace(f"{{{{% {i} %}}}}", match)
|
||||
|
||||
return res if res else value
|
||||
|
||||
def execute_on_cell(self, entry) -> str:
|
||||
if entry.value is None:
|
||||
return entry.value
|
||||
return re.sub(self.match, self.replace_with, entry.value)
|
||||
value = entry.value
|
||||
if value is None:
|
||||
return value
|
||||
|
||||
if self.matching_group_rearange:
|
||||
return self._reorder(value)
|
||||
|
||||
return regex.sub(self.match, self.replace_with, value)
|
||||
@ -11,7 +11,7 @@ class RegexMoveToCol:
|
||||
[[with_input_iteration]]
|
||||
if isinstance(with_input_iteration, str)
|
||||
else [
|
||||
x if isinstance(x, list) else [x]
|
||||
x if isinstance(x, list) or isinstance(x, tuple) else [x]
|
||||
for x in (with_input_iteration if isinstance(with_input_iteration, (list, tuple)) else [with_input_iteration])
|
||||
]
|
||||
)
|
||||
|
||||
52
table_handler/commands/row_color_to_value.py
Normal file
52
table_handler/commands/row_color_to_value.py
Normal file
@ -0,0 +1,52 @@
|
||||
class RowColorToCalue:
|
||||
def __init__(self, target: str,
|
||||
value: str,
|
||||
fg_colors: dict[tuple[int, int, int]] = [],
|
||||
bg_colors: dict[tuple[int, int, int]] = [],
|
||||
approximate: bool = False,
|
||||
threshold: str = "ANY",
|
||||
):
|
||||
|
||||
if not len(fg_colors) and not len(bg_colors):
|
||||
raise ValueError("fg_colors or bg_colors must be providet")
|
||||
|
||||
self.fg_color = fg_colors if fg_colors else []
|
||||
self.bg_color = bg_colors if bg_colors else []
|
||||
|
||||
self.value = value,
|
||||
self.target = target,
|
||||
self.approximate = approximate,
|
||||
|
||||
if threshold not in ("ALL", "ANY", "MAJORITY"):
|
||||
raise ValueError("threshold must be ALL, ANY or MAJORITY")
|
||||
self.threshold = threshold
|
||||
|
||||
|
||||
def _check_if_collection_matchers(self, collectin: dict[tuple[int, int, int]], color_place: str) -> bool:
|
||||
res_map = []
|
||||
if self.approximate:
|
||||
margin = 10
|
||||
for color in collectin:
|
||||
for allowed_color in self.__dict__[color_place]:
|
||||
res_map.append(all(abs(a - b) <= margin for a, b in zip(color, allowed_color)))
|
||||
else:
|
||||
res_map = list(map(lambda x: x in self.__dict__[color_place]))
|
||||
|
||||
match self.threshold:
|
||||
case "ALL":
|
||||
return all(res_map)
|
||||
case "ANY":
|
||||
return any(res_map)
|
||||
case _:
|
||||
return res_map.count(True) > len(res_map)/2
|
||||
|
||||
|
||||
def execute_on_row(self, row: dict[str, str], source_coll: str|None = None):
|
||||
color_pace_results = []
|
||||
for color_place in ["fg_color", "bg_color"]:
|
||||
color_collection = (map(lambda c: c.__dict__[color_place], row.values()))
|
||||
color_pace_results.append(self._check_if_collection_matchers(color_collection, color_place))
|
||||
|
||||
if all(color_pace_results):
|
||||
return {self.target: self.value}
|
||||
return {}
|
||||
@ -10,9 +10,17 @@ from .commands.regex_edit_col import RegexEditCol
|
||||
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))
|
||||
|
||||
@dataclass
|
||||
class Entry:
|
||||
@ -48,15 +56,15 @@ class TableSerializer:
|
||||
print(f"Skipping unexpected Col {cell_value}")
|
||||
|
||||
if set(map_expected_with_ws.keys()) != set(expected_cols):
|
||||
raise ValueError("Missing Col in input")
|
||||
raise ValueError("Missing Col in input, check your table header")
|
||||
|
||||
for rid, row in enumerate(self.ws.iter_rows(min_row=2, values_only=False), start=2):
|
||||
for expected_col, ws_idx in map_expected_with_ws.items():
|
||||
cell = row[ws_idx]
|
||||
data[expected_col][rid] = Entry(
|
||||
value=str(cell.value) if not cell.value is None else None,
|
||||
bg_color=cell.fill.start_color.rgb if cell.fill and cell.fill.start_color else (0,0,0),
|
||||
fg_color=cell.font.color.rgb if cell.font and cell.font.color else (0,0,0)
|
||||
value=str(cell.value) if not cell.value is None else "",
|
||||
bg_color=hex_to_rgb(cell.fill.start_color.rgb if cell.fill and cell.fill.start_color else "#0"),
|
||||
fg_color=hex_to_rgb(cell.font.color.rgb if cell.font and cell.font.color else "#0")
|
||||
)
|
||||
return data
|
||||
|
||||
@ -87,10 +95,11 @@ class TableSerializer:
|
||||
return True
|
||||
|
||||
def _merge_change_to_cel(self, value, changes):
|
||||
if not value and changes:
|
||||
#if not value and changes:
|
||||
# value = changes
|
||||
#elif changes:
|
||||
# value = f"{value}{self.seperator}{changes}"
|
||||
value = changes
|
||||
elif changes:
|
||||
value = f"{value}{self.seperator}{changes}"
|
||||
return value
|
||||
|
||||
def _extecute_instruct(self, source_cell: str, instruct: dict, row: dict[str, any], working_row: dict[str, any]):
|
||||
@ -104,7 +113,7 @@ class TableSerializer:
|
||||
res_row = RegexMoveToCol(instruct["match"], instruct["to"], instruct.get("with_input_iteration")) \
|
||||
.execute_on_row(row, source_cell)
|
||||
case "REGEX_EDIT_COL":
|
||||
res = RegexEditCol(instruct["match"], instruct["replace_with"])\
|
||||
res = RegexEditCol(instruct["match"], instruct["replace_with"], instruct.get("matching_group_rearange"))\
|
||||
.execute_on_cell(row[source_cell])
|
||||
case "REGEX_GROUP_VALUEBASED_CHOOS":
|
||||
res = RegexGroupValuebasedChoose(instruct["match"], instruct["highest"])\
|
||||
@ -112,6 +121,9 @@ class TableSerializer:
|
||||
case "CONDITIONAL_COPY":
|
||||
res_row = Copy(instruct["source"], instruct["target"], replace=instruct.get("replace"))\
|
||||
.execute_on_row(row)
|
||||
case "ROW_COLOR_TO_BOOL":
|
||||
res_row = RowColorToCalue(instruct["target"], instruct["value"], instruct.get("fg_colors"), instruct.get("bg_colors"), instruct.get("approximate"), instruct.get("threshold"))\
|
||||
.execute_on_row(row)
|
||||
|
||||
# Merge row_changes to working_row
|
||||
for target_cel, change in res_row.items():
|
||||
@ -132,19 +144,22 @@ class TableSerializer:
|
||||
|
||||
def run(self, col_instructions: dict[tuple[str], dict], row_instructions: dict[tuple[str], dict]):
|
||||
self.serialized_output = {}
|
||||
for rid in range(2, self.vertical_len + 2):
|
||||
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} ")
|
||||
working_row = {}
|
||||
source_row = self.get_row_as_dict(rid)
|
||||
for source_cols, instruct in list(col_instructions.items()) + [
|
||||
for source_cols, instructions in list(col_instructions.items()) + [
|
||||
((None,), instr) for instr in row_instructions
|
||||
]:
|
||||
for instruct in instructions if type(instructions) == list else [instructions]:
|
||||
for col in source_cols:
|
||||
if not self._check_conditions(instruct, source_row):
|
||||
continue
|
||||
working_row = self._extecute_instruct(col, instruct, source_row, working_row)
|
||||
if not self.implicit_copy:
|
||||
continue
|
||||
for col in self.expected_output_schema:
|
||||
for col in self. expected_output_schema:
|
||||
if col in working_row:
|
||||
continue
|
||||
working_row[col] = source_row.get(col).value
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user