# %% import datetime import json import pprint import re from collections import Counter from pathlib import Path import plotly.graph_objects as go import polars as pl import polars.selectors as cs from plotly.subplots import make_subplots # %% p_data_base = (Path.cwd() / "../data/Datenauszug_20260128").resolve() assert p_data_base.exists() # %% concat_data = p_data_base / "all_data.parquet" assert concat_data.exists() df_read = pl.read_parquet(concat_data).rename({"id_process": "id"}) # %% df_read.head() # %% print(df_read.select(pl.col.ts).min()) print(df_read.select(pl.col.ts).max()) # %% # // start df = df_read.clone() # %% # // Step 0: pre-process # drop all columns which are always zero (numeric) or false (boolean) expressions = [] for col, dtype in zip(df.columns, df.dtypes): if dtype == pl.Boolean: # check if all entries are false expressions.append((pl.col(col) == False).all().alias(col)) elif dtype.is_numeric(): # check if all numeric entries are 0 expressions.append((pl.col(col) == 0).all().alias(col)) summary = df.select(expressions) invalid_cols = [col for col in summary.columns if summary[col][0]] print(f"Invalid columns are: {pprint.pformat(invalid_cols)}") df = df.select(pl.exclude(invalid_cols)) df # %% # // Step 1: correct ps sequence # get unique ps sequences ps_seqs = df.group_by("id").agg( pl.col("ps") .sort_by("ts", descending=False) .rle() .struct.field("value") .alias("ps_sequence") ) print(ps_seqs) # count unique ps sequences ps_seqs_unique = ( ps_seqs.select(pl.col("ps_sequence")) .group_by("ps_sequence") .agg(pl.len().alias("count")) .sort("count", descending=True) ) print(ps_seqs_unique) # default order: is the ps order which has the maximum number of occurrences in the dataset default_ps_order = ps_seqs_unique.select( pl.col("ps_sequence").filter(pl.col.count == pl.col.count.max()) )[0, 0] assert isinstance(default_ps_order, pl.Series) print(default_ps_order) # %% # filter all entries which deviate from the default ps sequence expr_valid_seqs = pl.col("ps_sequence") == pl.concat_list(default_ps_order.to_list()) valid_seq_ids = ps_seqs.filter(expr_valid_seqs) invalid_seq_ids = ps_seqs.filter(~expr_valid_seqs) print( f"Num seqs: {ps_seqs.height}, valid: {valid_seq_ids.height}, invalid: {invalid_seq_ids.height}" ) assert ps_seqs.height == (valid_seq_ids.height + invalid_seq_ids.height) # invalid entries are filtered out valid_entries = df_read.join(invalid_seq_ids, on="id", how="anti") print( f"Number of valid processes over all type numbers: {valid_entries.group_by('id').agg().height}" ) # type 2 only valid_entries_type2 = valid_entries.filter(pl.col.type_num == 2) print( f"Number of valid processes for type number 2: {valid_entries_type2.group_by('id').agg().height}" ) # %% all_entries_type2 = df_read.filter(pl.col.type_num == 2) print(all_entries_type2.head()) valid_ids_type2 = valid_entries_type2.select(pl.col.id.unique()).to_series().to_list() # phase starts tmp = valid_entries_type2.sort("id", "ts") phase_starts = tmp.filter((pl.col("ps") != pl.col("ps").shift(1).over("id")).fill_null(True)) print(phase_starts) phase_starts_grouped = phase_starts.group_by("id").agg( pl.col("ps"), pl.col("ts").alias("start_times") ) print(phase_starts_grouped) # %% # get one specific entry TARGET_INDEX_FOR_ID = 0 PROCESS_ID = valid_ids_type2[TARGET_INDEX_FOR_ID] # df = all_entries_type2.clone() df = valid_entries_type2.clone() df = df.filter(pl.col.id == PROCESS_ID) df_phase_starts = phase_starts.filter(pl.col("id") == PROCESS_ID) df_phase_starts # %% fig = make_subplots(specs=[[{"secondary_y": True}]]) fig.add_trace( go.Scatter(x=df["ts"], y=df["DU1210"], name="DU1210 - Druck (mBar)", mode="lines"), secondary_y=False, ) fig.add_trace( go.Scatter(x=df["ts"], y=df["PRV1270"], name="PRV1270 - Druck (mBar)", mode="lines"), secondary_y=False, ) fig.add_trace( go.Scatter(x=df["ts"], y=df["VZ1210"], name="Volumen (ml)", mode="lines"), secondary_y=True, ) # Ventil step plot (discrete) # fig.add_trace( # go.Scatter( # x=df["ts"], # y=df["V1510"], # name="Ventil V1210", # mode="lines", # line_shape="hv", # line=dict(dash="dot"), # ), # secondary_y=True, # ) for row in df_phase_starts.iter_rows(named=True): t_ms = 0 # t_ms = (start - t_min).total_milliseconds() fig.add_vline( x=row["ts"], line=dict(color="red", dash="dash", width=1), annotation_text=str(row["ps"]), annotation_position="top left", annotation=dict( font=dict(size=11, color="red"), bgcolor="white", # no overlap with curves bordercolor="red", ), ) fig.update_layout(title="Zyklus Test — Rohsignal", hovermode="x unified") fig.show() # %% # multiples FEATURE = "DU1210" BREAK_EARLY = True BREAK_IDX = 20 fig = go.Figure() df = valid_entries_type2.clone() for idx, (process_id, group) in enumerate(df.group_by("id")): # Zeit relativ zum Phasenstart normieren, sonst keine Überlagerung möglich t_relative = (group["ts"] - group["ts"].min()).dt.total_milliseconds() print(t_relative) fig.add_trace( go.Scatter( x=t_relative, y=group[FEATURE], opacity=0.3, line=dict(color="steelblue"), showlegend=False, ) ) if BREAK_EARLY and idx == BREAK_IDX: break fig.show() # %%