Spaces:
Running
Running
| import panel as pn | |
| import holoviews as hv | |
| from stingray import DynamicalPowerspectrum | |
| import warnings | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| from utils.DashboardClasses import ( | |
| MainHeader, | |
| MainArea, | |
| OutputBox, | |
| WarningBox, | |
| HelpBox, | |
| Footer, | |
| WarningHandler, | |
| FloatingPlot, | |
| PlotsContainer, | |
| ) | |
| from utils.app_context import AppContext | |
| import hvplot.pandas | |
| import holoviews.operation.datashader as hd | |
| # Create a warning handler | |
| def create_warning_handler(): | |
| warning_handler = WarningHandler() | |
| warnings.showwarning = warning_handler.warn | |
| return warning_handler | |
| """ Header Section """ | |
| def create_quicklook_dynamicalpowerspectrum_header(context: AppContext): | |
| header_input = pn.widgets.TextInput( | |
| name="Heading", value="QuickLook Dynamical Power Spectrum" | |
| ) | |
| return MainHeader(heading=header_input) | |
| """ Output Box Section """ | |
| def create_loadingdata_output_box(content): | |
| return OutputBox(output_content=content) | |
| """ Warning Box Section """ | |
| def create_loadingdata_warning_box(content): | |
| return WarningBox(warning_content=content) | |
| """ Main Area Section """ | |
| def create_dynamicalpowerspectrum_tab( | |
| context: AppContext, | |
| warning_handler, | |
| ): | |
| event_list_dropdown = pn.widgets.Select( | |
| name="Select Event List(s)", | |
| options={name: i for i, (name, event) in enumerate(context.state.get_event_data())}, | |
| ) | |
| segment_size_input = pn.widgets.FloatInput(name="Segment Size", value=10, step=1) | |
| dt_input = pn.widgets.FloatInput( | |
| name="Select dt", value=1.0, step=0.0001, start=0.0000000001, end=1000.0 | |
| ) | |
| norm_select = pn.widgets.Select( | |
| name="Normalization", options=["leahy", "rms", "frac", "abs"], value="leahy" | |
| ) | |
| rebin_freq_checkbox = pn.widgets.Checkbox(name="Rebin Frequency", value=False) | |
| rebin_time_checkbox = pn.widgets.Checkbox(name="Rebin Time", value=False) | |
| trace_checkbox = pn.widgets.Checkbox(name="Trace Drifting Feature", value=False) | |
| shift_add_checkbox = pn.widgets.Checkbox(name="Apply Shift-and-Add", value=False) | |
| # Inputs for new rebinning parameters | |
| rebin_freq_input = pn.widgets.FloatInput( | |
| name="New Frequency Resolution (Hz)", value=1.0, step=0.1, start=0.1 | |
| ) | |
| rebin_time_input = pn.widgets.FloatInput( | |
| name="New Time Resolution (s)", value=6.0, step=1, start=1.0 | |
| ) | |
| floatpanel_plots_checkbox = pn.widgets.Checkbox( | |
| name="Add Plot to FloatingPanel", value=True | |
| ) | |
| dataframe_checkbox = pn.widgets.Checkbox( | |
| name="Add DataFrame to FloatingPanel", value=False | |
| ) | |
| def create_dynamical_powerspectrum(selected_event_list_index, dt, segment_size, norm): | |
| if selected_event_list_index is not None: | |
| event_list = context.state.get_event_data()[selected_event_list_index][1] | |
| # Use spectrum service to create dynamical power spectrum | |
| result = context.services.spectrum.create_dynamical_power_spectrum( | |
| event_list=event_list, | |
| dt=dt, | |
| segment_size=segment_size, | |
| norm=norm | |
| ) | |
| if result["success"]: | |
| return result["data"] | |
| else: | |
| context.update_container('output_box', | |
| create_loadingdata_output_box(f"Error: {result['message']}") | |
| ) | |
| return None | |
| return None | |
| def create_holoviews_panes(plot): | |
| return pn.pane.HoloViews(plot, width=600, height=600, linked_axes=False) | |
| def create_dataframe(dps): | |
| data = { | |
| "Time": dps.time, | |
| "Frequency": dps.freq, | |
| "Power": dps.dyn_ps.flatten(), | |
| } | |
| return pn.pane.DataFrame(data, width=600, height=400) | |
| def generate_dynamicalpowerspectrum(event=None): | |
| if not context.state.get_event_data(): | |
| context.update_container('output_box', | |
| create_loadingdata_output_box("No loaded event data available.") | |
| ) | |
| return | |
| selected_event_list_index = event_list_dropdown.value | |
| if selected_event_list_index is None: | |
| context.update_container('output_box', | |
| create_loadingdata_output_box("No event list selected.") | |
| ) | |
| return | |
| # Check for conflicting rebin selections | |
| if rebin_freq_checkbox.value and rebin_time_checkbox.value: | |
| context.update_container('warning_box', | |
| create_loadingdata_warning_box( | |
| "Error: You cannot select both 'Rebin Frequency' and 'Rebin Time' simultaneously." | |
| ) | |
| ) | |
| return | |
| dt = dt_input.value | |
| segment_size = segment_size_input.value | |
| norm = norm_select.value | |
| # Use spectrum service to create DynamicalPowerspectrum | |
| event_list = context.state.get_event_data()[selected_event_list_index][1] | |
| result = context.services.spectrum.create_dynamical_power_spectrum( | |
| event_list=event_list, | |
| dt=dt, | |
| segment_size=segment_size, | |
| norm=norm | |
| ) | |
| if not result["success"]: | |
| context.update_container('output_box', | |
| create_loadingdata_output_box(f"Error: {result['message']}") | |
| ) | |
| return | |
| dps = result["data"] | |
| if dps: | |
| # Create Matplotlib plots | |
| fig, ax = plt.subplots() | |
| extent = min(dps.time), max(dps.time), min(dps.freq), max(dps.freq) | |
| im = ax.imshow( | |
| dps.dyn_ps, | |
| aspect="auto", | |
| origin="lower", | |
| interpolation="none", | |
| extent=extent, | |
| ) | |
| ax.set_title("Dynamic Powerspectrum") | |
| ax.set_xlabel("Time (s)") | |
| ax.set_ylabel("Frequency (Hz)") | |
| plt.colorbar(im, ax=ax, label="Power") | |
| base_plot = pn.pane.Matplotlib(fig, width=600, height=400) | |
| # Add the base plot to the floating panel or plot area | |
| if floatpanel_plots_checkbox.value: | |
| context.append_to_container('float_panel', FloatingPlot(title="Base Dynamical Power Spectrum", content=base_plot)) | |
| else: | |
| context.append_to_container('plots', base_plot) | |
| # Rebin Frequency if checkbox is enabled | |
| if rebin_freq_checkbox.value: | |
| new_freq_res = rebin_freq_input.value | |
| dps_rebin_freq = dps.rebin_frequency(df_new=new_freq_res, method="average") | |
| fig, ax = plt.subplots() | |
| extent = min(dps_rebin_freq.time), max(dps_rebin_freq.time), min(dps_rebin_freq.freq), max(dps_rebin_freq.freq) | |
| # extent = [dps_rebin_freq.time[0], dps_rebin_freq.time[-1], dps_rebin_freq.freq[0], dps_rebin_freq.freq[-1]] | |
| im = ax.imshow(dps_rebin_freq.dyn_ps, origin="lower", aspect="auto", interpolation="none", extent=extent) | |
| ax.set_title("Rebinned Frequency") | |
| ax.set_xlabel("Time (s)") | |
| ax.set_ylabel("Frequency (Hz)") | |
| plt.colorbar(im, ax=ax, label="Power") | |
| rebin_freq_plot = pn.pane.Matplotlib(fig, width=600, height=400) | |
| if floatpanel_plots_checkbox.value: | |
| context.append_to_container('float_panel', FloatingPlot(title="Rebinned Frequency", content=rebin_freq_plot)) | |
| else: | |
| context.append_to_container('plots', rebin_freq_plot) | |
| # Rebin Time if checkbox is enabled | |
| if rebin_time_checkbox.value: | |
| new_time_res = rebin_time_input.value | |
| dps_rebin_time = dps.rebin_time(dt_new=new_time_res, method="average") | |
| fig, ax = plt.subplots() | |
| extent = [dps_rebin_time.time[0], dps_rebin_time.time[-1], dps_rebin_time.freq[0], dps_rebin_time.freq[-1]] | |
| im = ax.imshow(dps_rebin_time.dyn_ps, origin="lower", aspect="auto", interpolation="none", extent=extent) | |
| ax.set_title("Rebinned Time") | |
| ax.set_xlabel("Time (s)") | |
| ax.set_ylabel("Frequency (Hz)") | |
| plt.colorbar(im, ax=ax, label="Power") | |
| rebin_time_plot = pn.pane.Matplotlib(fig, width=600, height=400) | |
| if floatpanel_plots_checkbox.value: | |
| context.append_to_container('float_panel', FloatingPlot(title="Rebinned Time", content=rebin_time_plot)) | |
| else: | |
| context.append_to_container('plots', rebin_time_plot) | |
| # Trace Maximum Power if checkbox is enabled | |
| if trace_checkbox.value: | |
| max_pos = dps.trace_maximum() | |
| fig, ax = plt.subplots() | |
| im = ax.imshow(dps.dyn_ps, aspect="auto", origin="lower", interpolation="none", extent=extent, alpha=0.6) | |
| ax.plot(dps.time, dps.freq[max_pos], color="red", lw=2, label="Drifting Feature") | |
| ax.set_title("Dynamic Powerspectrum with Feature Trace") | |
| ax.set_xlabel("Time (s)") | |
| ax.set_ylabel("Frequency (Hz)") | |
| plt.colorbar(im, ax=ax, label="Power") | |
| ax.legend() | |
| trace_plot = pn.pane.Matplotlib(fig, width=600, height=400) | |
| if floatpanel_plots_checkbox.value: | |
| context.append_to_container('float_panel', FloatingPlot(title="Feature Trace Overlay", content=trace_plot)) | |
| else: | |
| context.append_to_container('plots', trace_plot) | |
| generate_dynamicalpowerspectrum_button = pn.widgets.Button( | |
| name="Generate Dynamical Power Spectrum", button_type="primary" | |
| ) | |
| generate_dynamicalpowerspectrum_button.on_click(generate_dynamicalpowerspectrum) | |
| tab_content = pn.Column( | |
| event_list_dropdown, | |
| dt_input, | |
| segment_size_input, | |
| norm_select, | |
| rebin_freq_checkbox, | |
| rebin_freq_input, | |
| rebin_time_checkbox, | |
| rebin_time_input, | |
| trace_checkbox, | |
| floatpanel_plots_checkbox, | |
| dataframe_checkbox, | |
| pn.Row(generate_dynamicalpowerspectrum_button), | |
| ) | |
| return tab_content | |
| def create_quicklook_dynamicalpowerspectrum_main_area(context: AppContext): | |
| warning_handler = create_warning_handler() | |
| tabs_content = { | |
| "Dynamical Power Spectrum": create_dynamicalpowerspectrum_tab( | |
| context=context, | |
| warning_handler=warning_handler, | |
| ), | |
| } | |
| return MainArea(tabs_content=tabs_content) | |
| def create_quicklook_dynamicalpowerspectrum_area(): | |
| return PlotsContainer() | |