13242820435

Committed 10 Feb 2025 02:12PM UTC coverage: 91.515% (-0.05%) from 91.566%

Build # 13242820435

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #13 from SamhammerAG/KIT-4026

KIT-4026 update

Run Details

151 of 165 relevant lines covered (91.52%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.57

/ai_data_preprocessing_queue/Steps/regex_replacement.py

import re
from io import StringIO
from typing import Any

import pandas


def step(item: Any, item_state: dict[str, Any], global_state: dict[str, Any] | None, preprocessor_data: str) -> Any:
    if preprocessor_data is None or not preprocessor_data:
        return item

    csv = _get_data_from_store_or_reload(global_state, preprocessor_data)

    for _, row in csv.iterrows():
        pattern = re.compile(row[0])
        item = pattern.sub(" " + row[1] + " ", item)

    return item


def _get_data_from_store_or_reload(global_state: dict[str, Any] | None, preprocessor_data: str) -> pandas.DataFrame:
    if global_state is None:
        return _prepare_pre_processor_data(preprocessor_data)

    dict_identifier = "regexReplacementpreprocessor_data"
    if dict_identifier in global_state:
        return global_state[dict_identifier]

    prepared_data = _prepare_pre_processor_data(preprocessor_data)
    global_state[dict_identifier] = prepared_data
    return prepared_data


def _prepare_pre_processor_data(preprocessor_data: str) -> pandas.DataFrame:
    csv = pandas.read_csv(StringIO(preprocessor_data), header=None, usecols=[0, 1, 2], quotechar='"', encoding="utf8")

    csv[0] = csv[0].str.strip()
    csv[1] = csv[1].str.strip()
    csv[2] = pandas.to_numeric(csv[2])
    csv["sort"] = csv[2]

    # sort
    return csv.sort_values("sort", inplace=False).drop("sort", axis=1)

1	import re	1✔
2	from io import StringIO	1✔
3	from typing import Any	1✔
4
5	import pandas	1✔
6
7
8	def step(item: Any, item_state: dict[str, Any], global_state: dict[str, Any] \| None, preprocessor_data: str) -> Any:	1✔
9	if preprocessor_data is None or not preprocessor_data:	1✔
10	return item	1✔
11
12	csv = _get_data_from_store_or_reload(global_state, preprocessor_data)	1✔
13
14	for _, row in csv.iterrows():	1✔
15	pattern = re.compile(row[0])	1✔
16	item = pattern.sub(" " + row[1] + " ", item)	1✔
17
18	return item	1✔
19
20
21	def _get_data_from_store_or_reload(global_state: dict[str, Any] \| None, preprocessor_data: str) -> pandas.DataFrame:	1✔
22	if global_state is None:	1✔
23	return _prepare_pre_processor_data(preprocessor_data)	1✔
24
25	dict_identifier = "regexReplacementpreprocessor_data"	×
26	if dict_identifier in global_state:	×
27	return global_state[dict_identifier]	×
28
29	prepared_data = _prepare_pre_processor_data(preprocessor_data)	×
30	global_state[dict_identifier] = prepared_data	×
31	return prepared_data	×
32
33
34	def _prepare_pre_processor_data(preprocessor_data: str) -> pandas.DataFrame:	1✔
35	csv = pandas.read_csv(StringIO(preprocessor_data), header=None, usecols=[0, 1, 2], quotechar='"', encoding="utf8")	1✔
36
37	csv[0] = csv[0].str.strip()	1✔
38	csv[1] = csv[1].str.strip()	1✔
39	csv[2] = pandas.to_numeric(csv[2])	1✔
40	csv["sort"] = csv[2]	1✔
41
42	# sort
43	return csv.sort_values("sort", inplace=False).drop("sort", axis=1)	1✔

SamhammerAG / ai-data-preprocessing-queue / 13242820435

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous