config_obs.yml

# Workflow to extract and analyse observation data


# This workflow is made to know exactly where to restart if the code is stopped.
# It looks if the result of a task exists in the project catalog before executing the task.
# The workflow does NOT automatically remove intermediate files. You might run out of space.

# List of task to accomplish in the workflow
# Each task will iterate over all simulations before moving on to the next task.
# It might be useful to start by running it once over all tasks on a single simulation
#  (by modifying extract:simulation:search_data_catalogs:other_search_criteria:)
# to make sure all tasks work properly before launching it for all simulations.
tasks:
  #- extract # Extract the simulation and reference dataset with the right domain and period.
  #- regrid # Regrid the simulation onto the reference grid.
  #- cleanup # Join each individually adjusted variable back in one scenario dataset and clean up other details.
  #- indicators # Compute xclim indicators on the scenario.
  #- climatologies # Compute climatological operations for the indicators (mean, std, trend).
  - plotting # Make plots of results from climatological operations.
  #- ensembles # Compute the ensemble statistics on indicators, climatology and deltas.

# Task Arguments

extract:
  # reanalysis
  reconstruction:
    dask:
      n_workers: 2
      threads_per_worker: 3
      memory_limit:
        15GB
    search_data_catalogs:
      allow_resampling: False
      allow_conversion: True
      variables_and_freqs: # &var_and_freq # this is an anchor. more on anchors:  https://support.atlassian.com/bitbucket-cloud/docs/yaml-anchors/
        tasmax: D
        tasmin: D
        pr: D
        tas: D
      periods: &ref_period
        #- '1980' #'1989' # defaults to None == extract all available data!
        #- '2018' #'2018'
      other_search_criteria:
        source:
          - ERA5-Land
          - RDRS
    extract_dataset:
      region: &region
        name: Quebec
        method: shape
        tile_buffer: 1.5 #1.5
        shape: /scen3/braun/data/obs_synthese23/gis/lpr_000b16a_e_QC_simpl1.zip!lpr_000b16a_e_QC_simpl1.shp # shape:shape to create a dictionary (deprecated)
    stack_drop_nans: &stack
      False
    save:
      mode: o
      encoding:
        tasmax: &f32
          dtype: float32
        tasmin:
          dtype: float32
        pr:
          dtype: float32
        tas:
          dtype: float32
      rechunk:
        time: -1
        X: 30
        Y: 30
  # station observations AHCCD
  station-obs:
    dask:
      n_workers: 1
      threads_per_worker: 2
      memory_limit:
        12GB
    search_data_catalogs:
      allow_resampling: False
      allow_conversion: True
      variables_and_freqs:
        tasmax: D
        tasmin: D
        tas: D
        #pr: D
        #dtr: D
      periods: #*ref_period
      other_search_criteria:
        source: AHCCD
    extract_dataset:
      region:
        name: Quebec
        method: shape
        tile_buffer: 0
        buffer: 0.03 # needed to capture all QC stations
        shape: /scen3/braun/data/obs_synthese23/gis/lpr_000b16a_e_QC_simpl1.zip!lpr_000b16a_e_QC_simpl1.shp
        # shape: /scen3/braun/data/obs_synthese23/gis/region_admin_poly.zip!region_admin_poly.shp # includes over water
    validation_period:
      miss: "WMO"
      window: 35
      min_years: 25
      freq: "YS"
    xarray_open_kwargs:
      chunks:
        station: 1
    save:
      mode: o
      rechunk:
        time: -1
        station: 50
#  # simulations
#  simulation:
#    dask:
#      n_workers: 2
#      threads_per_worker: 3
#      memory_limit: 10GB
#    search_data_catalogs:
#      #match_hist_and_fut: True
#      allow_resampling: True
#      #allow_conversion: True
#      variables_and_freqs:
#        #pr: D
#        tas: 3H
#      periods: &sim_period
#        - '1979'
#        - '1985'
#      other_search_criteria: # put the simulations you want here
#        code:
#          - bcs
#    extract_dataset:
#      xr_combine_kwargs:
#        combine_attrs: override
#      xr_open_kwargs:
#        drop_variables:
#          - height
#        chunks:
#          lat: 10
#          lon: 10
#          time: 365
#      region: # all
#    # floor: D  # ???
#    save:
#      mode: o
#      encoding:
#        tas: #*f32
#          dtype: float32
#        #pr: *f32
#      rechunk:
#        time: -1
#        X: 40
#        Y: 40


# regrid

#regrid:
#  dask:
#    n_workers: 2
#    threads_per_worker: 5
#    memory_limit: 10GB
#  inputs:
#    type: simulation
#    processing_level: extracted
#  output:
#    type: reconstruction
#    processing_level: extracted
#  regrid_dataset: # this will be automatically passed to the function.
#    regridder_kwargs:
#      method: bilinear
#      extrap_method: inverse_dist
#      locstream_out: *stack
#      reuse_weights: False
#  save: &save_time_-1
#    mode: o
#    rechunk:
#      time: -1

#cleanup:
#  dask:
#    n_workers: 4
#    threads_per_worker: 3
#    memory_limit: "6GB"
#  search_data_catalogs:
#    variables_and_freqs:
#      tasmax: D
#      tasmin: D
#      pr: D
#    allow_conversion: True
#    allow_resampling: False
#    other_search_criteria:
#      processing_level: extracted
#  xscen_clean_up:
#    to_level: cleaned
#    maybe_unstack_dict:
#      stack_drop_nans: *stack
#      rechunk:
#        lat: 15
#        lon: 15
#        time: -1
#    variables_and_units: &units
#      tasmax: degC
#      tasmin: degC
#      pr: mm d-1
#    convert_calendar_kwargs:
#      target: standard
#      align_on: random
#    missing_by_var:
#      tasmax: interpolate
#      tasmin: interpolate
#      pr: [ 0 ]
#  save:
#    mode: o
#    encoding:
#      tasmax:
#        dtype: float32
#      tasmin:
#        dtype: float32
#      pr:
#        dtype: float32
#
#

indicators:
  dask:
    n_workers: 8
    threads_per_worker: 5
    memory_limit: "8GB"
  inputs:
    processing_level: extracted
  compute_indicators:
    to_level: indicators
  save: #*save_time_-1
    mode: o
    rechunk:
      time: -1

aggregate:
  dask:
    n_workers: 10
    threads_per_worker: 4
    memory_limit: "12GB"
  periods: [['1951', '1980'], ['1961', '1990'], ['1971', '2000'],
            ['1981', '2010'], ['1991', '2020'], ['1991', '2018']] # ['1980', '1985']]
  input:
    obs:
      processing_level:
         - indicators
  climatological_mean: # automatically passed to the function
    op: mean
#    stride: 5
#    window: 30
#    min_periods: 28
    to_level: climatology
  climatological_std: # automatically passed to the function
    op: std
#    stride: 5
#    window: 30
#    min_periods: 28
    to_level: climatology
  climatological_trend: # automatically passed to the function
    op: linregress
#    stride: 5
#    window: 30
#    min_periods: 28
    to_level: climatology
  save:
    mode: 'o'

plotting:
  dask:
    n_workers: 5
    threads_per_worker: 4
    memory_limit: "12GB"
#  periods: [['1951', '1980'], ['1961', '1990'], ['1971', '2000'],
#            ['1981', '2010'], ['1991', '2020'], ['1991', '2018']]
  data:
    processing_level: climatology
  xrfreq:
    AS-JAN: year
    QS-DEC: season
    MS: month
  col_wrap:
    year:
    season: 2
    month: 3
  gridmap_kwars:
    contourf: True
  # mean
  tg_mean_clim_mean: &mean
    agg_operation: mean
    label: ''
    limits:
      year:
        vmin: -15
        vmax: 15
      season:
        vmin: -30
        vmax: 30
      month:
        vmin: -30
        vmax: 30
    levels: 21
    ticks: linspace
    units: degC
    divergent: 0
#      cmap: RdBu_r
  tn_mean_clim_mean: *mean
  tx_mean_clim_mean: *mean
  # interannual
  tg_mean_clim_std: &std_inter
    agg_operation: std
    label: interannual
    limits:
      year: &vmin_vmax_std
        vmin: 0
        vmax: 5
      season: *vmin_vmax_std
      month: *vmin_vmax_std
    levels: &levels_std
      [0, 0.5, 1, 2, 3, 4, 5, 7.5, 10, 12.5, 15, 20]
    ticks: *levels_std
    units: K
    divergent: False
#      cmap: afmhot_r
  tn_mean_clim_std: *std_inter
  tx_mean_clim_std: *std_inter
  # intra
  tg_std_clim_mean: &std_intra
    agg_operation: std
    label: intra
    limits:
      year: *vmin_vmax_std
      season: *vmin_vmax_std
      month: *vmin_vmax_std
    levels: *levels_std
    ticks: *levels_std
    units: K
    divergent: False
#      cmap: afmhot_r
  tn_std_clim_mean: *std_intra
  tx_std_clim_mean: *std_intra
  # total
  tg_std_clim_total: &std_total
    agg_operation: std
    label: total
    limits:
      year: *vmin_vmax_std
      season: *vmin_vmax_std
      month: *vmin_vmax_std
    levels: *levels_std
    ticks: *levels_std
    units: K
    divergent: False
#      cmap: afmhot_r
  tn_std_clim_total: *std_total
  tx_std_clim_total: *std_total
  # trend
  tg_mean_clim_linregress: &trend
    agg_operation: linregress
    label: ''
    limits:
      year: &vmin_vmax_trend
        vmin: -2
        vmax: 2
      season: *vmin_vmax_trend
      month: *vmin_vmax_trend
    levels: 9
    ticks: linspace
    units: K
    divergent: 0
#        cmap: RdBu_r
  tn_mean_clim_linregress: *trend
  tx_mean_clim_linregress: *trend

#ensembles:
#  dask:
#    n_workers: 3
#    threads_per_worker: 5
#    memory_limit: "5GB"
#  processing_levels:
#    - indicators
#    - climatology
#    - delta
#  ensemble_stats: # automatically passed to the function
#    statistics:
#      ensemble_percentiles: {}
#    stats_kwargs:
#      split: False
#    common_attrs_only: True
#  save:
#    mode: o


# General Arguments

project: # argument to create the project
  name: ObsFlow - A workflow for observation data
  version: 0.2
  description: Workflow for extracting observation data and performing climatic analysis
  id: obsflow

scripting: # send an email when code fails or succeed
  subject: ObsFlow - A workflow for observation data
  send_mail_on_exit:
    msg_err: I crashed, something went wrong!
    on_error_only: True


dask: # general dask arguments
  array.slicing.split_large_chunks: False


logging: # general logging args
  formatters:
    default:
      format: '%(asctime)s %(levelname)-8s %(name)-15s %(message)s'
      datefmt: '%Y-%m-%d %H:%M:%S'
  handlers:
    console:
      class : logging.StreamHandler
      formatter: default
      level : INFO
    file:
      class: logging.FileHandler
      formatter: default
      level : DEBUG
  loggers:
    xscen:
      propagate: False
      level: INFO
      handlers: [file, console]


to_dataset_dict: # parameters to open datasets
  xarray_open_kwargs:
    decode_timedelta: False