# pylint: disable=protected-access

import os
import glob

import radical.utils as ru

from ..       import states as s
from .session import fetch_json

_debug = os.environ.get('RP_PROF_DEBUG')


# ------------------------------------------------------------------------------
#
# pilot and unit activities: core hours are derived by multiplying the
# respective time durations with pilot size / unit size.  The 'idle'
# utilization and the 'agent' utilization are derived separately.
#
# Note that durations should add up to the `x_total` generations to ensure
# accounting for the complete unit/pilot utilization.
#
# An updated list of events is available at docs/source/events.md


PILOT_DURATIONS = {
    'provide' : {
        'total'     : [{ru.EVENT: 'bootstrap_0_start'},
                       {ru.EVENT: 'bootstrap_0_stop' }]
    },
    # times between PMGR_ACTIVE and the termination command are not
    # considered pilot specific consumptions.  If some resources remain
    # unused during that time, it is either due to inefficiencies of
    # workload management (accounted for in the unit consumption metrics),
    # or the pilot is starving for workload.
    'consume' : {
        'boot'      : [{ru.EVENT: 'bootstrap_0_start'},
                       {ru.EVENT: 'sync_rel'         }],
        'setup_1'   : [{ru.EVENT: 'sync_rel'         },
                       {ru.STATE: s.PMGR_ACTIVE      }],
        'ignore'    : [{ru.STATE: s.PMGR_ACTIVE      },
                       {ru.EVENT: 'cmd'              ,
                        ru.MSG  : 'cancel_pilot'     }],
        'term'      : [{ru.EVENT: 'cmd'              ,
                        ru.MSG  : 'cancel_pilot'     },
                       {ru.EVENT: 'bootstrap_0_stop' }],
    },
    # FIXME: separate out DVM startup time
    #   'rte'       : [{ru.STATE: s.PMGR_ACTIVE    },
    #                  {ru.STATE: s.PMGR_ACTIVE    }],
    #   'setup_2'   : [{ru.STATE: s.PMGR_ACTIVE    },
    #                  {ru.STATE: s.PMGR_ACTIVE    }],
    #
    # resources on agent nodes are consumed for all of the pilot's lifetime
    'agent' : {
        'total'     : [{ru.EVENT: 'bootstrap_0_start'},
                       {ru.EVENT: 'bootstrap_0_stop' }]
    }
}


# The set of default unit durations that are available for every unit
# description, default resource configuration, and default scheduler and
# launcher.
UNIT_DURATIONS_DEFAULT = {
    'consume' : {
        'exec_queue'  : [{ru.EVENT: 'schedule_ok'            },
                         {ru.STATE: s.AGENT_EXECUTING        }],
        'exec_prep'   : [{ru.STATE: s.AGENT_EXECUTING        },
                         {ru.EVENT: 'exec_start'             }],
        'exec_rp'     : [{ru.EVENT: 'exec_start'             },
                         {ru.EVENT: 'cu_start'               }],
        'exec_sh'     : [{ru.EVENT: 'cu_start'               },
                         {ru.EVENT: 'cu_exec_start'          }],
        'exec_cmd'    : [{ru.EVENT: 'cu_exec_start'          },
                         {ru.EVENT: 'cu_exec_stop'           }],
        'term_sh'     : [{ru.EVENT: 'cu_exec_stop'           },
                         {ru.EVENT: 'cu_stop'                }],
        'term_rp'     : [{ru.EVENT: 'cu_stop'                },
                         {ru.EVENT: 'exec_stop'              }],
        'unschedule'  : [{ru.EVENT: 'exec_stop'              },
                         {ru.EVENT: 'unschedule_stop'        }]

      # # if we have cmd_start / cmd_stop:
      # 'exec_sh'     : [{ru.EVENT: 'cu_start'               },
      #                  {ru.EVENT: 'cmd_start'              }],
      # 'exec_cmd'    : [{ru.EVENT: 'cmd_start'              },
      #                  {ru.EVENT: 'cmd_stop'               }],
      # 'term_sh'     : [{ru.EVENT: 'cmd_stop'               },
      #                  {ru.EVENT: 'cu_stop'                }],
    }
}


# The set of default unit durations augmented with the durations of the app
# events. App events are generated by RADICAL Synapse and by `hello_rp.sh`. The
# latter is useful for testing as a sleep command drop-in.
UNIT_DURATIONS_APP = {
    'consume' : {
        'exec_queue'  : [{ru.EVENT: 'schedule_ok'            },
                         {ru.STATE: s.AGENT_EXECUTING        }],
        'exec_prep'   : [{ru.STATE: s.AGENT_EXECUTING        },
                         {ru.EVENT: 'exec_start'             }],
        'exec_rp'     : [{ru.EVENT: 'exec_start'             },
                         {ru.EVENT: 'cu_start'               }],
        'exec_sh'     : [{ru.EVENT: 'cu_start'               },
                         {ru.EVENT: 'cu_exec_start'          }],
        'init_app'    : [{ru.EVENT: 'cu_exec_start'          },
                         {ru.EVENT: 'app_start'              }],
        'exec_cmd'    : [{ru.EVENT: 'app_start'              },
                         {ru.EVENT: 'app_stop'               }],
        'term_app'    : [{ru.EVENT: 'app_stop'               },
                         {ru.EVENT: 'cu_exec_stop'           }],
        'term_sh'     : [{ru.EVENT: 'cu_exec_stop'           },
                         {ru.EVENT: 'cu_stop'                }],
        'term_rp'     : [{ru.EVENT: 'cu_stop'                },
                         {ru.EVENT: 'exec_stop'              }],
        'unschedule'  : [{ru.EVENT: 'exec_stop'              },
                         {ru.EVENT: 'unschedule_stop'        }]
    }
}


