我有一个参数化的输入数据类，它将加载的模型输出到管道的下一阶段：
class InputData(param.Parameterized):
# placeholders for the incoming parameters
load_sim_widget = param.ClassSelector(class_=LoadAdhSimulation)
att_widget = param.ClassSelector(class_=Attributes)
proj_widget = param.ClassSelector(class_=projections.Projection)
label = param.String(default='Basic Input', precedence=-1)
def __init__(self, **params):
super(InputData, self).__init__(**params)
self.dat_files = []
self.model = None
# output the adh_viz object
@param.output()
def adh_viz(self):
return self.load_data()
@param.depends('load_sim_widget.load_netcdf', 'load_sim_widget.adh_root_filename',
'load_sim_widget.adh_directory', watch=True)
def available_attributes(self):
att_list = []
# if netcdf is selected
if self.load_sim_widget.load_netcdf:
filename = os.path.join(self.load_sim_widget.adh_directory,
self.load_sim_widget.adh_root_filename + '.nc')
try:
# open the xarray dataset (does not load into memory)
ncdf = xr.open_dataset(filename)
except FileNotFoundError:
print('File Not Found: {}'.format(filename))
else:
with ncdf:
# enable and add all variables in the netcdf file
for var in ncdf.data_vars:
att_name = var.lower().replace(" ", "_")
# if the variable has results dimensions
if 'times' in ncdf[var].dims and 'nodes_ids' in ncdf[var].dims:
# add to the list
att_list.append(att_name)
# close the dataset
ncdf.close()
# otherwise read from available *.dat files
else:
# read in extension dicts & standardize extensions
ext_to_widget, widget_to_ext = attribute_dict()
# fix
ext_to_widget['error'] = ext_to_widget.pop('err_hyd')
# get the list of filenames # todo this isn't foolproof e.g. `SanDieg` finds files
glob_items = glob.glob(os.path.join(self.load_sim_widget.adh_directory,
self.load_sim_widget.adh_root_filename + '*.dat'))
# convert filenames list to attribute list
att_list = []
for filename in glob_items:
# suffix.append(get_variable_from_file_name(filename))
att_list.append(ext_to_widget[get_variable_from_file_name(filename)])
# dictionary for naming inconsistencies # todo add complexity for err_con in future
label_to_widget = {'error': 'error_hydro',
'depth-averaged_velocity': 'velocity'}
# loop over the inconsistencies
for key in label_to_widget.keys():
# if this is found in the list
if key in att_list:
# remove the key
att_list.remove(key)
# add back in the corrected value
att_list.append(label_to_widget[key])
# adjust available attribute widgets based on available data
for key, value in self.att_widget.params().items():
# if this attribute wasn't in the list
if key in att_list:
# ensure the widget is enabled
self.att_widget.params()[key].constant = False
# set the widget value
setattr(self.att_widget, key, True)
elif key != 'name':
# ensure the widget is enabled
self.att_widget.params()[key].constant = False
# uncheck the attribute
setattr(self.att_widget, key, False)
# disable attribute
self.att_widget.params()[key].constant = True
def load_data(self):
# if file is netcdf
if self.load_sim_widget.load_netcdf:
self.model, avail_attributes = load_model(self.load_sim_widget.adh_directory,
project_name=self.load_sim_widget.adh_root_filename,
netcdf=self.load_sim_widget.load_netcdf)
# request to load data from *dat files:
else:
# get a list of requested suffix strings
slist = self.att_widget.suffix_list(value=True)
# construct list of filenames
fnames = []
[fnames.append(os.path.join(self.load_sim_widget.adh_directory,
self.load_sim_widget.adh_root_filename + '_' + x + '.dat')) for x in slist]
# read the requested files
self.model, avail_attributes = load_model(self.load_sim_widget.adh_directory,
project_name=self.load_sim_widget.adh_root_filename,
netcdf=False, crs=self.proj_widget.get_crs(), filenames=fnames)
adh_viz = AdhViz()
adh_viz.set_model(self.model)
roams_model = None
return adh_viz
# visualize the page
def panel(self):
self.available_attributes() # for the initial load
return pn.Row(pn.Spacer(height=700), pn.Param(self.load_sim_widget, show_name=False),
pn.Param(self.att_widget, show_name=False), pn.Param(self.proj_widget, show_name=False),
name=self.label)
然后管道的下一阶段应该注入第一阶段的输出并将其可视化。
class ViewResults(param.Parameterized):
load_sim_widget = param.ClassSelector(class_=LoadAdhSimulation)
att_widget = param.ClassSelector(class_=Attributes)
proj_widget = param.ClassSelector(class_=projections.Projection)
display_range = param.ClassSelector(class_=display_opts.DisplayRangeOpts)
cmap_opts = param.ClassSelector(class_=display_opts.ColormapOpts)
adh_viz = param.ClassSelector(class_=AdhViz)
wmts_widget = param.ClassSelector(class_=display_opts.WMTS)
wireframe = param.ClassSelector(class_=display_opts.ShowElements)
def __init__(self, **params):
for k, p in self.params().items():
if k in params or k == 'name':
continue
params[k] = p.class_()
super(ViewResults, self).__init__(**params)
self.annotator = ESAnnotator(path_type=gv.Path,
crs=ccrs.GOOGLE_MERCATOR,
point_columns=['depth_elevation'],
poly_columns=['name']
)
self.map_pane_ = pn.Spacer(width=0)
self.analysis_pane_ = pn.Spacer(width=0)
self.tool_pane_ = pn.Spacer(width=0)
# self.adh_viz = adh_viz
@property
def tabs(self):
result = pn.panel(self.adh_viz, parameters=['result_label'], show_name=False)
disp_tab = pn.Column(self.wmts_widget,
pn.Pane(self.cmap_opts, show_name=False),
pn.Pane(self.display_range, show_name=False),
pn.Pane(self.wireframe),
result)
return [('Display', disp_tab)]
# what to pass out of this page (for pipeline)
@param.output()
def output(self):
pass
# how to build this page
def panel(self):
return pn.panel(self.run)
@param.depends('adh_viz.result_label', 'wireframe.mesh_elements') # never need watch=True
def run(self):
# create the meshes for the dynamic map
meshes = self.adh_viz.create_animation2()
if self.wireframe.mesh_elements is True:
edgepaths_overlay = self.adh_viz.view_elements() # transparent/ invisible overlay
else:
edgepaths_overlay = hv.Points(data=[]) # existing edgepaths overlay
# Define function which applies colormap and color_range
def apply_opts(obj, colormap, color_range):
return obj.options(cmap=colormap, height=600, width=600).redim.range(**{obj.vdims[0].name: color_range}).options(
clipping_colors={'NaN': 'transparent', 'min': 'transparent'})
if meshes.label == 'scalar':
# Apply the colormap and color range dynamically
dynamic = hv.util.Dynamic(rasterize(meshes), operation=apply_opts,
streams=[Params(self.cmap_opts), Params(self.display_range)]
) * \
self.wmts_widget.view() * \
self.annotator.polys * \
self.annotator.points * \
edgepaths_overlay
elif meshes.label == 'vector':
# Apply the colormap and color range dynamically
dynamic = hv.util.Dynamic(rasterize(vectorfield_to_paths(meshes, color='Magnitude', magnitude='Magnitude',
scale=0.05), aggregator='mean', cmap=process_cmap('viridis'), precompute=True),
operation=apply_opts, streams=[Params(self.cmap_opts),
Params(self.display_range)]) * \
self.wmts_widget.view() * \
self.annotator.polys * \
self.annotator.points * \
edgepaths_overlay
time = pn.panel(self.adh_viz, parameters=['time'], widgets={'time': pn.widgets.DiscretePlayer}, show_name=False)
# time = pn.panel(self.adh_viz, parameters=['time'], show_name=False)
hv_panel = pn.panel(dynamic)
map_pane = pn.Column(hv_panel[0], pn.Row(pn.Spacer(width=100), time))
tool_pane = pn.Tabs(*self.tabs)
return pn.Row(map_pane, pn.Spacer(width=100, height=900), pn.Column(tool_pane, pn.Pane(LOGO, width=300)))
这是仪表板：
def results_dashboard(directory=os.path.join(ROOTDIR, 'data/SanDiego'), rootname='SanDiego'):
# generic single simulation options
load_sim_widget = LoadAdhSimulation(adh_root_filename=rootname, adh_directory=directory)
att_widget = Attributes()
# projection options
proj_widget = projections.Projection(name='')
# generic display options
wmts_widget = display_opts.WMTS(name='')
display_range = display_opts.DisplayRangeOpts()
cmap_opts = display_opts.ColormapOpts()
wireframe = display_opts.ShowElements(name='')
# create stages for pipeline
stages = [
('Input', InputData(load_sim_widget=load_sim_widget, att_widget=att_widget, proj_widget=proj_widget)),
('View', ViewResults(load_sim_widget=load_sim_widget, att_widget=att_widget, proj_widget=proj_widget,
display_range=display_range, cmap_opts=cmap_opts, wmts_widget=wmts_widget,
wireframe=wireframe))
]
# create the pipeline
pipeline = pn.pipeline.Pipeline(stages, debug=True)
# modify button width (not exposed)
pipeline.layout[0][1]._widget_box.width = 100
pipeline.layout[0][2]._widget_box.width = 100
# return a display of the pipline
return pipeline.layout
但是，当我运行此仪表板时，第二阶段只是显示一个空的 adh_viz 对象。我已确保数据已加载，但似乎它没有正确传递它。
几天前我让这段代码正常工作，但我做了一个小改动，在这个过程中加载数据 EARLIER（应该与管道无关），现在它不起作用。
关于我在这里缺少什么的任何想法？
编辑添加：当我results_dashboard从课堂中提取代码并在 jupyter 笔记本中运行它时。我在控制台中收到错误：
Error: Model 'ClearTool' does not exist. This could be due to a widget or a custom model not being registered before first usage.
但是，如果我使用相同的内核（不要重新启动）但重新运行导入，我的管道将成功地将数据从一个阶段传递到下一个阶段。
results_dashboard不幸的是，如果我将代码留在类中，这将不起作用。
results_dashboard再次编辑：如果我将面板、hv 和 hv.extension('bokeh') 添加到笔记本上的导入并重新加载它们，它将与类一起工作。