Source code for floatcsep.postprocess.plot_handler

import importlib.util
import logging
import os
from typing import TYPE_CHECKING, Union

from cartopy import crs as ccrs
from matplotlib import pyplot

from floatcsep.utils.helpers import (
    timewindow2str,
    magnitude_vs_time,
)

if TYPE_CHECKING:
    from floatcsep.experiment import Experiment

log = logging.getLogger("floatLogger")


[docs] def plot_results(experiment: "Experiment") -> None: """ Plots all evaluation results, according to the plotting function given in the tests configuration file. Args: experiment: The experiment instance, whose results were already calculated. """ log.info("Plotting evaluation results") timewindows = timewindow2str(experiment.timewindows) for test in experiment.tests: test.plot_results(timewindows, experiment.models, experiment.registry)
[docs] def plot_forecasts(experiment: "Experiment") -> None: """ Plots and saves all the generated forecasts. It can be set specified in the experiment ``config.yml`` as: :: postprocess: plot_forecasts: True or by specifying arguments as: :: postprocess: plot_forecasts: projection: Mercator basemap: google-satellite cmap: magma The default is ``plot_forecasts: True`` Args: experiment: The experiment instance, whose models were already run and their forecast are located in the filesystem/database """ # Parsing plot configuration file plot_forecast_config: dict = parse_plot_config( experiment.postprocess.get("plot_forecasts", {}) ) if not isinstance(plot_forecast_config, dict): return ##################################### # Default forecast plotting function. ##################################### log.info("Plotting forecasts") # Get the time windows to be plotted. Defaults to only the last time window. time_windows = ( timewindow2str(experiment.timewindows) if plot_forecast_config.get("all_time_windows") else [timewindow2str(experiment.timewindows[-1])] ) # Get the projection of the plots plot_forecast_config["projection"]: ccrs.Projection = parse_projection( plot_forecast_config.get("projection") ) for model in experiment.models: for window in time_windows: ax = model.get_forecast(window, experiment.region).plot( plot_args=plot_forecast_config ) # If catalog option is passed, catalog is plotted on top of the forecast if plot_forecast_config.get("catalog"): cat_args = plot_forecast_config.get("catalog", {}) if cat_args is True: cat_args = {} experiment.catalog_repo.get_test_cat(window).plot( ax=ax, extent=ax.get_extent(), plot_args=cat_args.update( { "basemap": plot_forecast_config.get("basemap", None), "title": ax.get_title(), } ), ) fig_path = experiment.registry.get_figure(window, "forecasts", model.name) pyplot.savefig(fig_path, dpi=plot_forecast_config.get("dpi", 300))
[docs] def plot_catalogs(experiment: "Experiment") -> None: """ Plots and saves the testing catalogs. It can be set specified in the experiment ``config.yml`` as: :: postprocess: plot_catalog: True or by specifying arguments as: :: postprocess: plot_catalog: projection: Mercator basemap: google-satellite markersize: 2 The default is ``plot_catalog: True`` Args: experiment: The experiment instance, whose catalogs were already accessed and filtered. """ # Parsing plot configuration file plot_catalog_config: dict = parse_plot_config( experiment.postprocess.get("plot_catalog", {}) ) if not isinstance(plot_catalog_config, dict): return #################################### # Default catalog plotting function. #################################### log.info("Plotting catalogs") # Get the projection of the plots plot_catalog_config["projection"]: ccrs.Projection = parse_projection( plot_catalog_config.get("projection") ) # Get the start and end dates of the experiment (as a string) experiment_timewindow = timewindow2str([experiment.start_date, experiment.end_date]) # Get the catalog for the entire duration of the experiment main_catalog = experiment.catalog_repo.get_test_cat(experiment_timewindow) # Skip plotting if no events if main_catalog.get_number_of_events() == 0: log.debug(f"Catalog has zero events in {experiment_timewindow}") return # Plot catalog map ax = main_catalog.plot(plot_args=plot_catalog_config) cat_map_path = experiment.registry.get_figure("main_catalog_map") ax.get_figure().savefig(cat_map_path, dpi=plot_catalog_config.get("dpi", 300)) # Plot catalog time series vs. magnitude ax = magnitude_vs_time(main_catalog) cat_time_path = experiment.registry.get_figure("main_catalog_time") ax.get_figure().savefig(cat_time_path, dpi=plot_catalog_config.get("dpi", 300)) # If selected, plot the test catalogs for each of the time windows if plot_catalog_config.get("all_time_windows"): for tw in experiment.timewindows: test_catalog = experiment.catalog_repo.get_test_cat(timewindow2str(tw)) if test_catalog.get_number_of_events() != 0: log.debug(f"Catalog has zero events in {tw}. Skip plotting") continue ax = test_catalog.plot(plot_args=plot_catalog_config) cat_map_path = experiment.registry.get_figure(tw, "catalog_map") ax.get_figure().savefig(cat_map_path, dpi=plot_catalog_config.get("dpi", 300)) ax = magnitude_vs_time(test_catalog) cat_time_path = experiment.registry.get_figure(tw, "catalog_time") ax.get_figure().savefig(cat_time_path, dpi=plot_catalog_config.get("dpi", 300))
[docs] def plot_custom(experiment: "Experiment"): """ Hook for user-based plotting functions. It corresponds to a function within a python file, specified in the experiment ``config.yml`` as: :: postprocess: plot_custom: {module}.py:{function} Args: experiment: The experiment instance, whose models were already run and their forecast are located in the filesystem/database """ plot_config = parse_plot_config(experiment.postprocess.get("plot_custom", False)) if plot_config is None: return script_path, func_name = plot_config log.info(f"Plotting from script {script_path} and function {func_name}") script_abs_path = experiment.registry.abs(script_path) allowed_directory = os.path.dirname(experiment.registry.abs(experiment.config_file)) if not os.path.isfile(script_path) or ( os.path.dirname(script_abs_path) != os.path.realpath(allowed_directory) ): log.error(f"Script {script_path} is not in the configuration file directory.") log.info( "\t Skipping plotting. Script can be reallocated and re-run the plotting only" " by typing 'floatcsep plot {config}'" ) return module_name = os.path.splitext(os.path.basename(script_abs_path))[0] spec = importlib.util.spec_from_file_location(module_name, script_abs_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # Execute the script securely try: func = getattr(module, func_name) except AttributeError: log.error(f"Function {func_name} not found in {script_path}") log.info( "\t Skipping plotting. Plot script can be modified and re-run the plotting only" " by typing 'floatcsep plot {config}'" ) return try: func(experiment) except Exception as e: log.error(f"Error executing {func_name} from {script_path}: {e}") log.info( "\t Skipping plotting. Plot script can be modified and re-run the plotting only" " by typing 'floatcsep plot {config}'" ) return
[docs] def parse_plot_config(plot_config: Union[dict, str, bool]): """ Parses the configuration of a given plot directive, usually gotten from the experiment ``config.yml`` as: :: postprocess: {plot_config} Args: plot_config: The plotting directive, which can be a dictionary, a boolean, or a string. If it is a dictionary, then it is directly returned. If it is a boolean, then the default plotting configuration is used. If it is a string, then it is expected to be of the form ``{script_path}.py:{func_name}``. """ if plot_config is True: return {} elif plot_config in (None, False): return elif isinstance(plot_config, dict): return plot_config elif isinstance(plot_config, str): # Parse the script path and function name try: script_path, func_name = plot_config.split(".py:") script_path += ".py" return script_path, func_name except ValueError: log.error( f"Invalid format for custom plot function: {plot_config}. " "Try {script_name}.py:{func}" ) log.info( "\t Skipping plotting. The script can be modified and re-run the plotting only " "by typing 'floatcsep plot {config}'" ) return else: log.error("Plot configuration not understood. Skipping plotting") return
[docs] def parse_projection(proj_config: Union[dict, str, bool]): """ Retrieve projection configuration. e.g., as defined in the config file: :: projection: Mercator: central_longitude: 0.0 """ if proj_config is None: return ccrs.PlateCarree(central_longitude=0.0) if isinstance(proj_config, dict): proj_name, proj_args = next(iter(proj_config.items())) else: proj_name, proj_args = proj_config, {} if not isinstance(proj_name, str): return ccrs.PlateCarree(central_longitude=0.0) return getattr(ccrs, proj_name, ccrs.PlateCarree)(**proj_args)