# The set of default unit durations with the durations generated when using
# PRRTE as launch method.
UNIT_DURATIONS_PRTE = {
    'consume' : {
        'exec_queue'  : [{ru.EVENT: 'schedule_ok'            },
                         {ru.STATE: s.AGENT_EXECUTING        }],
        'exec_prep'   : [{ru.STATE: s.AGENT_EXECUTING        },
                         {ru.EVENT: 'exec_start'             }],
        'exec_rp'     : [{ru.EVENT: 'exec_start'             },
                         {ru.EVENT: 'cu_start'               }],
        'exec_sh'     : [{ru.EVENT: 'cu_start'               },
                         {ru.EVENT: 'cu_exec_start'          }],
        'prte_phase_1': [{ru.EVENT: 'cu_exec_start'          },
                         {ru.EVENT: 'prte_init_complete'     }],
        'prte_phase_2': [{ru.EVENT: 'prte_init_complete'     },
                         {ru.EVENT: 'prte_sending_launch_msg'}],
        'exec_cmd'    : [{ru.EVENT: 'prte_sending_launch_msg'},
                         {ru.EVENT: 'prte_iof_complete'      }],
        'prte_phase_3': [{ru.EVENT: 'prte_iof_complete'      },
                         {ru.EVENT: 'prte_notify_completed'  }],
        'prte_phase_4': [{ru.EVENT: 'prte_notify_completed'  },
                         {ru.EVENT: 'cu_exec_stop'           }],
        'term_sh'     : [{ru.EVENT: 'cu_exec_stop'           },
                         {ru.EVENT: 'cu_stop'                }],
        'term_rp'     : [{ru.EVENT: 'cu_stop'                },
                         {ru.EVENT: 'exec_stop'              }],
        'unschedule'  : [{ru.EVENT: 'exec_stop'              },
                         {ru.EVENT: 'unschedule_stop'        }],

      # # if we have app_start / app_stop:
      # 'prte_phase_2': [{ru.EVENT: 'prte_init_complete'     },
      #                  {ru.EVENT: 'cmd_start'              }],
      # 'exec_cmd'    : [{ru.EVENT: 'cmd_start'              },
      #                  {ru.EVENT: 'cmd_stop'               }],
      # 'prte_phase_3': [{ru.EVENT: 'cmd_stop'               },
      #                  {ru.EVENT: 'prte_notify_completed'  }],
    }
}


# The set of default unit durations with the durations generated when using
# PRRTE as launch method and an app that records app events (e.g., RADICAL
# Synapse and `hello_rp.sh`).
UNIT_DURATIONS_PRTE_APP  = {
    'consume' : {
        'exec_queue'  : [{ru.EVENT: 'schedule_ok'            },
                         {ru.STATE: s.AGENT_EXECUTING        }],
        'exec_prep'   : [{ru.STATE: s.AGENT_EXECUTING        },
                         {ru.EVENT: 'exec_start'             }],
        'exec_rp'     : [{ru.EVENT: 'exec_start'             },
                         {ru.EVENT: 'cu_start'               }],
        'exec_sh'     : [{ru.EVENT: 'cu_start'               },
                         {ru.EVENT: 'cu_exec_start'          }],
        'prte_phase_1': [{ru.EVENT: 'cu_exec_start'          },
                         {ru.EVENT: 'prte_init_complete'     }],
        'prte_phase_2': [{ru.EVENT: 'prte_init_complete'     },
                         {ru.EVENT: 'prte_sending_launch_msg'}],
        'init_app'    : [{ru.EVENT: 'prte_sending_launch_msg'},
                         {ru.EVENT: 'app_start'              }],
        'exec_cmd'    : [{ru.EVENT: 'app_start'              },
                         {ru.EVENT: 'app_stop'               }],
        'term_app'    : [{ru.EVENT: 'app_stop'               },
                         {ru.EVENT: 'prte_iof_complete'      }],
        'prte_phase_3': [{ru.EVENT: 'prte_iof_complete'      },
                         {ru.EVENT: 'prte_notify_completed'  }],
        'prte_phase_4': [{ru.EVENT: 'prte_notify_completed'  },
                         {ru.EVENT: 'cu_exec_stop'           }],
        'term_sh'     : [{ru.EVENT: 'cu_exec_stop'           },
                         {ru.EVENT: 'cu_stop'                }],
        'term_rp'     : [{ru.EVENT: 'cu_stop'                },
                         {ru.EVENT: 'exec_stop'              }],
        'unschedule'  : [{ru.EVENT: 'exec_stop'              },
                         {ru.EVENT: 'unschedule_stop'        }]
    }
}


# ----------------------------------------------------------------------------
#
def _convert_sdurations(sdurations):
    '''
    Converts a collection of durations expressed in short form to the same
    collection of durations expressed in long form.

    Definitions:

    - Short form collection: one dictionary of short form durations
    - Long form: one dictionary of long form durations.

    Args:

        sdurations (dict): a collections of durations in short form

    Return:

        ldurations (dict): a collection of long form durations

    Example:

        sdurations = {'name_of_duration': [{'STATE': s.STATE_NAME},
                                           {'EVENT': 'event_name'}]}
        ldurations = {'name_of_duration': [{ru.EVENT: 'state',
                                            ru.STATE: s.STATE_NAME},
                                           {ru.EVENT: 'event_name',
                                            ru.STATE: None}]}
        sdurations = {'name_of_duration': [{'STATE': s.STATE_NAME},
                                           [{'EVENT': 'event_name'},
                                            {'STATE': s.STATE_NAME}]]}
        ldurations = {'name_of_duration': [{ru.EVENT: 'state',
                                            ru.STATE: s.STATE_NAME},
                                           [{ru.EVENT: 'event_name',
                                             ru.STATE: None},
                                            {ru.EVENT: 'state',
                                             ru.STATE: s.STATE_NAME}]]}
        sdurations = {'name_of_duration': [{'STATE': s.STATE_NAME},
                                           {'MSG': 'message_name'}]}
        ldurations = {'name_of_duration': [{ru.EVENT: 'state',
                                            ru.STATE: s.STATE_NAME},
                                           {ru.EVENT: 'cmd',
                                            ru.MSG: 'message_name'}]}
    '''

    ldurations = dict()

    for k,v in sdurations.items():

        ldurations[k] = list()
        for ts in v:

            if isinstance(ts, dict):
                ldurations[k].append(_expand_sduration(ts))

            if isinstance(ts, list):
                lds = list()
                for i in ts:
                    lds.append(_expand_sduration(i))
                ldurations[k].append(lds)

    return ldurations


# ----------------------------------------------------------------------------
#
def _expand_sduration(sduration):
    '''
    Expands a duration expressed in short form to its long form for the
    timestamp types `ru.STATE`, `ru.EVENT` and `ru.MSG`.

    Definitions:

    - Short form duration: one dictionary containing a state or event name.
    - Long form duration: one dictionary containing two keys, one of type
      `ru.EVENT` and one of type `ru.STATE`. The `ru.EVENT` key has a string
      value while the `ru.STATE` key has a `s.STATE_NAME` object as its value.

    Args:

        sduration (dict): a duration in short form

    Return:

        lduration (dict): sduration in long form

    Example:

        sduration = {'STATE': s.STATE_NAME}
        lduration = {ru.EVENT: 'state', ru.STATE: s.STATE_NAME}
        sduration = {'EVENT': 'event_name'}
        lduration = {ru.EVENT: 'event_name', ru.STATE: None}
        sduration = {'MSG': 'mesage_name'}
        lduration = {ru.EVENT: 'cmd', ru.MSG: 'message_name'}
    '''

    # Allow durations with both ru.EVENT and ru.STATE.
    tt = list(sduration.keys())
    if len(tt) == 1 and tt[0] not in ['STATE', 'EVENT', 'MSG']:
        raise Exception('unknown timestamp type: %s' % tt)
    if len(tt) == 2:
        return sduration
    if len(tt) > 2:
        raise Exception('invalid duration: too many timestamps (%s)' % tt)

    # Expand known timestamps.
    lduration = None

    for k,v in sduration.items():
        if k == 'STATE':
            lduration = {ru.EVENT: 'state', ru.STATE: v}
        elif k == 'EVENT':
            lduration = {ru.EVENT: v, ru.STATE: None}
        elif k == 'MSG':
            lduration = {ru.EVENT: 'cmd', ru.MSG: v}

    return lduration


# Set of default pilot durations for RADICAL-Analytics. All the durations
# are contiguos.
# NOTE: _init durations are most often 0.
PILOT_DURATIONS_DEBUG_SHORT = {
    'p_pmgr_create'           : [{'STATE': s.NEW                   },
                                 {'STATE': s.PMGR_LAUNCHING_PENDING}],
    'p_pmgr_launching_init'   : [{'STATE': s.PMGR_LAUNCHING_PENDING},
                                 {'STATE': s.PMGR_LAUNCHING        }],
    'p_pmgr_launching'        : [{'STATE': s.PMGR_LAUNCHING        },
                                 {'EVENT': 'staging_in_start'      }],
    'p_pmgr_stage_in'         : [{'EVENT': 'staging_in_start'      },
                                 {'EVENT': 'staging_in_stop'       }],
    'p_pmgr_submission_init'  : [{'EVENT': 'staging_in_stop'       },
                                 {'EVENT': 'submission_start'      }],
    'p_pmgr_submission'       : [{'EVENT': 'submission_start'      },
                                 {'EVENT': 'submission_stop'       }],
    'p_pmgr_scheduling_init'  : [{'EVENT': 'submission_stop'       },
                                 {'STATE': s.PMGR_ACTIVE_PENDING   }],
    # batch system queue time
    'p_pmgr_scheduling'       : [{'STATE': s.PMGR_ACTIVE_PENDING   },
                                 {'EVENT': 'bootstrap_0_start'     }],
    'p_agent_ve_setup_init'   : [{'EVENT': 'bootstrap_0_start'     },
                                 {'EVENT': 've_setup_start'        }],
    'p_agent_ve_setup'        : [{'EVENT': 've_setup_start'        },
                                 {'EVENT': 've_setup_stop'         }],
    'p_agent_ve_activate_init': [{'EVENT': 've_setup_stop'         },
                                 {'EVENT': 've_activate_start'     }],
    'p_agent_ve_activate'     : [{'EVENT': 've_activate_start'     },
                                 {'EVENT': 've_activate_stop'      }],
    'p_agent_install_init'    : [{'EVENT': 've_activate_stop'      },
                                 {'EVENT': 'rp_install_start'      }],
    'p_agent_install'         : [{'EVENT': 'rp_install_start'      },
                                 {'EVENT': 'rp_install_stop'       }],
    'p_agent_launching'       : [{'EVENT': 'rp_install_stop'       },
                                 {'STATE': s.PMGR_ACTIVE           }],
    'p_agent_terminate_init'  : [{'STATE': s.PMGR_ACTIVE           },
                                 {'MSG'  : 'cancel_pilot'          }],
    'p_agent_terminate'       : [{'MSG'  : 'cancel_pilot'          },
                                 {'EVENT': 'bootstrap_0_stop'      }],
    # total pilot runtime
    'p_agent_finalize'        : [{'EVENT': 'bootstrap_0_stop'      },
                                 [{'STATE': s.DONE                 },
                                  {'STATE': s.CANCELED             },
                                  {'STATE': s.FAILED               }]],
    'p_agent_runtime'         : [{'EVENT': 'bootstrap_0_start'     },
                                 {'EVENT': 'bootstrap_0_stop'      }]
}

PILOT_DURATIONS_DEBUG = _convert_sdurations(PILOT_DURATIONS_DEBUG_SHORT)


# Debug pilot durations tagged with keys taht can be used when calculating
# resource utilization.
# TODO: add the 'client' tag to relevant resource utilization methods.
_pdd = PILOT_DURATIONS_DEBUG
PILOT_DURATIONS_DEBUG_RU = {
    'provide' : {
        'p_agent_runtime'         : _pdd['p_agent_runtime']
    },
    'client'  : {
        'p_pmgr_create'           : _pdd['p_pmgr_create'],
        'p_pmgr_launching_init'   : _pdd['p_pmgr_launching_init'],
        'p_pmgr_launching'        : _pdd['p_pmgr_launching'],
        'p_pmgr_stage_in'         : _pdd['p_pmgr_stage_in'],
        'p_pmgr_submission_init'  : _pdd['p_pmgr_submission_init'],
        'p_pmgr_submission'       : _pdd['p_pmgr_submission'],
        'p_pmgr_scheduling_init'  : _pdd['p_pmgr_scheduling_init'],
        'p_pmgr_scheduling'       : _pdd['p_pmgr_scheduling'],
        'p_agent_finalize'        : _pdd['p_agent_finalize']
    },
    'consume' : {
        'p_agent_ve_setup_init'   : _pdd['p_agent_ve_setup_init'],
        'p_agent_ve_setup'        : _pdd['p_agent_ve_setup'],
        'p_agent_ve_activate_init': _pdd['p_agent_ve_activate_init'],
        'p_agent_ve_activate'     : _pdd['p_agent_ve_activate'],
        'p_agent_install_init'    : _pdd['p_agent_install_init'],
        'p_agent_install'         : _pdd['p_agent_install'],
        'p_agent_launching'       : _pdd['p_agent_launching'],
        'p_agent_terminate_init'  : _pdd['p_agent_terminate_init'],
        'p_agent_terminate'       : _pdd['p_agent_terminate']
    },
    'agent'   : {
        'p_agent_runtime'         : _pdd['p_agent_runtime']
    }
}


# Set of default unit durations for RADICAL-Analytics. All the durations
# are contiguos.
UNIT_DURATIONS_DEBUG_SHORT = {
    'u_umgr_create'              : [{'STATE': s.NEW                         },
                                    {'STATE': s.UMGR_SCHEDULING_PENDING     }],
    'u_umgr_schedule_queue'      : [{'STATE': s.UMGR_SCHEDULING_PENDING     },
                                    {'STATE': s.UMGR_SCHEDULING             }],
    'u_umgr_schedule'            : [{'STATE': s.UMGR_SCHEDULING             },
                                    {'STATE': s.UMGR_STAGING_INPUT_PENDING  }],
    # push to mongodb
    'u_umgr_stage_in_queue'      : [{'STATE': s.UMGR_STAGING_INPUT_PENDING  },
                                    {'STATE': s.UMGR_STAGING_INPUT          }],
    # wait in mongodb
    'u_umgr_stage_in'            : [{'STATE': s.UMGR_STAGING_INPUT          },
                                    {'STATE': s.AGENT_STAGING_INPUT_PENDING }],
    # pull from mongodb
    'u_agent_stage_in_queue'     : [{'STATE': s.AGENT_STAGING_INPUT_PENDING },
                                    {'STATE': s.AGENT_STAGING_INPUT         }],
    'u_agent_stage_in'           : [{'STATE': s.AGENT_STAGING_INPUT         },
                                    {'STATE': s.AGENT_SCHEDULING_PENDING    }],
    'u_agent_schedule_queue'     : [{'STATE': s.AGENT_SCHEDULING_PENDING    },
                                    {'STATE': s.AGENT_SCHEDULING            }],
    'u_agent_schedule'           : [{'STATE': s.AGENT_SCHEDULING            },
                                    {'STATE': s.AGENT_EXECUTING_PENDING     }],
    'u_agent_execute_queue'      : [{'STATE': s.AGENT_EXECUTING_PENDING     },
                                    {'STATE': s.AGENT_EXECUTING             }],
    'u_agent_execute_prepare'    : [{'STATE': s.AGENT_EXECUTING             },
                                    {'EVENT': 'exec_mkdir'                  }],
    'u_agent_execute_mkdir'      : [{'EVENT': 'exec_mkdir'                  },
                                    {'EVENT': 'exec_mkdir_done'             }],
    'u_agent_execute_layer_start': [{'EVENT': 'exec_mkdir_done'             },
                                    {'EVENT': 'exec_start'                  }],
    # orte, ssh, mpi, ...
    'u_agent_execute_layer'      : [{'EVENT': 'exec_start'                  },
                                    [{'EVENT': 'exec_ok'                    },
                                     {'EVENT': 'exec_fail'                  }]],
    # PROBLEM: discontinuity
    'u_agent_lm_start'           : [{'EVENT': 'cu_start'                    },
                                    {'EVENT': 'cu_pre_start'                }],
    'u_agent_lm_pre_execute'     : [{'EVENT': 'cu_pre_start'                },
                                    {'EVENT': 'cu_pre_stop'                 }],
    'u_agent_lm_execute_start'   : [{'EVENT': 'cu_pre_stop'                 },
                                    {'EVENT': 'cu_exec_start'               }],
    'u_agent_lm_execute'         : [{'EVENT': 'cu_exec_start'               },
                                    {'EVENT': 'cu_exec_stop'                }],
    'u_agent_lm_stop'            : [{'EVENT': 'cu_exec_stop'                },
                                    {'EVENT': 'cu_stop'                     }],
    'u_agent_stage_out_start'    : [{'EVENT': 'cu_stop'                     },
                                    {'STATE': s.AGENT_STAGING_OUTPUT_PENDING}],
    'u_agent_stage_out_queue'    : [{'STATE': s.AGENT_STAGING_OUTPUT_PENDING},
                                    {'STATE': s.AGENT_STAGING_OUTPUT        }],
    'u_agent_stage_out'          : [{'STATE': s.AGENT_STAGING_OUTPUT        },
                                    {'STATE': s.UMGR_STAGING_OUTPUT_PENDING }],
    # push/pull mongodb
    'u_agent_push_to_umgr'       : [{'STATE': s.UMGR_STAGING_OUTPUT_PENDING },
                                    {'STATE': s.UMGR_STAGING_OUTPUT         }],
    'u_umgr_destroy'             : [{'STATE': s.UMGR_STAGING_OUTPUT         },
                                    [{'STATE': s.DONE                       },
                                     {'STATE': s.CANCELED                   },
                                     {'STATE': s.FAILED                     }]],
    'u_agent_unschedule'         : [{'EVENT': 'unschedule_start'            },
                                    {'EVENT': 'unschedule_stop'             }]
}

UNIT_DURATIONS_DEBUG = _convert_sdurations(UNIT_DURATIONS_DEBUG_SHORT)


