page_logic.py

from collections.abc import Mapping
from enum import Enum

import pandas as pd
import streamlit as st
from mdata.core import MDConcepts, MachineData, ObservationTypes
from mdata.core.factory import as_base
from mdata.core.protocols import TSSpec
from mdata.visualization import plotting, matplot


class Widgets(Enum):
    Specifications = 'Specifications'
    RawData = 'Raw Data'
    Overview = 'Overview'
    Inspector = 'Inspector'
    Dummy = 'Dummy'


available_widgets = [w for w in Widgets]


def generate_page(md: MachineData, widget):
    md = as_base(md)

    c = st.columns(3)
    c[0].metric('#Observations', md.observation_count)
    c[1].metric('#Event Types', len(md.event_specs))
    c[2].metric('#Measurement Types', len(md.measurement_specs))

    c = st.columns(2)
    cont = c[0].container()
    cont.text(md.index_frame['time'].min())
    cont.caption('First Observation')
    cont = c[1].container()
    cont.text(md.index_frame['time'].max())
    cont.caption('Last Observation')

    if widget == Widgets.Specifications:
        st.markdown('## Events')
        st.table(make_spec_df(md.event_specs))
        st.markdown('## Measurements')
        st.table(make_spec_df(md.measurement_specs))
        #fig = plotting.create_measurement_frequency_plot(md)
        #st.plotly_chart(fig)

    elif widget == Widgets.RawData:

        c1, c2 = st.columns(2)
        with c1:
            st.markdown('## Index')
            st.write(md.index_frame)
        with c2:
            selection = st.selectbox('Observation Spec',
                                     [tsc.timeseries_spec.type + ':' + tsc.timeseries_spec.label for tsc in
                                      md.series_containers])
            if selection is not None:
                label = selection[2:]
                if selection[0] == ObservationTypes.E:
                    st.write(md.get_events(label).df)
                if selection[0] == ObservationTypes.M:
                    st.write(md.get_measurements(label).df)

    elif widget == Widgets.Overview:
        f = plotting.create_overview_plot(md, downsample_to=5_000)
        st.plotly_chart(f, use_container_width=True)

    elif widget == Widgets.Inspector:
        c1, c2, c3 = st.columns(3)
        event_specs_list = list(md.event_series.keys())
        measurement_spec_list = list(md.measurement_series.keys())
        selected_measurement_spec = c1.selectbox('Measurement Spec', measurement_spec_list)
        fs, object_list = [], []
        if selected_measurement_spec is not None:
            fs = list(md.measurement_specs[selected_measurement_spec].features)
            object_list = list(md.measurement_series[selected_measurement_spec].objects)
        measurement_feature_selection = c1.multiselect('Feature Selection', fs)

        event_selection = c2.multiselect('Event Spec Selection', event_specs_list)
        plot_separately = c2.checkbox('plot separately', False)

        object_selection = c3.selectbox('Object Selection', object_list)

        f = plotting.create_timeseries_plot(md, measurement_spec=selected_measurement_spec, obj=object_selection,
                                            events=event_selection, features=measurement_feature_selection,
                                            split_into_subplots=plot_separately)

        st.plotly_chart(f, use_container_width=True)

    elif widget == Widgets.Dummy:
        measurement_spec_list = list(md.measurement_specs)
        selected_measurement_spec = st.selectbox('Measurement Spec', measurement_spec_list)
        if selected_measurement_spec is not None:
            f = matplot.create_basic_stacked_subplots(md, measurement_spec=selected_measurement_spec)
            st.write(f)

    else:
        st.write('Invalid Widget')


def make_spec_df(spec_types: Mapping[str, TSSpec]):
    df = pd.DataFrame.from_dict({label: list(spec.features) for label, spec in spec_types.items()},
                                orient='index').replace(pd.NA, '')
    df.columns = [f'f_{i}' for i in range(1, len(df.columns) + 1)]
    return df