Compare Campaign Measurements

Define measurement query and which measurement quantities to plot

Available Measurement Quantities

Ambient, Coolant, I_d, I_q, Motor_speed, Pm, Stator_tooth, Stator_winding, Stator_yoke, Time, Torque, U_d, U_q

mea_result_conditions = {
                "Name": {"$like": "Profile_*"},
                "TestStep.Test.Name": {"$in": ["Campaign_03", "Campaign_04"]},
                "TestStep.Test.StructureLevel.Project.Name": {"$eq": "ElectricMotorTemperature"}
            }

mq_names = ["Motor_speed", "Torque"]

Retrieve content from ASAM ODS ODS service

from odsbox import ConI

with ConI(url="https://docker.peak-solution.de:10032/api", auth=("Demo", "mdm")) as con_i:
    # Query measurements to compare
    measurements = con_i.query_data({
            "MeaResult": mea_result_conditions,
            "$attributes": {
                "Name": 1,
                "Id": 1,
                "TestStep": {
                    "Name": 1,
                    "Test": {
                        "Name": 1,
                        "StructureLevel": {
                            "Name": 1,
                            "Project": {
                                "Name": 1
                                }}}},
            },
        })
    measurement_ids = measurements["MeaResult.Id"].tolist()

    # Query submatrices of measurements containing the specified measurement quantities
    submatrices = con_i.query({
            "AoSubMatrix": {
                "measurement": {"$in": measurement_ids},
                "local_columns": {
                    "name": {
                        "$in": mq_names
                    },
                }
            },
            "$attributes": {
                "id": 1,
                "number_of_rows": 1,
                "measurement": 1,
            },
            "$groupby": {
                "id": 1,
                "measurement": 1,
                "number_of_rows": 1,
            }
        })
    submatrix_ids = submatrices["id"].tolist()
    # Get units and other info of local columns in submatrices
    local_columns = con_i.query(
        {
            "AoLocalColumn": {
                "submatrix": {"$in": submatrix_ids},
                "$or": [
                    {"name": {"$in": mq_names}},
                    {"independent": 1}
                ]
            },
            "$attributes": {
                "id": 1,
                "name": 1,
                "independent": 1,
                "measurement_quantity.unit:OUTER.name": 1,
            }
        })
    local_column_ids = local_columns["id"].tolist()
    # Fetch bulk data for local columns
    local_columns_signals = con_i.bulk.query({"id":{"$in": local_column_ids}})

Prepare collected data for plotting

import pandas as pd

# lookup for units of local columns
local_column_unit_lookup = dict({row["id"]: row["measurement_quantity.unit:OUTER.name"]
                                 for _, row in local_columns.iterrows()})

# generate title for each submatrix
submatrix_title_lookup = {}
for submatrix_id in submatrix_ids:
    measurement_id = submatrices[submatrices['id'] == submatrix_id]['measurement'].values[0]
    measurement_info = measurements[measurements['MeaResult.Id'] == measurement_id]
    if not measurement_info.empty:
        project = measurement_info['Project.Name'].values[0]
        profile = measurement_info['MeaResult.Name'].values[0]
        campaign = measurement_info['Test.Name'].values[0]
        submatrix_title_lookup[submatrix_id] = f"{project} - {campaign} - {profile}"
    else:
        submatrix_title_lookup[submatrix_id] = f"Submatrix {submatrix_id}"

# Create dataframes for each measurement to compare
measurement_data_items = []
for submatrix_id, df in local_columns_signals.groupby("submatrix"):
    # Create dataframe using name as column names and values as column data
    data_dict = {row['name']: row['values'] for _, row in df.iterrows()}
    sub_df = pd.DataFrame(data_dict)
    if not all(col in sub_df.columns for col in mq_names):
        print(f"Skipping submatrix {submatrix_id} as it lacks required columns.")
        continue

    independent_rows = df[df['independent'] == 1]
    independent_name = "Index"
    independent_unit = "-"
    # Set the independent column as the index
    if not independent_rows.empty:
        independent_name = independent_rows['name'].iloc[0]
        sub_df.set_index(independent_name, inplace=True)
        independent_unit = local_column_unit_lookup.get(independent_rows['id'].iloc[0], '-')
    # create axis labels from name and unit
    label_dict = {name: f"{name} [{local_column_unit_lookup.get(id, '-')}]"
                  for name, id in zip(df['name'], df['id'])}
    # information to be plotted
    measurement_data_items.append({
        "title": f"{submatrix_title_lookup.get(submatrix_id, 'Unknown')} \n(Color: {independent_name} [{independent_unit}])",
        "data": sub_df,
        "labels": label_dict })

# Sort measurement_data_items by title (moved outside the loop)
measurement_data_items.sort(key=lambda x: x["title"])

Plot measurements

import matplotlib.pyplot as plt

# Number of submatrices
num_of_plots = len(measurement_data_items)

# Calculate number of rows (3 submatrices per row)
cols = 3
rows = (num_of_plots + cols - 1) // cols  # Ceiling division

fig, axes = plt.subplots(rows, cols, figsize=(15, 5 * rows))
axes = axes.flatten()  # Flatten to 1D array for easy indexing

for i, measurement_data in enumerate(measurement_data_items):
    ax = axes[i]
    measurement_data["data"].plot.scatter(
        x=mq_names[0],
        y=mq_names[1],
        c=measurement_data["data"].index,
        colormap="viridis",
        ax=ax,
        alpha=0.7,
        s=20
    )
    ax.set_xlabel(measurement_data["labels"].get(mq_names[0], mq_names[0]))
    ax.set_ylabel(measurement_data["labels"].get(mq_names[1], mq_names[1]))
    ax.set_title(measurement_data["title"])

# Hide any unused subplots
for j in range(i + 1, len(axes)):
    axes[j].axis('off')

plt.tight_layout()
plt.show()