Umbreit/umbreit-py
1
0
Fork 0
generated from dopt-python/py311
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

implemented first reference workflows

Browse Source
This commit is contained in:
Florian Förster
2025-12-03 16:09:09 +01:00
parent 82ad0691c0
commit 4f234670b8
4 changed files with 389 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
data_analysis/02-3_oracle_workflow_test.py
View File

@@ -191,7 +191,9 @@ stmt = sql.select(
else_=db.ext_bedpbed.c.BEDP_MENGE_BEDARF_VM,
).label("BEDP_MENGE_BEDARF_VM"),
db.ext_titel_info.c.MELDENUMMER,
db.ext_titel_info.c.VERLAGSNR,
db.ext_titel_info.c.MENGE_VORMERKER,
db.ext_titel_info.c.MANDFUEHR,
).select_from(db.ext_bedpbed.join(db.ext_titel_info, join_condition, isouter=True))
print(stmt.compile(engine))
@@ -206,18 +208,31 @@ elapsed = t2 - t1
print(f"Query duration: {elapsed:.4f} sec")
print("Number of entries: ", len(df))
print(f"Estimated size in memory: {df.estimated_size(unit='mb')} MB")
# %%
df.head()
# %%
# // NO LIVE DATA NEEDED
# SAVING/LOADING
p_save = Path.cwd() / "raw_data_from_sql_query_20251203-2.arrow"
p_save = Path.cwd() / "raw_data_from_sql_query_20251203-3.arrow"
# df.write_ipc(p_save)
df = pl.read_ipc(p_save)
# %%
print(len(df))
df.head()
# %%
# ** CHECK: differences MANDANT in BEDP and in TINFO
# 4591588: in title database with different MANDANT (are MANDANTFUEHR and BEDP_MAN feasible for matching?)
df.filter(pl.col("BEDP_MAN") != pl.col("MANDFUEHR")).select(pl.col("BEDP_MAN").unique())
# %%
df.group_by("BEDP_MAN").agg(pl.col("MANDFUEHR").unique())
# %%
df.filter(pl.col("MANDFUEHR").is_null()).filter(pl.col("BEDP_MAN") == 1)
# %%
# df.filter(pl.col("BEDP_MAN") != pl.col("MANDFUEHR")).filter(pl.col("BEDP_MAN") == 5)
df.filter(pl.col("BEDP_MAN") == 60).filter(pl.col("MANDFUEHR").is_null())
# %%
# ** CHECK: different MANDANTEN
# check for valid entries for unknown MANDANTEN
@@ -242,7 +257,11 @@ df.head()
# %%
# ** PREFILTER
# always needed, entries filtered out are to be disposed
df = df.filter(pl.col("BEDP_MAN").is_in((1, 90))).filter(pl.col("MELDENUMMER") != 26)
filter_meldenummer_null = pl.col("MELDENUMMER").is_not_null()
filter_mandant = pl.col("MANDFUEHR").is_in((1, 90))
df.filter(filter_meldenummer_null).filter(filter_mandant)
# df = df.filter(pl.col("BEDP_MAN").is_in((1, 90))).filter(pl.col("MELDENUMMER") != 26)
# %%
len(df)
# %%
@@ -283,8 +302,8 @@ df.filter(pl.col("BEDP_MENGE_BEDARF_VM") > pl.col("MENGE_VORMERKER"))
not_in_title_table = df.filter(pl.col("MELDENUMMER").is_null())
EXPORT_FEAT = "BEDP_TITELNR"
to_save = {EXPORT_FEAT: not_in_title_table.select(EXPORT_FEAT).to_series().to_list()}
p_save_not_in_title_table = Path.cwd() / "not_in_title_table_20251203-1.json"
to_save
p_save_not_in_title_table = Path.cwd() / "not_in_title_table_20251203-2.json"
print(to_save)
# with open(p_save_not_in_title_table, "w") as file:
# json.dump(to_save, file, indent=4)
# %%
@@ -295,6 +314,7 @@ print(len(df.filter(pl.col("MELDENUMMER") == 18)))
VM_CRITERION = "BEDP_MENGE_BEDARF_VM"
# TODO exchange to new query focusing on TINFO table
def get_raw_data() -> pl.DataFrame:
join_condition = sql.and_(
db.ext_bedpbed.c.BEDP_TITELNR == db.ext_titel_info.c.TI_NUMMER,
@@ -320,7 +340,7 @@ def get_raw_data() -> pl.DataFrame:
)
def get_empyt_pipeline_result(
def get_empty_pipeline_result(
data: pl.DataFrame,
) -> types.PipelineResult:
schema = db.results_schema_map.copy()
@@ -329,27 +349,6 @@ def get_empyt_pipeline_result(
return types.PipelineResult(results=results, open=data)
def prepare_base_data(
df: pl.DataFrame,
) -> pl.DataFrame:
"""pre-routine to handle non-feasible entries
Parameters
----------
df : pl.DataFrame
raw data collected from database query
Returns
-------
pl.DataFrame
pre-processed data
"""
df = df.with_columns(pl.col("MENGE_VORMERKER").fill_null(0))
df = df.with_columns(pl.col("BEDP_MENGE_BEDARF_VM").fill_null(0))
return df
def _apply_several_filters(
df: pl.DataFrame,
filters: Sequence[pl.Expr],
@@ -409,13 +408,45 @@ def _write_results(
"BEDP_MAN",
"BEDP_MENGE_BEDARF_VM",
"MELDENUMMER",
"VERLAGSNR",
"MENGE_VORMERKER",
"MANDFUEHR",
]
)
return pl.concat([results_table, data])
def workflow_900(
pipe_result: types.PipelineResult,
) -> types.PipelineResult:
"""pre-routine to handle non-feasible entries"""
filter_meldenummer_null = pl.col("MELDENUMMER").is_not_null()
filter_mandant = pl.col("MANDFUEHR").is_in((1, 90))
res = _apply_several_filters(
pipe_res.open,
(
filter_meldenummer_null,
filter_mandant,
),
)
pipe_result.results = _write_results(
pipe_result.results,
data=res.out_,
vorlage=False,
wf_id=900,
freigabe_auto=types.Freigabe.WF_900,
is_out=True,
)
pipe_result.open = res.in_.with_columns(pl.col("MENGE_VORMERKER").fill_null(0))
pipe_result.open = res.in_.with_columns(pl.col("BEDP_MENGE_BEDARF_VM").fill_null(0))
return pipe_result
# main routine
# results for filtered out entries written
def workflow_910(
@@ -476,73 +507,177 @@ def workflow_100_umbreit(
return pipe_result
# Petersen not present in data
def workflow_100_petersen(
pipe_result: types.PipelineResult,
vm_criterion: str,
) -> types.PipelineResult:
# difference WDB and others
# WDB branch
filter_meldenummer = pl.col("MELDENUMMER") == 18
filter_mandant = pl.col("BEDP_MAN") == 90
filter_WDB = pl.col("VERLAGSNR").is_in((76008, 76070))
filter_number_vm = pl.col(vm_criterion) > 0
res = _apply_several_filters(
pipe_result.open,
(
filter_meldenummer,
filter_mandant,
filter_WDB,
filter_number_vm,
),
)
pipe_result.results = _write_results(
results_table=pipe_result.results,
data=res.in_,
vorlage=True,
wf_id=100,
freigabe_auto=types.Freigabe.WF_100,
is_out=False,
)
pipe_result.open = res.out_
# order quantity 0, no further action in other WFs
filter_meldenummer = pl.col("MELDENUMMER") == 18
filter_mandant = pl.col("BEDP_MAN") == 90
filter_WDB = pl.col("VERLAGSNR").is_in((76008, 76070))
filter_number_vm = pl.col(vm_criterion) == 0
res = _apply_several_filters(
pipe_result.open,
(
filter_meldenummer,
filter_mandant,
filter_WDB,
filter_number_vm,
),
)
pipe_result.results = _write_results(
results_table=pipe_result.results,
data=res.in_,
vorlage=False,
wf_id=100,
freigabe_auto=types.Freigabe.WF_100,
is_out=False,
)
pipe_result.open = res.out_
# other branch
filter_meldenummer = pl.col("MELDENUMMER") == 18
filter_mandant = pl.col("BEDP_MAN") == 90
filter_number_vm = pl.col(vm_criterion) > 0
res = _apply_several_filters(
pipe_result.open,
(
filter_meldenummer,
filter_mandant,
filter_number_vm,
),
)
pipe_result.results = _write_results(
results_table=pipe_result.results,
data=res.in_,
vorlage=True,
wf_id=100,
freigabe_auto=types.Freigabe.WF_100,
is_out=False,
)
pipe_result.open = res.out_
return pipe_result
# %%
# SAVING/LOADING
p_save = Path.cwd() / "raw_data_from_sql_query_20251203-1.arrow"
# df.write_ipc(p_save)
p_save = Path.cwd() / "raw_data_from_sql_query_20251203-3.arrow"
df = pl.read_ipc(p_save)
print(f"Number of entries: {len(df)}")
# %%
df.head()
# %%
removed_rows = []
# removed_rows = []
raw_data = df.clone()
print(f"Length raw data: {len(raw_data)}")
filter_mandant = pl.col("BEDP_MAN").is_in((1, 90))
filter_ignore_MNR26 = pl.col("MELDENUMMER") != 26
# raw_data = df.clone()
# print(f"Length raw data: {len(raw_data)}")
# filter_mandant = pl.col("BEDP_MAN").is_in((1, 90))
# filter_ignore_MNR26 = pl.col("MELDENUMMER") != 26
filtered = raw_data.filter(filter_mandant)
filtered_n = raw_data.filter(~filter_mandant)
num_filter = len(filtered)
num_filter_n = len(filtered_n)
removed_rows.append(filtered_n)
print(f"Length filtered: {num_filter}")
print(f"Length filtered out: {num_filter_n}")
print(f"Length all: {num_filter + num_filter_n}")
raw_data = filtered
out = pl.concat(removed_rows)
print(f"Length out: {len(out)}")
# filtered = raw_data.filter(filter_mandant)
# filtered_n = raw_data.filter(~filter_mandant)
# num_filter = len(filtered)
# num_filter_n = len(filtered_n)
# removed_rows.append(filtered_n)
# print(f"Length filtered: {num_filter}")
# print(f"Length filtered out: {num_filter_n}")
# print(f"Length all: {num_filter + num_filter_n}")
# raw_data = filtered
# out = pl.concat(removed_rows)
# print(f"Length out: {len(out)}")
# %%
print("---------------------------------------")
filtered = raw_data.filter(filter_ignore_MNR26)
filtered_n = raw_data.filter(~filter_ignore_MNR26)
num_filter = len(filtered)
num_filter_n = len(filtered_n)
len(filtered_n)
# %%
removed_rows.append(filtered_n)
print(f"Length filtered: {num_filter}")
print(f"Length filtered out: {num_filter_n}")
print(f"Length all: {num_filter + num_filter_n}")
out = pl.concat(removed_rows)
print(f"Length out: {len(out)}")
# # %%
# print("---------------------------------------")
# filtered = raw_data.filter(filter_ignore_MNR26)
# filtered_n = raw_data.filter(~filter_ignore_MNR26)
# num_filter = len(filtered)
# num_filter_n = len(filtered_n)
# len(filtered_n)
# # %%
# removed_rows.append(filtered_n)
# print(f"Length filtered: {num_filter}")
# print(f"Length filtered out: {num_filter_n}")
# print(f"Length all: {num_filter + num_filter_n}")
# out = pl.concat(removed_rows)
# print(f"Length out: {len(out)}")
# %%
raw_data = df.clone()
pipe_res = get_empyt_pipeline_result(raw_data)
pipe_res = get_empty_pipeline_result(raw_data)
pipe_res.results
pipe_res = workflow_900(pipe_res)
print(f"Length of base data: {len(raw_data):>18}")
print(f"Number of entries pipe data: {len(pipe_res):>10}")
print(f"Number of entries result data: {len(pipe_res.results):>8}")
print(f"Number of entries open data: {len(pipe_res.open):>10}")
# %%
pipe_res.results
# raw_data.filter(pl.col("BEDARFNR") == 166982).filter(pl.col("BEDP_SEQUENZ") == 1)
# %%
pipe_res.open.filter(pl.col("BEDP_MENGE_BEDARF_VM") > pl.col("MENGE_VORMERKER"))
# print(f"Base data and pipe result in line: {}")
# %%
pipe_res = workflow_910(pipe_res)
pipe_res
# df_start = prepare_base_data(df_raw)
# df_start
print(f"Length of base data: {len(raw_data):>18}")
print(f"Number of entries pipe data: {len(pipe_res):>10}")
print(f"Number of entries result data: {len(pipe_res.results):>8}")
print(f"Number of entries open data: {len(pipe_res.open):>10}")
# %%
pipe_res.results.select(pl.col("vorlage").value_counts())
# %%
pipe_res = workflow_100_umbreit(pipe_res, VM_CRITERION)
print(f"Length of base data: {len(raw_data):>18}")
print(f"Number of entries pipe data: {len(pipe_res):>10}")
print(f"Number of entries result data: {len(pipe_res.results):>8}")
print(f"Number of entries open data: {len(pipe_res.open):>10}")
# %%
pipe_res = workflow_100_petersen(pipe_res, VM_CRITERION)
print(f"Length of base data: {len(raw_data):>18}")
print(f"Number of entries pipe data: {len(pipe_res):>10}")
print(f"Number of entries result data: {len(pipe_res.results):>8}")
print(f"Number of entries open data: {len(pipe_res.open):>10}")
# %%
pipe_res.results.select(pl.col("vorlage").value_counts())
# %%
pipe_res.results.filter(pl.col("vorlage") == True)
# %%
raw_data.filter(pl.col("BEDARFNR") == 922160).filter(pl.col("BEDP_SEQUENZ") == 3)
# %%
raw_data.head()
# %%
results_init = get_empyt_pipeline_result()
results, filt_out = workflow_100_umbreit(results_init, df_start, VM_CRITERION)
# df is where results are known
# filt_out contains entries for other workflows
# filt_out at this point represents all entries which are to be analysed in other workflows
# %%
results
# %%
filt_out