<commit_before>"""Convenience routines for common actions on VOEvent objects"""

from __future__ import absolute_import

from collections import OrderedDict
from copy import deepcopy

import astropy.time
import iso8601
import lxml
import pytz
from voeventparse.misc import (Position2D)
from orderedmultidict import omdict as OMDict


def get_event_time_as_utc(voevent, index=0):
    """
    Extracts the event time from a given `WhereWhen.ObsDataLocation`.

    Returns a datetime (timezone-aware, UTC).

    Accesses a `WhereWhere.ObsDataLocation.ObservationLocation`
    element and returns the AstroCoords.Time.TimeInstant.ISOTime element,
    converted to a (UTC-timezoned) datetime.

    Note that a packet may include multiple 'ObsDataLocation' entries
    under the 'WhereWhen' section, for example giving locations of an object
    moving over time. Most packets will have only one, however, so the
    default is to access the first.

    This function now implements conversion from the
    TDB (Barycentric Dynamical Time) time scale in ISOTime format,
    since this is the format used by GAIA VOEvents.
    (See also http://docs.astropy.org/en/stable/time/#time-scale )

    Other timescales (i.e. TT, GPS) will presumably be formatted as a
    TimeOffset, parsing this format is not yet implemented.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent
            etree.
        index (int): Index of the ObsDataLocation to extract an ISOtime from.

    Returns:
        :class:`datetime.datetime`: Datetime representing the event-timestamp,
        converted to UTC (timezone aware).

    """
    try:
        od = voevent.WhereWhen.ObsDataLocation[index]
        ol = od.ObservationLocation
        coord_sys = ol.AstroCoords.attrib['coord_system_id']
        timesys_identifier = coord_sys.split('-')[0]

        if timesys_identifier == 'UTC':
            isotime_str = str(ol.AstroCoords.Time.TimeInstant.ISOTime)
            return iso8601.parse_date(isotime_str)
        elif (timesys_identifier == 'TDB'):
            isotime_str = str(ol.AstroCoords.Time.TimeInstant.ISOTime)
            isotime_dtime = iso8601.parse_date(isotime_str)
            tdb_time = astropy.time.Time(isotime_dtime, scale='tdb')
            return tdb_time.utc.to_datetime().replace(tzinfo=pytz.UTC)
        elif (timesys_identifier == 'TT' or timesys_identifier == 'GPS'):
            raise NotImplementedError(
                "Conversion from time-system '{}' to UTC not yet implemented"
            )
        else:
            raise ValueError(
                'Unrecognised time-system: {} (badly formatted VOEvent?)'.format(
                    timesys_identifier
                )
            )

    except AttributeError:
        return None


def get_event_position(voevent, index=0):
    """Extracts the `AstroCoords` from a given `WhereWhen.ObsDataLocation`.

    Note that a packet may include multiple 'ObsDataLocation' entries
    under the 'WhereWhen' section, for example giving locations of an object
    moving over time. Most packets will have only one, however, so the
    default is to just return co-ords extracted from the first.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the
            VOEvent etree.
        index (int): Index of the ObsDataLocation to extract AstroCoords from.

    Returns:
        Position (:py:class:`.Position2D`): The sky position defined in the
        ObsDataLocation.
    """
    od = voevent.WhereWhen.ObsDataLocation[index]
    ac = od.ObservationLocation.AstroCoords
    ac_sys = voevent.WhereWhen.ObsDataLocation.ObservationLocation.AstroCoordSystem
    sys = ac_sys.attrib['id']

    if hasattr(ac.Position2D, "Name1"):
        assert ac.Position2D.Name1 == 'RA' and ac.Position2D.Name2 == 'Dec'
    posn = Position2D(ra=float(ac.Position2D.Value2.C1),
                      dec=float(ac.Position2D.Value2.C2),
                      err=float(ac.Position2D.Error2Radius),
                      units=ac.Position2D.attrib['unit'],
                      system=sys)
    return posn

def _get_param_children_as_omdict(subtree_element):
    elt = subtree_element
    omd = OMDict()
    if elt.find('Param') is not None:
        for p in elt.Param:
            omd.add(p.attrib.get('name'),p.attrib)
    return omd

def get_grouped_params(voevent):
    """
    Fetch grouped Params from the `What` section of a voevent as an omdict.

    This fetches 'grouped' Params, i.e. those enclosed in a Group element,
    and returns them as a nested dict-like structure, keyed by
    GroupName->ParamName->AttribName.

    Note that since multiple Params may share the same ParamName, the returned
    data-structure is actually an
    `orderedmultidict.omdict <https://github.com/gruns/orderedmultidict>`_
    and has extra methods such as 'getlist' to allow retrieval of all values.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent etree.
    Returns (orderedmultidict.omdict):
        Mapping of ``ParamName->Attribs``.
        Typical access like so::

            foo_val = top_params['foo']['value']
            # If there are multiple Param entries named 'foo':
            all_foo_vals = [atts['value'] for atts in top_params.getlist('foo')]

    """
    groups_omd = OMDict()
    w = deepcopy(voevent.What)
    lxml.objectify.deannotate(w)
    if w.find('Group') is not None:
        for grp in w.Group:
            groups_omd.add(grp.attrib.get('name'),
                           _get_param_children_as_omdict(grp))
    return groups_omd


