#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
:class:`DisplayData`, its classes, interfaces and methods.

The classes in this module allow users and transform developers to define
static display data to be displayed when a pipeline runs.
:class:`~apache_beam.transforms.ptransform.PTransform` s,
:class:`~apache_beam.transforms.core.DoFn` s
and other pipeline components are subclasses of the :class:`HasDisplayData`
mixin. To add static display data to a component, you can override the
:meth:`HasDisplayData.display_data()` method.

Available classes:

* :class:`HasDisplayData` - Components that inherit from this class can have
  static display data shown in the UI.
* :class:`DisplayDataItem` - This class represents static display data
  elements.
* :class:`DisplayData` - Internal class that is used to create display data
  and communicate it to the API.
"""

from __future__ import absolute_import

import calendar
import inspect
import json
from builtins import object
from datetime import datetime
from datetime import timedelta

from past.builtins import unicode

__all__ = ['HasDisplayData', 'DisplayDataItem', 'DisplayData']


class HasDisplayData(object):
  """ Basic mixin for elements that contain display data.

  It implements only the display_data method and a _namespace method.
  """

  def display_data(self):
    """ Returns the display data associated to a pipeline component.

    It should be reimplemented in pipeline components that wish to have
    static display data.

    Returns:
      Dict[str, Any]: A dictionary containing ``key:value`` pairs.
      The value might be an integer, float or string value; a
      :class:`DisplayDataItem` for values that have more data
      (e.g. short value, label, url); or a :class:`HasDisplayData` instance
      that has more display data that should be picked up. For example::

        {
          'key1': 'string_value',
          'key2': 1234,
          'key3': 3.14159265,
          'key4': DisplayDataItem('apache.org', url='http://apache.org'),
          'key5': subComponent
        }
    """
    return {}

  def _namespace(self):
    return '{}.{}'.format(self.__module__, self.__class__.__name__)


class DisplayData(object):
  """ Static display data associated with a pipeline component.
  """

  def __init__(self, namespace, display_data_dict):
    self.namespace = namespace
    self.items = []
    self._populate_items(display_data_dict)

  def _populate_items(self, display_data_dict):
    """ Populates the list of display data items.
    """
    for key, element in display_data_dict.items():
      if isinstance(element, HasDisplayData):
        subcomponent_display_data = DisplayData(element._namespace(),
                                                element.display_data())
        self.items += subcomponent_display_data.items
        continue

      if isinstance(element, DisplayDataItem):
        if element.should_drop():
          continue
        element.key = key
        element.namespace = self.namespace
        self.items.append(element)
        continue

      # If it's not a HasDisplayData element,
      # nor a dictionary, then it's a simple value
      self.items.append(
          DisplayDataItem(element,
                          namespace=self.namespace,
                          key=key))

  @classmethod
  def create_from_options(cls, pipeline_options):
    """ Creates :class:`DisplayData` from a
    :class:`~apache_beam.options.pipeline_options.PipelineOptions` instance.

    When creating :class:`DisplayData`, this method will convert the value of
    any item of a non-supported type to its string representation.
    The normal :meth:`.create_from()` method rejects those items.

    Returns:
      DisplayData: A :class:`DisplayData` instance with populated items.

    Raises:
      ~exceptions.ValueError: If the **has_display_data** argument is
        not an instance of :class:`HasDisplayData`.
    """
    from apache_beam.options.pipeline_options import PipelineOptions
    if not isinstance(pipeline_options, PipelineOptions):
      raise ValueError(
          'Element of class {}.{} does not subclass PipelineOptions'
          .format(pipeline_options.__module__,
                  pipeline_options.__class__.__name__))

    items = {k: (v if DisplayDataItem._get_value_type(v) is not None
                 else str(v))
             for k, v in pipeline_options.display_data().items()}
    return cls(pipeline_options._namespace(), items)

  @classmethod