# Debug unit durations tagged with keys taht can be used when calculating
# resource utilization.
# TODO: add the 'client' tag to relevant resource utilization methods.
_udd = UNIT_DURATIONS_DEBUG
UNIT_DURATIONS_DEBUG_RU = {
    'client' : {
        'u_umgr_create'              : _udd['u_umgr_create'],
        'u_umgr_schedule_queue'      : _udd['u_umgr_schedule_queue'],
        'u_umgr_schedule'            : _udd['u_umgr_schedule'],
        'u_umgr_stage_in_queue'      : _udd['u_umgr_stage_in_queue'],
        'u_umgr_stage_in'            : _udd['u_umgr_stage_in'],
        'u_umgr_destroy'             : _udd['u_umgr_destroy'],
        'u_agent_unschedule'         : _udd['u_agent_unschedule']
    },
    'consume'  : {
        'u_agent_stage_in_queue'     : _udd['u_agent_stage_in_queue'],
        'u_agent_stage_in'           : _udd['u_agent_stage_in'],
        'u_agent_schedule_queue'     : _udd['u_agent_schedule_queue'],
        'u_agent_schedule'           : _udd['u_agent_schedule'],
        'u_agent_execute_queue'      : _udd['u_agent_execute_queue'],
        'u_agent_execute_prepare'    : _udd['u_agent_execute_prepare'],
        'u_agent_execute_mkdir'      : _udd['u_agent_execute_mkdir'],
        'u_agent_execute_layer_start': _udd['u_agent_execute_layer_start'],
        'u_agent_execute_layer'      : _udd['u_agent_execute_layer'],
        'u_agent_lm_start'           : _udd['u_agent_lm_start'],
        'u_agent_lm_pre_execute'     : _udd['u_agent_lm_pre_execute'],
        'u_agent_lm_execute_start'   : _udd['u_agent_lm_execute_start'],
        'u_agent_lm_execute'         : _udd['u_agent_lm_execute'],
        'u_agent_lm_stop'            : _udd['u_agent_lm_stop'],
        'u_agent_stage_out_start'    : _udd['u_agent_stage_out_start'],
        'u_agent_stage_out_queue'    : _udd['u_agent_stage_out_queue'],
        'u_agent_stage_out'          : _udd['u_agent_stage_out'],
        'u_agent_push_to_umgr'       : _udd['u_agent_push_to_umgr'],
    }
}


# ------------------------------------------------------------------------------
#
def get_hostmap(profile):
    '''
    We abuse the profile combination to also derive a pilot-host map, which
    will tell us on what exact host each pilot has been running.  To do so, we
    check for the PMGR_ACTIVE advance event in agent.0.prof, and use the NTP
    sync info to associate a hostname.
    '''
    # FIXME: This should be replaced by proper hostname logging
    #        in `pilot.resource_details`.

    hostmap = dict()  # map pilot IDs to host names
    for entry in profile:
        if  entry[ru.EVENT] == 'hostname':
            hostmap[entry[ru.UID]] = entry[ru.MSG]

    return hostmap


# ------------------------------------------------------------------------------
#
def get_hostmap_deprecated(profiles):
    '''
    This method mangles combine_profiles and get_hostmap, and is deprecated.
    At this point it only returns the hostmap
    '''

    hostmap = dict()  # map pilot IDs to host names
    for pname, prof in profiles.items():

        if not len(prof):
            continue

        if not prof[0][ru.MSG]:
            continue

        host, ip, _, _, _ = prof[0][ru.MSG].split(':')
        host_id = '%s:%s' % (host, ip)

        for row in prof:

            if 'agent.0.prof' in pname    and \
                row[ru.EVENT] == 'advance' and \
                row[ru.STATE] == s.PMGR_ACTIVE:
                hostmap[row[ru.UID]] = host_id
                break

    return hostmap


# ------------------------------------------------------------------------------
#
def get_session_profile(sid, src=None):

    if not src:
        src = "%s/%s" % (os.getcwd(), sid)

    if os.path.exists(src):
        # we have profiles locally
        profiles  = glob.glob("%s/*.prof"   % src)
        profiles += glob.glob("%s/*/*.prof" % src)
    else:
        # need to fetch profiles
        from .session import fetch_profiles
        profiles = fetch_profiles(sid=sid, skip_existing=True)

    #  filter out some frequent, but uninteresting events
    efilter = {ru.EVENT: [
                        # 'get',
                          'publish',
                          'schedule_skip',
                          'schedule_fail',
                          'staging_stderr_start',
                          'staging_stderr_stop',
                          'staging_stdout_start',
                          'staging_stdout_stop',
                          'staging_uprof_start',
                          'staging_uprof_stop',
                          'update_pushed',
                         ]}

    profiles          = ru.read_profiles(profiles, sid, efilter=efilter)
    profile, accuracy = ru.combine_profiles(profiles)
    profile           = ru.clean_profile(profile, sid, s.FINAL, s.CANCELED)
    hostmap           = get_hostmap(profile)

    if not hostmap:
        # FIXME: legacy host notation - deprecated
        hostmap = get_hostmap_deprecated(profiles)

    return profile, accuracy, hostmap


