import datetime as dt
import holoviews as hv
import hvplot.pandas
import hvplot.xarray
import pandas as pd
import panel as pn
import pyspedas
from pytplot import get_data
="stretch_width")
pn.extension(sizing_modefrom loguru import logger
from utils import *
In [1]:
THEMIS Level 2 FGM file info:
Fluxgate Magnetometer (FGM) data: - The Level 2 file includes FGE (engineering) magnetic field, FGH (high-resolution) magnetic field, FGL (low-resolution) magnetic field, and FGS (spin-resolution) magnetic field. - The FGH, FGL, and FGE data are given in Spinning Spacecraft (SSL), Despun Spacecraft (DSL), Geocentric Solar Ecliptic (GSE) and Geocentric Solar Magnetospheric (GSM) coordinates. The FGS data is given in DSL, GSE, and GSM coordinates. - Units are nanotesla.
In [38]:
def calculate_PVI_xr(vec, tau, resample_frequency=None, interval_of_averaging=None):
"""
This function calculates the Partial Variance of Increments (PVI) series for a given time series.
Parameters:
vec (np.array): The input time series, represented as a numpy array.
tau (int): The time lag, in unit `s`,typically selected to lie in the inertial range of the fluctuations.
resample_frequency (int): The resample frequency, in unit `s`. If None, defaults to tau.
interval_of_averaging (int): The number of samples over which to compute the trailing average. It's often chosen to be comparable to, or greater than, a correlation length (or time) of the signal.
Returns:
PVI_series (np.array): The resulting PVI series.
"""
# Sample the vector at the given time lag (tau)
# vec_sampled = vec.resample(time=tau).mean(dim='time')
if resample_frequency is None:
= tau
resample_frequency
# Interpolate to a regular time grid
# vec_sampled = vec.resample(time=pd.Timedelta(resample_frequency, unit='s')).interpolate('linear')
= vec.resample(
vec_sampled =pd.Timedelta(resample_frequency, unit="s")
time"linear")
).interpolate(
# Calculate the increments in the vector
= int(tau / resample_frequency)
n = vec_sampled.diff(dim="time", n=n, label="lower")
increments # Calculate the magnitudes of these increments
= linalg.norm(increments, dims="v_dim")
mag_increments
# Square the magnitudes of the increments, and compute a moving average over the specified interval
if interval_of_averaging is None:
= np.sqrt(np.mean(np.square(mag_increments)))
normalized_factor = mag_increments / normalized_factor
PVI_series else:
= mag_increments
PVI_series
...# TODO: Implement moving average
if "units" in PVI_series.attrs:
del PVI_series.attrs["units"]
"time"] = PVI_series["time"] + pd.Timedelta(tau / 2, unit="s")
PVI_series[return PVI_series.rename("PVI")
def PVI_map(vec, tau_range, resample_frequency=None):
"""_summary_
Args:
vec (_type_): _description_
"""
if resample_frequency == None:
= xr.concat(
PVI_series for tau in tau_range], dim="tau"
[calculate_PVI_xr(vec, tau)
)else:
= xr.concat(
PVI_series for tau in tau_range],
[calculate_PVI_xr(vec, tau, resample_frequency) ="tau",
dim
)return PVI_series
In [2]:
= "c"
probe = dt.datetime(2019, 1, 7)
start = dt.date.today()
end = "fgs"
datatype
= f"th{probe}_fgs_gsm"
thx_fgs = f"th{probe}_fgl_gsm"
thx_fgl
= start.strftime("%Y-%m-%d")
tstart = (start + dt.timedelta(1)).strftime("%Y-%m-%d")
tstop = [tstart, tstop]
trange
=probe, trange=trange, varnames=[thx_fgs, thx_fgl])
pyspedas.themis.fgm(probe= get_data(thx_fgs, xarray=True)
thx_fgs_data # thx_fgl_data = get_data(thx_fgl, xarray=True)
= pytplot.get_timespan(thx_fgl)
fgl_tstart, fgl_tstop = pd.Timestamp(fgl_tstart, unit="s")
fgl_tstart = pd.Timestamp(fgl_tstop, unit="s") fgl_tstop
22-Jul-23 16:17:39: Downloading remote index: http://themis.ssl.berkeley.edu/data/themis/thc/l2/fgm/2019/
22-Jul-23 16:17:39: File is current: /Users/zijin/data/themis/thc/l2/fgm/2019/thc_l2_fgm_20190107_v01.cdf
In [3]:
# Time index may contain duplicate values
= thx_fgs_data.groupby("time").first()
thx_fgs_data
# Interpolate to a regular time grid
= thx_fgs_data.resample(time=pd.Timedelta(8, unit="s")).interpolate(
thx_fgs_data "linear"
)
In [42]:
= [8, 16, 32, 64, 128, 256]
tau_range = PVI_map(thx_fgs_data, tau_range, tau_range[0]) pvi
In [43]:
= pvi.assign_coords({"tau": tau_range}) pvi
In [44]:
="time", y="tau", logy=True) pvi.hvplot.quadmesh(x
22-Jul-23 16:41:56: /Users/zijin/mambaforge/envs/cool_solar_wind/lib/python3.10/site-packages/bokeh/core/property/bases.py:259: DeprecationWarning: elementwise comparison failed; this will raise an error in the future.
return new == old
In [2]:
= pn.widgets.DateSlider(
date_slider ="Date Slider",
name=start,
start=end,
end=1000
step* 60
* 60
* 24, # (number): The selected step i the slider in milliseconds
)
= pn.pane.PNG(
overview_pane "overview_moms"), name="overview_moms"
themis_image_url(start, probe,
)= pn.pane.GIF(
orbit_pane "orbit_moon"), name="orbit_moon"
themis_image_url(start, probe,
)
def update_pane_image_callback(target, event):
= event.new
date = themis_image_url(date, probe, target.name)
image_url object = image_url
target.
={"value": update_pane_image_callback})
date_slider.link(overview_pane, callbacks={"value": update_pane_image_callback})
date_slider.link(orbit_pane, callbacks
def index_plot(probe, date):
= f"th{probe}_fgs_gsm"
thx_fgm = f"th{probe}_fgl_gsm"
thx_fgl
= date.strftime("%Y-%m-%d")
tstart = (date + dt.timedelta(1)).strftime("%Y-%m-%d")
tstop = [tstart, tstop]
trange
=probe, trange=trange, varnames=[thx_fgm, thx_fgl])
pyspedas.themis.fgm(probe= get_data(thx_fgm, xarray=True)
thx_fgm_data
= pytplot.get_timespan(thx_fgl)
fgl_tstart, fgl_tstop = pd.Timestamp(fgl_tstart, unit="s")
fgl_tstart = pd.Timestamp(fgl_tstop, unit="s")
fgl_tstop
= calculate_PVI_xr(thx_fgm_data, "10s")
pvi = 10 * calculate_mag_change_xr(thx_fgm_data, "10s")
mag_change
= pvi.hvplot(x="time")
pvi_plot = hv.Curve([(fgl_tstart, 5), (fgl_tstop, 5)])
threshold_line = thx_fgm_data.hvplot(x="time", by="v_dim")
thx_fgm_plot
# thx_fgl_data = get_data(thx_fgl, xarray=True)
# thx_fgl_plot = thx_fgl_data.hvplot(x="time", by="v_dim", kind="scatter")
=(0, 10), alpha=0.3)
pvi_plot.opts(ylim=0.3)
threshold_line.opts(alpha
= pvi_plot * mag_change.hvplot(x="time", alpha=0.3) * threshold_line
index_plot
= pn.pane.Markdown()
markdown
def update_widget(event):
= pd.Timestamp(event.new[0])
temp_tstart = pd.Timestamp(event.new[1])
temp_tstop = "fgs"
temp_datatype
if temp_tstart > fgl_tstart and temp_tstop < fgl_tstop:
= "fgl"
temp_datatype
object = f"""
markdown. ```
{{"probe": "{probe}", "tstart": "{temp_tstart.strftime('%Y-%m-%dT%H:%M:%S')}", "tstop": "{temp_tstop.strftime('%Y-%m-%dT%H:%M:%S')}", "datatype": "{temp_datatype}"}},
```
"""
= hv.streams.RangeX(source=index_plot)
rng "x_range")
rng.param.watch(update_widget,
return pn.Column(markdown, pn.Row(index_plot, thx_fgm_plot))
= hvplot.bind(lambda date: index_plot(probe, date), date_slider)
index_pane
= pn.Column(pn.Row(date_slider), index_pane, pn.Row(overview_pane, orbit_pane))
app app
In [34]:
= dt.datetime(2019, 1, 7)
date = date.strftime("%Y-%m-%d")
tstart = (date + dt.timedelta(1)).strftime("%Y-%m-%d")
tstop = [tstart, tstop]
trange =probe, trange=trange) pyspedas.themis.fgm(probe
29-Jun-23 12:28:55: Downloading remote index: http://themis.ssl.berkeley.edu/data/themis/thc/l2/fgm/2019/
29-Jun-23 12:28:55: File is current: /Users/zijin/data/themis/thc/l2/fgm/2019/thc_l2_fgm_20190107_v01.cdf
['thc_fgs_btotal',
'thc_fgs_gse',
'thc_fgs_gsm',
'thc_fgs_dsl',
'thc_fgl_btotal',
'thc_fgl_gse',
'thc_fgl_gsm',
'thc_fgl_dsl',
'thc_fgl_ssl',
'thc_fgh_btotal',
'thc_fgh_gse',
'thc_fgh_gsm',
'thc_fgh_dsl',
'thc_fgh_ssl']
In [3]:
import holoviews as hv
# Widgets
import pandas as pd
import panel as pn
# (index_plot + thx_fgm_plot).cols(1)
= thx_fgm_data
xr = xr.time.data.min()
start = xr.time.data.max()
end
= pd.Timestamp(start).to_pydatetime()
start = pd.Timestamp(end).to_pydatetime()
end
# Create a DateRangeSlider widget
= pn.widgets.DateRangeSlider(
date_range_slider ="Date Range Slider",
name=start,
start=end,
end=(start, end),
valueformat="%H:%M:%S",
)
# Create a DatetimeRangeInput widget
= pn.widgets.DatetimeRangeInput(
datetime_range_input ="Datetime Range Input",
name=(start, end),
value=start,
start=end,
end
)
# Create a Markdown widget to copy
= pn.pane.Markdown()
markdown
def markdown_callback(target, event):
object = f"""
target. ```
event = {{
"probe": "{probe}",
"tstart": "{event.new[0]}",
"tstop": "{event.new[1]}",
"datatype": "{datatype}",
}}
```"""
# Link the values of the DateRangeSlider and DatetimeRangeInput widgets
="value")
date_range_slider.link(datetime_range_input, value={"value": markdown_callback})
date_range_slider.link(markdown, callbacks="value")
datetime_range_input.link(date_range_slider, value
def plot(dt_range):
= str(dt_range[0])
tstart = str(dt_range[1])
tstop return plot_event(probe, tstart, tstop, datatype, output="all")
= pn.bind(plot, date_range_slider)
interactive_plot
pn.Column(
pn.Row(date_range_slider, datetime_range_input),
markdown,apply.opts(xlim=date_range_slider)),
pn.Row(index_plot, thx_fgm_plot.
interactive_plot, )
In [54]:
pd.Timestamp(tstart)
Timestamp('2019-01-01 00:00:00')
In [51]:
="s").strftime("%Y-%m-%dT%H:%M:%S") pd.Timestamp(_tstart, unit
'2019-01-01T00:00:30'
In [18]:
import numpy as np
"2002-06-28") > pd.Timestamp("2002-06-29") pd.Timestamp(
False