6941677447

Committed 21 Nov 2023 09:16AM UTC coverage: 91.566%. Remained the same

Build # 6941677447

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #11 from SamhammerAG/KIT-3065

KIT 3065 Update & Maintenance

Run Details

49 of 62 new or added lines in 11 files covered. (79.03%)

1 existing line in 1 file now uncovered.

152 of 166 relevant lines covered (91.57%)

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, Dict, Optional

import pandas


def step(item: Any, item_state: Dict[str, Any], global_state: Optional[Dict[str, Any]], 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: Optional[Dict[str, Any]], 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, Dict, Optional	1✔
4
5	import pandas	1✔
6
7
8	def step(item: Any, item_state: Dict[str, Any], global_state: Optional[Dict[str, Any]], 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: Optional[Dict[str, Any]], preprocessor_data: str) -> pandas.DataFrame:	1✔
22	if global_state is None:	1✔
23	return _prepare_pre_processor_data(preprocessor_data)	1✔
24
NEW 25	dict_identifier = "regexReplacementpreprocessor_data"	×
NEW 26	if dict_identifier in global_state:	×
NEW 27	return global_state[dict_identifier]	×
28
NEW 29	prepared_data = _prepare_pre_processor_data(preprocessor_data)	×
NEW 30	global_state[dict_identifier] = prepared_data	×
NEW 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 / 6941677447

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