# ------------------------------------------------------------------------------
#
def get_session_description(sid, src=None, dburl=None):
    """
    This will return a description which is usable for radical.analytics
    evaluation.  It informs about
      - set of stateful entities
      - state models of those entities
      - event models of those entities (maybe)
      - configuration of the application / module

    If `src` is given, it is interpreted as path to search for session
    information (json dump).  `src` defaults to `$PWD/$sid`.

    if `dburl` is given, its value is used to fetch session information from
    a database.  The dburl value defaults to `RADICAL_PILOT_DBURL`.
    """

    if not src:
        src = "%s/%s" % (os.getcwd(), sid)

    if os.path.isfile('%s/%s.json' % (src, sid)):
        json = ru.read_json('%s/%s.json' % (src, sid))
    else:
        ftmp = fetch_json(sid=sid, dburl=dburl, tgt=src, skip_existing=True)
        json = ru.read_json(ftmp)

    # make sure we have uids
    # FIXME v0.47: deprecate
    def fix_json(json):
        def fix_uids(json):
            if isinstance(json, list):
                for elem in json:
                    fix_uids(elem)
            elif isinstance(json, dict):
                if 'unitmanager' in json and 'umgr' not in json:
                    json['umgr'] = json['unitmanager']
                if 'pilotmanager' in json and 'pmgr' not in json:
                    json['pmgr'] = json['pilotmanager']
                if '_id' in json and 'uid' not in json:
                    json['uid'] = json['_id']
                    if 'cfg' not in json:
                        json['cfg'] = dict()
                for v in json.values():
                    fix_uids(v)
        fix_uids(json)
    fix_json(json)

    assert(sid == json['session']['uid']), 'sid inconsistent'

    ret             = dict()
    ret['entities'] = dict()

    tree      = dict()
    tree[sid] = {'uid'      : sid,
                 'etype'    : 'session',
                 'cfg'      : json['session']['cfg'],
                 'has'      : ['umgr', 'pmgr'],
                 'children' : list()
                }

    for pmgr in sorted(json['pmgr'], key=lambda k: k['uid']):
        uid = pmgr['uid']
        tree[sid]['children'].append(uid)
        tree[uid] = {'uid'      : uid,
                     'etype'    : 'pmgr',
                     'cfg'      : pmgr['cfg'],
                     'has'      : ['pilot'],
                     'children' : list()
                    }

    for umgr in sorted(json['umgr'], key=lambda k: k['uid']):
        uid = umgr['uid']
        tree[sid]['children'].append(uid)
        tree[uid] = {'uid'      : uid,
                     'etype'    : 'umgr',
                     'cfg'      : umgr['cfg'],
                     'has'      : ['unit'],
                     'children' : list()
                    }
        # also inject the pilot description, and resource specifically
        tree[uid]['description'] = dict()

    for pilot in sorted(json['pilot'], key=lambda k: k['uid']):
        uid  = pilot['uid']
        pmgr = pilot['pmgr']
        pilot['cfg']['resource_details'] = pilot['resource_details']
        tree[pmgr]['children'].append(uid)
        tree[uid] = {'uid'        : uid,
                     'etype'      : 'pilot',
                     'cfg'        : pilot['cfg'],
                     'description': pilot['description'],
                     'has'        : ['unit'],
                     'children'   : list()
                    }
        # also inject the pilot description, and resource specifically

    for unit in sorted(json['unit'], key=lambda k: k['uid']):
        uid  = unit['uid']
        pid  = unit['pilot']
        umgr = unit['umgr']
        tree[pid ]['children'].append(uid)
        tree[umgr]['children'].append(uid)
        tree[uid] = {'uid'         : uid,
                     'etype'       : 'unit',
                     'cfg'         : unit,
                     'description' : unit['description'],
                     'has'         : list(),
                     'children'    : list()
                    }
        # remove duplicate
        del(tree[uid]['cfg']['description'])

    ret['tree'] = tree

    ret['entities']['pilot']   = {'state_model'  : s._pilot_state_values,
                                  'state_values' : s._pilot_state_inv_full,
                                  'event_model'  : dict()}
    ret['entities']['unit']    = {'state_model'  : s._unit_state_values,
                                  'state_values' : s._unit_state_inv_full,
                                  'event_model'  : dict()}
    ret['entities']['session'] = {'state_model'  : None,  # has no states
                                  'state_values' : None,
                                  'event_model'  : dict()}

    ret['config'] = dict()  # session config goes here

    return ret


# ------------------------------------------------------------------------------
#
def get_node_index(node_list, node, cpn, gpn):

    r0 = node_list.index(node) * (cpn + gpn)
    r1 = r0 + cpn + gpn - 1

    return [r0, r1]


# ------------------------------------------------------------------------------
#
def get_duration(thing, dur):


    for e in dur:
        if ru.STATE in e and ru.EVENT not in e:
            e[ru.EVENT] = 'state'

    t0 = thing.timestamps(event=dur[0])
    t1 = thing.timestamps(event=dur[1])

    if not len(t0) or not len(t1):
        return [None, None]

    return(t0[0], t1[-1])


# ------------------------------------------------------------------------------
#
def cluster_resources(resources):
    # resources is a list of
    #   - single index (single core of gpu
    #   - [r0, r1] tuples (ranges of core, gpu indexes)
    # cluster continuous stretches of resources

    ret = list()
    idx = set()
    for r in resources:
        if isinstance(r, int):
            idx.add(r)
        else:
            for i in range(r[0], r[1] + 1):
                idx.add(i)

    r0 = None
    r1 = None
    for i in sorted(list(idx)):
        if r0 is None:
            r0 = i
            continue
        if r1 is None:
            if i == r0 + 1:
                r1 = i
                continue
            ret.append([r0, r0])
            r0 = None
            continue
        if i == r1 + 1:
            r1 = i
            continue
        ret.append([r0, r1])
        r0 = i
        r1 = None

    if r0 is not None:
        if r1 is not None:
            ret.append([r0, r1])
        else:
            ret.append([r0, r0])

    return ret


# ------------------------------------------------------------------------------
#
def _get_pilot_provision(pilot):

    pid   = pilot.uid
    cpn   = pilot.cfg['resource_details']['rm_info']['cores_per_node']
    gpn   = pilot.cfg['resource_details']['rm_info']['gpus_per_node']
    ret   = dict()

    nodes, _, _ = _get_nodes(pilot)

    for metric in PILOT_DURATIONS['provide']:

        boxes  = list()
        t0, t1 = get_duration(pilot, PILOT_DURATIONS['provide'][metric])

        if t0 is None:
            t0 = pilot.events [0][ru.TIME]
            t1 = pilot.events[-1][ru.TIME]

        for node in nodes:
            r0, r1 = get_node_index(nodes, node, cpn, gpn)
            boxes.append([t0, t1, r0, r1])

        ret['total'] = {pid: boxes}

    return ret


