ag-duerr/prototypes/03_visualise.py

# %%
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()

# %%