"""
"Widgets: HTC Visualizer" Module
================================
A core module running interactive widgets in notebook
"""
import difflib
import ipywidgets as widgets
import pandas as pd
import seaborn as sn
from IPython.display import display
import matplotlib.pyplot as plt
from panpython._libs.widgets import checkbox_gen, range_slider_gen
from panpython._libs.widgets import get_table_filtered_by_range_slider
from typing import List, Dict
# def checkbox_creator(properties, inactive_by_default):
# """
# Return checkboxes for visualization
#
# Parameters
# ----------
# properties: list
# A list of properties
# inactive_by_default: list
# A list of inactive properties by default.
#
# Returns
# -------
# Dict[str, widgets.Checkbox]:
# A dictionary of properties checkbox instances
# """
# return checkbox_gen(properties, inactive_by_default)
# def slider_creator(properties, range_description=None, table=None,
# slider_width='80%', text_width='30%', step_num=100):
# """
# Return slider for visualization
#
# Parameters
# ----------
# properties: list
# A list of properties
# range_description: Dict[list]
# A dictionary define the min and max value of a property
# table: pd.DataFrame
# A table used to decide the range of a slide
# slider_width: str
# width of a slider to be displayed
# text_width: str
# width of text description of a slider
# step_num: int
# number of steps of the slider
#
# Returns
# -------
# Dict[str, widgets.FloatRangeSlider]:
# A dictionary of properties FloatRangeSlider instances
# """
# return range_slider_gen(properties, range_description, table,
# slider_width, text_width, step_num)
# def update_table_using_range_slider(table, range_sliders, units=None,
# htc_condition_table=pd.DataFrame(),
# htc_condition_units=None):
# """
# Return updated tables filtered by the ranges from property sliders
#
# Parameters
# ----------
# table: pd.DataFrame
# Original table without units
# range_sliders: List[widgets.FloatRangeSlider]
# A list containing the range slider
# units: dict
# Units of table columns
# htc_condition_table: pd.DataFrame
# table of htc conditions with task_id (and an optional source_id)
# htc_condition_units: dict
# Units of htc condition table columns
#
# Returns
# -------
# pd.DataFrame:
# Updated table with units
# pd.DataFrame:
# Updated table without units
# pd.DataFrame:
# Updated htc condition table with units
# pd.DataFrame:
# Updated htc condition table without units
#
# """
# return get_table_filtered_by_range_slider(table, range_sliders, units,
# htc_condition_table, htc_condition_units)
[docs]class PrecipitateHtcViz:
"""A visualization class of Precipitation HTC results
Parameters
----------
cond_table: pd.DataFrame
the table containing HTC conditions with task_id (and source_id)
cond_units: Dict[str, str]
the units of columns of the condition table
prop_table: pd.DataFrame
The mega table to be analyzed
prop_units: Dict[str, str]
The units of columns of the mega table
descriptions: List[str]
The properties to be analyzed
Note
----
Please see **"Solution Examples\\panhtc_precipitation_post_process\\panhtc_precipitation_post_process.ipynb"** for a detailed example.
"""
def __init__(self, cond_table, cond_units,
prop_table, prop_units,
descriptions):
self.cond_table = cond_table
self.cond_units = cond_units
self.prop_table = prop_table
self.prop_units = prop_units
self.descriptions = descriptions
state_space = []
columns_cond_table = list(cond_table.columns)
for c in columns_cond_table:
if c == 'task_id':
continue
state_space.append(c)
self.HTC_viz = PointHtcViz(state_space=state_space, descriptions=descriptions, table=prop_table,
units=prop_units)
self.HTC_viz.append_cond(cond_table=cond_table, cond_units=cond_units)
[docs] def run(self):
"""Launch the visualization widget
Parameters
----------
Returns
-------
"""
self.HTC_viz.run()
[docs]class PointHtcViz:
"""A visualization class of point HTC results
Parameters
----------
state_space: List[str]
The state space to plot box
descriptions: List[str]
The properties to be analyzed
table: pd.DataFrame
The mega table to be analyzed
units: Dict[str, str]
The units of columns of the mega table
cal_type: str, default='point'
type of HTC calculation
Examples
--------
>>> %matplotlib inline
>>> import ipywidgets as widgets
>>> from ipywidgets import fixed, interactive, interact, interact_manual, HBox, VBox
>>> import pandas as pd
>>>
>>> from panpython.sdk.stat.data_cleanup_for_htc_point import HtcPointInterpreter
>>> from panpython.sdk.widgets.htc_visualizer import PointHtcViz
>>>
>>> # -- get table (Dataframe) and units (list) from merged csv file
>>> merged_file = './output/merged.csv'
>>> m_htc_point_interpreter = HtcPointInterpreter(merged_file)
>>> table = pd.DataFrame()
>>> units = []
>>> try:
>>> table, units = m_htc_point_interpreter.get_table_and_units()
>>> x_columns = table.filter(regex=r"x\(*").columns
>>> vf_columns = table.filter(regex=r"f\(@*").columns
>>> print(merged_file, 'is loaded')
>>> except FileNotFoundError as e:
>>> print(e)
>>>
>>> # visualize
>>> if (not table.empty) and units:
>>> m_htc_viz = PointHtcViz(['T'] + list(x_columns), vf_columns, table=table, units=units)
>>> m_htc_viz.run()
>>> else:
>>> print('table is not loaded.')
"""
def __init__(self, state_space, descriptions, table, units, cal_type='point'):
"""Initialize the visualization widget
Parameters
----------
state_space: List[str]
The state space to plot box
descriptions: List[str]
The properties to be analyzed
table: pd.DataFrame
The mega table to be analyzed
units: Dict[str, str]
The units of columns of the mega table
cal_type: str, optional
type of HTC calculation
Returns
-------
"""
self.state_space = state_space
self.descriptions = descriptions
self.table = table
self.units = units
self.cal_type = cal_type
self.cond_table = pd.DataFrame()
self.cond_units = pd.DataFrame()
self.cond_appended = False
# -- state space checkbox -- #
# -- by default, T and time is not selected -- #
inactive_state_by_default = ['T', 'time']
self.state_space_dict = checkbox_gen(state_space, inactive_state_by_default)
self.state_options = [self.state_space_dict[state] for state in state_space]
# -- property search box -- #
self.search_prop_widget = widgets.Text(placeholder='property name', description='Search:')
# -- property option checkbox -- #
# -- by default, nothing is selected -- #
self.prop_options_dict = checkbox_gen(descriptions, inactive_by_default=descriptions)
self.prop_options = [self.prop_options_dict[description] for description in descriptions]
# -- property range slider -- #
self.prop_ranges_dict = range_slider_gen(descriptions, table=table)
# by default, no slider is displayed
self.prop_ranges = []
# -- Construct state and property selectors -- #
self.options_widget = widgets.HBox(
[widgets.VBox([widgets.HTML(value="<b style=\"color:blue\">Select state</b>")] + self.state_options,
layout=widgets.Layout(width='40%', overflow='scroll')),
widgets.VBox(
[widgets.HTML(value="<b style=\"color:blue\">Select properties</b>"),
self.search_prop_widget] + self.prop_options,
layout=widgets.Layout(width='60%', overflow='scroll'))],
layout=widgets.Layout(width='100%')
)
# -- Construct range slider -- #
self.prop_ranges_widget = widgets.VBox(self.prop_ranges, layout={'overflow': 'scroll'})
# -- Construct the bottom to trigger displaying plot and table -- #
self.button_box_plot = widgets.Button(description="Display/Refresh")
# -- Construct the table display -- #
self.out_df = widgets.Output()
# -- Construct the plot display -- #
self.out_box_plot = widgets.Output()
self.out_corr = widgets.Output()
self.prop_corr = widgets.Output()
sub_plots = {'State Dist.': self.out_box_plot,
'State/Prop. Corr.': self.out_corr,
'Prop. Corr.': self.prop_corr}
self.subplot_tabs = widgets.Tab(children=[p for key, p in sub_plots.items()])
cnt = 0
for key, p in sub_plots.items():
self.subplot_tabs.set_title(cnt, key)
cnt += 1
# -- Combine all elements -- #
# ----------------------------------------- #
# state space selector | property selector #
# ----------------------------------------- #
# property range slider #
# ----------------------------------------- #
# sub-plots #
# ----------------------------------------- #
# table display #
# ----------------------------------------- #
self.HTC_viz = widgets.VBox([self.options_widget, self.prop_ranges_widget,
self.button_box_plot, self.subplot_tabs, self.out_df])
[docs] def append_cond(self, cond_table, cond_units):
""" Add htc condition table with htc task_id (and source_id) and its units to visualization
Parameters
----------
cond_table: pd.DataFrame
the table containing HTC conditions with task_id (and source_id)
cond_units: Dict[str, str]
the units of columns of the condition table
Returns
-------
"""
self.cond_appended = True
self.cond_table = cond_table
self.cond_units = cond_units
[docs] def run(self):
"""Launch the visualization widget
Parameters
----------
Returns
-------
"""
self.search_prop_widget.observe(self._on_text_change, names='value')
for key, op in self.prop_options_dict.items():
op.observe(self._on_options_change, names='value')
self.button_box_plot.on_click(self._on_click_box_plot_button)
display(self.HTC_viz)
# Wire the search field to the checkboxes
def _on_text_change(self, change):
search_input = change['new']
if search_input == '':
# Reset search field
new_options = [self.prop_options_dict[description] for description in self.descriptions]
else:
# Filter by search field using difflib.
close_matches = difflib.get_close_matches(search_input, self.descriptions, cutoff=0.0)
new_options = [self.prop_options_dict[description] for description in close_matches]
self.options_widget.children = new_options
# Wire the options to the range
def _on_options_change(self, change):
check_input = change['new']
if change['new'] != change['old']:
# print(change['owner'].description, 'is changed')
if not change['new']:
new_ranges = [w for w in self.prop_ranges_widget.children
if w.description != change['owner'].description]
else:
new_ranges = [w for w in self.prop_ranges_widget.children]
new_ranges.append(self.prop_ranges_dict[change['owner'].description])
self.prop_ranges_widget.children = new_ranges
# Wire button click to box plot
def _on_click_box_plot_button(self, b):
if len(self.prop_ranges_widget.children) == 0:
with self.out_box_plot:
self.out_box_plot.clear_output()
print('Please select at least one property to visualize!!!')
return
df_copy, df_copy_no_units, cond_subset, cond_subset_no_units \
= get_table_filtered_by_range_slider(self.table, self.prop_ranges_widget.children,
self.units, self.cond_table, self.cond_units)
combined_no_units = pd.DataFrame()
if self.cond_appended:
combined_no_units = pd.merge(df_copy_no_units, cond_subset_no_units, on='task_id', how='left')
box_index = []
for index in self.state_space:
if self.state_space_dict[index].value:
box_index.append(index)
prop_index = []
for index, w in self.prop_options_dict.items():
if w.value:
prop_index.append(index)
with self.out_box_plot:
self.out_box_plot.clear_output()
if not box_index:
print('Please select at least one state (for example, composition or temperature) to plot!!!')
return
fig1, axes1 = plt.subplots()
fig1.suptitle('Composition distribution', fontsize=14)
if not self.cond_appended:
df_copy_no_units.boxplot(box_index, showfliers=False)
else:
cond_subset_no_units.boxplot(box_index, showfliers=False)
fig1.savefig('distribution.svg', format='svg', dpi=1200, bbox_inches="tight")
plt.show()
with self.out_corr:
self.out_corr.clear_output()
fig2, axes2 = plt.subplots()
fig2.suptitle('State/Prop. Correlation', fontsize=14)
corrMatrix = pd.DataFrame()
if self.cond_appended:
corrMatrix = combined_no_units.corr()
corrMatrix = corrMatrix[prop_index]
corrMatrix = corrMatrix.loc[box_index]
else:
corrMatrix = df_copy_no_units.corr()
corrMatrix = corrMatrix[prop_index]
corrMatrix = corrMatrix.loc[box_index]
sn.heatmap(corrMatrix, cmap='jet', linewidths=.5, annot=True)
fig2.savefig('corrMatrix_state_prop.svg', format='svg', dpi=1200, bbox_inches="tight")
plt.show()
with self.prop_corr:
self.prop_corr.clear_output()
fig3, axes3 = plt.subplots()
fig3.suptitle('Prop. Correlation', fontsize=14)
corrMatrix = df_copy_no_units.corr()
corrMatrix = corrMatrix[prop_index]
corrMatrix = corrMatrix.loc[prop_index]
sn.heatmap(corrMatrix, cmap='jet', linewidths=.5, annot=True)
fig3.savefig('corrMatrix_prop.svg', format='svg', dpi=1200, bbox_inches="tight")
plt.show()
with self.out_df:
self.out_df.clear_output()
display(df_copy)