# ------------------------------------------------------------------------------
#
def get_provided_resources(session):
    '''
    For all ra.pilots, return the amount and time of resources provided.
    This computes sets of 4-tuples of the form: [t0, t1, r0, r1] where:

        t0: time, begin of resource provision
        t1: time, begin of resource provision
        r0: int,  index of resources provided (min)
        r1: int,  index of resources provided (max)

    The tuples are formed so that t0 to t1 and r0 to r1 are continuous.
    '''

    provided = dict()
    for p in session.get(etype='pilot'):

        data = _get_pilot_provision(p)

        for metric in data:

            if metric not in provided:
                provided[metric] = dict()

            for uid in data[metric]:
                provided[metric][uid] = data[metric][uid]

    return provided


# ------------------------------------------------------------------------------
#
def get_consumed_resources(session):
    '''
    For all ra.pilot or ra.unit entities, return the amount and time of
    resources consumed.  A consumed resource is characterized by:

      - a resource type (we know about cores and gpus)
      - a metric name (what the resource was used for)
      - a list of 4-tuples of the form: [t0, t1, r0, r1]
          - t0: time, begin of resource consumption
          - t1: time, begin of resource consumption
          - r0: int,  index of resources consumed (min)
          - r1: int,  index of resources consumed (max)
        The tuples are formed so that t0 to t1 and r0 to r1 are continuous.

    An entity can consume different resources under different metrics - but the
    returned consumption specs will never overlap, meaning, that any resource is
    accounted for exactly one metric at any point in time.  The returned
    structure has the following overall form:

        {
          'metric_1' : {
              uid_1 : [[t0, t1, r0, r1],
                       [t2, t3, r2, r3],
                       ...
                      ],
              uid_2 : ...
          },
          'metric_2' : ...
        }
    '''

    log = ru.Logger('radical.pilot.utils')

    consumed = dict()
    for e in session.get(etype=['pilot', 'unit']):

        if   e.etype == 'pilot': data = _get_pilot_consumption(e)
        elif e.etype == 'unit' : data = _get_unit_consumption(session,  e)

        for metric in data:

            if metric not in consumed:
                consumed[metric] = dict()

            for uid in data[metric]:
                consumed[metric][uid] = data[metric][uid]


    # we defined two additional metrics, 'warmup' and 'drain', which are defined
    # for all resources of the pilot.  `warmup` is defined as the time from
    # when the pilot becomes active, to the time the resource is first consumed
    # by a unit.  `drain` is the inverse: the  time from when any unit last
    # consumed the resource to the time when the pilot begins termination.
    for pilot in session.get(etype='pilot'):

        if pilot.cfg['task_launch_method'] == 'PRTE':
          # print('\nusing prte configuration')
            unit_durations = UNIT_DURATIONS_PRTE
        else:
          # print('\nusing default configuration')
            unit_durations = UNIT_DURATIONS_DEFAULT

        pt    = pilot.timestamps
        log.debug('timestamps:')
        for ts in pt():
            log.debug('    %10.2f  %-20s  %-15s  %-15s  %-15s  %-15s  %s',
                           ts[0],  ts[1], ts[2], ts[3], ts[4], ts[5], ts[6])

        p_min = pt(event=PILOT_DURATIONS['consume']['ignore'][0]) [0]
        p_max = pt(event=PILOT_DURATIONS['consume']['ignore'][1])[-1]
      # p_max = pilot.events[-1][ru.TIME]
        log.debug('pmin, pmax: %10.2f / %10.2f', p_min, p_max)

        pid = pilot.uid
        cpn = pilot.cfg['resource_details']['rm_info']['cores_per_node']
        gpn = pilot.cfg['resource_details']['rm_info']['gpus_per_node']

        nodes, _, pnodes = _get_nodes(pilot)

        # find resource utilization scope for all resources. We begin filling
        # the resource dict with
        #
        #   resource_id : [t_min=None, t_max=None]
        #
        # and then iterate over all units.  Wen we find a unit using some
        # resource id, we set or adjust t_min / t_max.
        resources = dict()
        for pnode in pnodes:
            idx = get_node_index(nodes, pnode, cpn, gpn)
            for c in range(idx[0], idx[1] + 1):
                resources[c] = [None, None]


        for unit in session.get(etype='unit'):

            if unit.cfg.get('pilot') != pid:
                continue

            try:
                snodes = unit.cfg['slots']['nodes']
                ut     = unit.timestamps
                u_min  = ut(event=unit_durations['consume']['exec_queue'][0]) [0]
                u_max  = ut(event=unit_durations['consume']['unschedule'][1])[-1]
            except:
                continue

            for snode in snodes:

                node  = [snode['name'], snode['uid']]
                r0, _ = get_node_index(nodes, node, cpn, gpn)

                for core_map in snode['core_map']:
                    for core in core_map:
                        idx = r0 + core
                        t_min = resources[idx][0]
                        t_max = resources[idx][1]
                        if t_min is None or t_min > u_min: t_min = u_min
                        if t_max is None or t_max < u_max: t_max = u_max
                        resources[idx] = [t_min, t_max]

                for gpu_map in snode['gpu_map']:
                    for gpu in gpu_map:
                        idx = r0 + cpn + gpu
                        t_min = resources[idx][0]
                        t_max = resources[idx][1]
                        if t_min is None or t_min > u_min: t_min = u_min
                        if t_max is None or t_max < u_max: t_max = u_max
                        resources[idx] = [t_min, t_max]

        # now sift through resources and find buckets of pairs with same t_min
        # or same t_max
        bucket_min  = dict()
        bucket_max  = dict()
        bucket_none = list()
        for idx in resources:

            t_min = resources[idx][0]
            t_max = resources[idx][1]

            if t_min is None:

                assert(t_max is None)
                bucket_none.append(idx)

            else:

                if t_min not in bucket_min:
                    bucket_min[t_min] = list()
                bucket_min[t_min].append(idx)

                if t_max not in bucket_max:
                    bucket_max[t_max] = list()
                bucket_max[t_max].append(idx)

        boxes_warm  = list()
        boxes_drain = list()
        boxes_idle  = list()

        # now cluster all lists and add the respective boxes
        for t_min in bucket_min:
            for r in cluster_resources(bucket_min[t_min]):
                boxes_warm.append([p_min, t_min, r[0], r[1]])

        for t_max in bucket_max:
            for r in cluster_resources(bucket_max[t_max]):
                boxes_drain.append([t_max, p_max, r[0], r[1]])

        for r in cluster_resources(bucket_none):
            boxes_idle.append([p_min, p_max, r[0], r[1]])

        if 'warm'  not in consumed: consumed['warm']  = dict()
        if 'drain' not in consumed: consumed['drain'] = dict()
        if 'idle'  not in consumed: consumed['idle']  = dict()

        consumed['warm'][pid]  = boxes_warm
        consumed['drain'][pid] = boxes_drain
        consumed['idle'][pid]  = boxes_idle

  # pprint.pprint(consumed)

    return consumed