def get_toplevel_params(voevent):
    """
    Fetch ungrouped Params from the `What` section of a voevent as an omdict.

    This fetches 'toplevel' Params, i.e. those not enclosed in a Group element,
    and returns them as a nested dict-like structure, keyed like
    ParamName->AttribName.

    Note that since multiple Params may share the same ParamName, the returned
    data-structure is actually an
    `orderedmultidict.omdict <https://github.com/gruns/orderedmultidict>`_
    and has extra methods such as 'getlist' to allow retrieval of all values.

    Any Params with no defined name (technically off-spec, but not invalidated
    by the XML schema) are returned under the dict-key ``None``.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent etree.
    Returns (orderedmultidict.omdict):
        Mapping of ``ParamName->Attribs``.
        Typical access like so::

            foo_val = top_params['foo']['value']
            # If there are multiple Param entries named 'foo':
            all_foo_vals = [atts['value'] for atts in top_params.getlist('foo')]

    """
    result = OrderedDict()
    w = deepcopy(voevent.What)
    lxml.objectify.deannotate(w)
    return _get_param_children_as_omdict(w)


def pull_astro_coords(voevent, index=0):
    """
    Deprecated alias of :func:`.get_event_position`
    """
    import warnings
    warnings.warn(
        """
        The function `get_event_position` has been renamed to
        `get_event_position1`. This alias is preserved for backwards
        compatibility, and may be removed in a future release.
        """,
        FutureWarning)
    return get_event_position(voevent, index)


def pull_isotime(voevent, index=0):
    """
    Deprecated alias of :func:`.get_event_time_as_utc`
    """
    import warnings
    warnings.warn(<commit_msg>Update misleading deprecation warning<commit_after>"""Convenience routines for common actions on VOEvent objects"""

from __future__ import absolute_import

from collections import OrderedDict
from copy import deepcopy

import astropy.time
import iso8601
import lxml
import pytz
from voeventparse.misc import (Position2D)
from orderedmultidict import omdict as OMDict


def get_event_time_as_utc(voevent, index=0):
    """
    Extracts the event time from a given `WhereWhen.ObsDataLocation`.

    Returns a datetime (timezone-aware, UTC).

    Accesses a `WhereWhere.ObsDataLocation.ObservationLocation`
    element and returns the AstroCoords.Time.TimeInstant.ISOTime element,
    converted to a (UTC-timezoned) datetime.

    Note that a packet may include multiple 'ObsDataLocation' entries
    under the 'WhereWhen' section, for example giving locations of an object
    moving over time. Most packets will have only one, however, so the
    default is to access the first.

    This function now implements conversion from the
    TDB (Barycentric Dynamical Time) time scale in ISOTime format,
    since this is the format used by GAIA VOEvents.
    (See also http://docs.astropy.org/en/stable/time/#time-scale )

    Other timescales (i.e. TT, GPS) will presumably be formatted as a
    TimeOffset, parsing this format is not yet implemented.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent
            etree.
        index (int): Index of the ObsDataLocation to extract an ISOtime from.

    Returns:
        :class:`datetime.datetime`: Datetime representing the event-timestamp,
        converted to UTC (timezone aware).

    """
    try:
        od = voevent.WhereWhen.ObsDataLocation[index]
        ol = od.ObservationLocation
        coord_sys = ol.AstroCoords.attrib['coord_system_id']
        timesys_identifier = coord_sys.split('-')[0]

        if timesys_identifier == 'UTC':
            isotime_str = str(ol.AstroCoords.Time.TimeInstant.ISOTime)
            return iso8601.parse_date(isotime_str)
        elif (timesys_identifier == 'TDB'):
            isotime_str = str(ol.AstroCoords.Time.TimeInstant.ISOTime)
            isotime_dtime = iso8601.parse_date(isotime_str)
            tdb_time = astropy.time.Time(isotime_dtime, scale='tdb')
            return tdb_time.utc.to_datetime().replace(tzinfo=pytz.UTC)
        elif (timesys_identifier == 'TT' or timesys_identifier == 'GPS'):
            raise NotImplementedError(
                "Conversion from time-system '{}' to UTC not yet implemented"
            )
        else:
            raise ValueError(
                'Unrecognised time-system: {} (badly formatted VOEvent?)'.format(
                    timesys_identifier
                )
            )

    except AttributeError:
        return None


def get_event_position(voevent, index=0):
    """Extracts the `AstroCoords` from a given `WhereWhen.ObsDataLocation`.

    Note that a packet may include multiple 'ObsDataLocation' entries
    under the 'WhereWhen' section, for example giving locations of an object
    moving over time. Most packets will have only one, however, so the
    default is to just return co-ords extracted from the first.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the
            VOEvent etree.
        index (int): Index of the ObsDataLocation to extract AstroCoords from.

    Returns:
        Position (:py:class:`.Position2D`): The sky position defined in the
        ObsDataLocation.
    """
    od = voevent.WhereWhen.ObsDataLocation[index]
    ac = od.ObservationLocation.AstroCoords
    ac_sys = voevent.WhereWhen.ObsDataLocation.ObservationLocation.AstroCoordSystem
    sys = ac_sys.attrib['id']

    if hasattr(ac.Position2D, "Name1"):
        assert ac.Position2D.Name1 == 'RA' and ac.Position2D.Name2 == 'Dec'
    posn = Position2D(ra=float(ac.Position2D.Value2.C1),
                      dec=float(ac.Position2D.Value2.C2),
                      err=float(ac.Position2D.Error2Radius),
                      units=ac.Position2D.attrib['unit'],
                      system=sys)
    return posn

def _get_param_children_as_omdict(subtree_element):
    elt = subtree_element
    omd = OMDict()
    if elt.find('Param') is not None:
        for p in elt.Param:
            omd.add(p.attrib.get('name'),p.attrib)
    return omd

def get_grouped_params(voevent):
    """
    Fetch grouped Params from the `What` section of a voevent as an omdict.

    This fetches 'grouped' Params, i.e. those enclosed in a Group element,
    and returns them as a nested dict-like structure, keyed by
    GroupName->ParamName->AttribName.

    Note that since multiple Params may share the same ParamName, the returned
    data-structure is actually an
    `orderedmultidict.omdict <https://github.com/gruns/orderedmultidict>`_
    and has extra methods such as 'getlist' to allow retrieval of all values.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent etree.
    Returns (orderedmultidict.omdict):
        Mapping of ``ParamName->Attribs``.
        Typical access like so::

            foo_val = top_params['foo']['value']
            # If there are multiple Param entries named 'foo':
            all_foo_vals = [atts['value'] for atts in top_params.getlist('foo')]

    """
    groups_omd = OMDict()
    w = deepcopy(voevent.What)
    lxml.objectify.deannotate(w)
    if w.find('Group') is not None:
        for grp in w.Group:
            groups_omd.add(grp.attrib.get('name'),
                           _get_param_children_as_omdict(grp))
    return groups_omd


def get_toplevel_params(voevent):
    """
    Fetch ungrouped Params from the `What` section of a voevent as an omdict.

    This fetches 'toplevel' Params, i.e. those not enclosed in a Group element,
    and returns them as a nested dict-like structure, keyed like
    ParamName->AttribName.

    Note that since multiple Params may share the same ParamName, the returned
    data-structure is actually an
    `orderedmultidict.omdict <https://github.com/gruns/orderedmultidict>`_
    and has extra methods such as 'getlist' to allow retrieval of all values.

    Any Params with no defined name (technically off-spec, but not invalidated
    by the XML schema) are returned under the dict-key ``None``.

    Args:
        voevent (:class:`voeventparse.voevent.Voevent`): Root node of the VOevent etree.
    Returns (orderedmultidict.omdict):
        Mapping of ``ParamName->Attribs``.
        Typical access like so::

            foo_val = top_params['foo']['value']
            # If there are multiple Param entries named 'foo':
            all_foo_vals = [atts['value'] for atts in top_params.getlist('foo')]

    """
    result = OrderedDict()
    w = deepcopy(voevent.What)
    lxml.objectify.deannotate(w)
    return _get_param_children_as_omdict(w)


def pull_astro_coords(voevent, index=0):
    """
    Deprecated alias of :func:`.get_event_position`
    """
    import warnings
    warnings.warn(
        """
        The function `pull_astro_coords` has been renamed to
        `get_event_position`. This alias is preserved for backwards
        compatibility, and may be removed in a future release.
        """,
        FutureWarning)
    return get_event_position(voevent, index)


def pull_isotime(voevent, index=0):
    """
    Deprecated alias of :func:`.get_event_time_as_utc`
    """
    import warnings
    warnings.warn(
{'fraction_non_alphanumeric': 0.08698556921432389, 'fraction_numerical': 0.004743452699091395, 'mean_word_length': 2.8540164778578783, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5154870', 'n_tokens_mistral': 4555, 'n_tokens_neox': 4383, 'n_words': 1328}