Dynamic catalogue and plot function
- Dominant language
- Python
- Stars
- 1.1k
- Forks
- 151
- Avg merge
- 8h 24m
- Merged PRs (30d)
- 2
Description
Suppose I have a massive file system whose directories get updated multiple times a day. The files are a proprietary format that I have developed a Panel application that opens and visualizes them. My datasets are opened into an `Xarray.Dataset` and depending on the attributes of the opened dataset the `crs`, `projection`, `global_extent` and `project` are set for a `hvplot.quadmesh`.
I have made everything work individually and even as a Panel app but I am looking to take the `intake.catalog`s I have created and combine it with the plotting I've developed using `hvplot`. Ideally I can pass a Python file with the plotting function and helper methods that would be used for a given dataset.
I have added the plotting code a Panel app I made below in case it adds relevant context to my question.
```python
def get_fsts_subset(model: str, run: str) -> List[str]:
assert model in MODELS_DICT, "Invalid model provided"
fsts_subset = []
for f in MODELS_DICT[model]["fsts"]:
if run == f.split("/")[-1][8:10]:
fsts_subset.append(f)
return fsts_subset
def get_xarray_dataset(fsts_list: List[str], data_var: str) -> Union[xr.Dataset, None]:
try:
temp_ds = fstd2nc.Buffer(
fsts_list,
vars=[data_var],
rpnstd_metadata=True,
opdict=True,
forecast_axis=True,
).to_xarray()
try:
temp_ds['forecast'] = temp_ds.forecast / timedelta64(3600, "s")
except Exception as e:
print(f"Error converting forecast to hours: {e}")
return temp_ds
except Exception as e:
print(f"Error getting xarray dataset: {e}")
return None
def get_crs(ds_in: xr.Dataset) -> List[Union[RotatedPole, Stereographic, PlateCarree, None]]:
assert ds_in is not None, "Invalid input dataset"
crs_out = None
data_vars_list = list(ds_in.data_vars)
if "rotated_pole" in data_vars_list:
crs_out = RotatedPole(
pole_latitude=ds_in.rotated_pole.attrs.get("grid_north_pole_latitude"),
pole_longitude=ds_in.rotated_pole.attrs.get("grid_north_pole_longitude"),
central_rotated_longitude=0.0,
globe=Globe(
ellipse=None,
semimajor_axis=ds_in.rotated_pole.attrs.get("earth_radius"),
semiminor_axis=None,
),
)
if "polar_stereo" in data_vars_list:
crs_out = Stereographic(
central_latitude=ds_in.polar_stereo.attrs.get("latitude_of_projection_origin"),
central_longitude=ds_in.polar_stereo.attrs.get("straight_vertical_longitude_from_pole"),
false_easting=ds_in.polar_stereo.attrs.get("false_easting"),
false_northing=ds_in.polar_stereo.attrs.get("false_northing"),
true_scale_latitude=ds_in.polar_stereo.attrs.get("latitude_of_projection_origin"),
globe=Globe(semimajor_axis=ds_in.polar_stereo.attrs.get("earth_radius")),
)
if "crs_latlon" in data_vars_list:
crs_out = PlateCarree(globe=Globe(semimajor_axis=ds_in.crs_latlon.attrs.get("earth_radius")))
return crs_out
def get_hvplot(model: str, run: str, data_var: str, cmap: str, alpha: float) -> Union[QuadMesh, None]:
fsts_subset = get_fsts_subset(model, run)
ds = get_xarray_dataset(fsts_subset, data_var)
if ds is None:
print("Error: unable to create hvplot due to missing dataset")
return None
crs_plot = get_crs(ds)
project_bool = not isinstance(crs_plot, PlateCarree)
global_extent_bool = isinstance(crs_plot, PlateCarree)
coastline_projected = (coastline * borders * ocean * lakes * rivers).opts(projection=crs_plot)
return coastline_projected * ds[data_var].hvplot.quadmesh(
rasterize=True,
data_aspect=1,
frame_height=550,
cmap=cmap,
crs=crs_plot,
projection=crs_plot,
project=project_bool,
global_extent=global_extent_bool,
geo=True,
).opts(alpha=alpha)
```
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.