# ------------------------------------------------------------------------------
#
def _get_nodes(pilot):

    pnodes = pilot.cfg['resource_details']['rm_info']['node_list']
    agents = pilot.cfg['resource_details']['rm_info'].get('agent_nodes', [])
    anodes = list()
    nodes  = list()

    for agent in agents:
        anodes.append(agents[agent])

    nodes = pnodes + anodes

    return nodes, anodes, pnodes


# ------------------------------------------------------------------------------
#
def _get_pilot_consumption(pilot):

    # Pilots consume resources in different ways:
    #
    #   - the pilot needs to bootstrap and initialize before becoming active,
    #     i.e., before it can begin to manage the workload, and needs to
    #     terminate and clean up during shutdown;
    #   - the pilot may block one or more nodes or cores for it's own components
    #     (sub-agents), and those components are not available for workload
    #     execution
    #   - the pilot may perform operations while managing the workload.
    #
    # This method will compute the first two contributions and part of the 3rd.
    # It will *not* account for those parts of the 3rd which are performed while
    # specfic resources are blocked for the affected workload element (task)
    # - those resource consumption is considered to be a consumption *of that
    # task*, which allows us to compute tasks specific resource utilization
    # overheads.

    pid   = pilot.uid
    cpn   = pilot.cfg['resource_details']['rm_info']['cores_per_node']
    gpn   = pilot.cfg['resource_details']['rm_info']['gpus_per_node']
    ret   = dict()

    # Account for agent resources.  Agents use full nodes, i.e., cores and GPUs
    # We happen to know that agents use the first nodes in the allocation and
    # their resource tuples thus start at index `0`, but for completeness we
    # ensure that by inspecting the pilot cfg.

    # Duration is for all of the pilot runtime. This is not precises really,
    # since several bootstrapping phases happen before the agents exist - but we
    # consider the nodes blocked for the sub-agents from the get-go.
    t0, t1 = get_duration(pilot, PILOT_DURATIONS['agent']['total'])
    boxes  = list()

    # Substract agent nodes from the nodelist, so that we correctly attribute
    # other global pilot metrics to the remaining nodes.
    nodes, anodes, pnodes = _get_nodes(pilot)

    if anodes and t0 is not None:
        for anode in anodes:
            r0, r1 = get_node_index(nodes, anode, cpn, gpn)
            boxes.append([t0, t1, r0, r1])

    ret['agent'] = {pid: boxes}

    # account for all other pilot metrics
    for metric in PILOT_DURATIONS['consume']:

        if metric == 'ignore':
            continue

        boxes = list()
        t0, t1 = get_duration(pilot, PILOT_DURATIONS['consume'][metric])

        if t0 is not None:
            for node in pnodes:
                r0, r1 = get_node_index(nodes, node, cpn, gpn)
                boxes.append([t0, t1, r0, r1])

        ret[metric] = {pid: boxes}

    return ret


# ------------------------------------------------------------------------------
#
def _get_unit_consumption(session, unit):

    # we need to know what pilot the unit ran on.  If we don't find a designated
    # pilot, no resources were consumed
    uid = unit.uid
    pid = unit.cfg['pilot']

    if not pid:
        return dict()

    # get the pilot for inspection
    pilot = session.get(uid=pid)

    if isinstance(pilot, list):
        assert(len(pilot) == 1)
        pilot = pilot[0]

    # FIXME: it is inefficient to query those values again and again
    cpn   = pilot.cfg['resource_details']['rm_info']['cores_per_node']
    gpn   = pilot.cfg['resource_details']['rm_info']['gpus_per_node']
    nodes, _, _ = _get_nodes(pilot)

    # Units consume only those resources they are scheduled on.
    if 'slots' not in unit.cfg:
        return dict()

    snodes    = unit.cfg['slots']['nodes']
    resources = list()
    for snode in snodes:

        node  = [snode['name'], snode['uid']]
        r0, _ = get_node_index(nodes, node, cpn, gpn)

        for core_map in snode['core_map']:
            for core in core_map:
                resources.append(r0 + core)

        for gpu_map in snode['gpu_map']:
            for gpu in gpu_map:
                resources.append(r0 + cpn + gpu)

    # find continuous stretched of resources to minimize number of boxes
    resources = cluster_resources(resources)

    # we heuristically switch between PRTE event traces and normal (fork) event
    # traces
    if pilot.cfg['task_launch_method'] == 'PRTE':
        unit_durations = UNIT_DURATIONS_PRTE
    else:
        unit_durations = UNIT_DURATIONS_DEFAULT

    if _debug:
        print()

    ret = dict()
    for metric in unit_durations['consume']:

        boxes = list()
        t0, t1 = get_duration(unit, unit_durations['consume'][metric])


        if t0 is not None:

            if _debug:
                print('%s: %-15s : %10.3f - %10.3f = %10.3f'
                     % (unit.uid, metric, t1, t0, t1 - t0))
            for r in resources:
                boxes.append([t0, t1, r[0], r[1]])

        else:
            if _debug:
                print('%s: %-15s : -------------- ' % (unit.uid, metric))
                dur = unit_durations['consume'][metric]
                print(dur)

                for e in dur:
                    if ru.STATE in e and ru.EVENT not in e:
                        e[ru.EVENT] = 'state'

                t0 = unit.timestamps(event=dur[0])
                t1 = unit.timestamps(event=dur[1])
                print(t0)
                print(t1)
                for e in unit.events:
                    print('\t'.join([str(x) for x in e]))

              # sys.exit()

        ret[metric] = {uid: boxes}

    return ret


# ------------------------------------------------------------------------------

