try:
  import colab
  !pip install --upgrade pip
except:
  pass

"""
Explanation: TFX Pipeline Tutorial using Penguin dataset
Learning objectives

Prepare example data.
Create a pipeline.
Run the pipeline.

Introduction
In this notebook, you will create and run a TFX pipeline
for a simple classification model.
The pipeline will consist of three essential TFX components: ExampleGen,
Trainer and Pusher. The pipeline includes the most minimal ML workflow like
importing data, training a model and exporting the trained model.
Please see
Understanding TFX Pipelines
to learn more about various concepts in TFX.
Each learning objective will correspond to a #TODO in this student lab notebook -- try to complete this notebook first and then review the solution notebook
Set Up
You first need to install the TFX Python package and download
the dataset which you will use for our model.
Upgrade Pip
To avoid upgrading Pip in a system when running locally,
check to make sure that you are running in Colab.
Local systems can of course be upgraded separately.
End of explanation
"""


!pip install -U tfx

"""
Explanation: Install TFX
End of explanation
"""


import tensorflow as tf
print('TensorFlow version: {}'.format(tf.__version__))
from tfx import v1 as tfx
print('TFX version: {}'.format(tfx.__version__))

"""
Explanation: Note: Restart the kernel before proceeding further (On the Notebook menu, click Kernel > Restart Kernel).
Check the TensorFlow and TFX versions.
End of explanation
"""


import os

PIPELINE_NAME = "penguin-simple"

# Output directory to store artifacts generated from the pipeline.
PIPELINE_ROOT = os.path.join('pipelines', PIPELINE_NAME)
# Path to a SQLite DB file to use as an MLMD storage.
METADATA_PATH = os.path.join('metadata', PIPELINE_NAME, 'metadata.db')
# Output directory where created models from the pipeline will be exported.
SERVING_MODEL_DIR = os.path.join('serving_model', PIPELINE_NAME)

from absl import logging
logging.set_verbosity(logging.INFO)  # Set default logging level.

"""
Explanation: Set up variables
There are some variables used to define a pipeline. You can customize these
variables as you want. By default all output from the pipeline will be
generated under the current directory.
End of explanation
"""


import urllib.request
import tempfile

DATA_ROOT = tempfile.mkdtemp(prefix='tfx-data')  # Create a temporary directory.
_data_url = 'https://raw.githubusercontent.com/tensorflow/tfx/master/tfx/examples/penguin/data/labelled/penguins_processed.csv'
_data_filepath = os.path.join(DATA_ROOT, "data.csv")
urllib.request.urlretrieve(_data_url, _data_filepath)

"""
Explanation: Prepare example data
You will download the example dataset for use in our TFX pipeline. The dataset we
are using is
Palmer Penguins dataset
which is also used in other
TFX examples.
There are four numeric features in this dataset:

culmen_length_mm
culmen_depth_mm
flipper_length_mm
body_mass_g

All features were already normalized to have range [0,1]. you will build a
classification model which predicts the species of penguins.
Because TFX ExampleGen reads inputs from a directory, you need to create a
directory and copy dataset to it.
End of explanation
"""


# Review the contents of the CSV file
# TODO 1: Your code goes here



"""
Explanation: Take a quick look at the CSV file.
End of explanation
"""


_trainer_module_file = 'penguin_trainer.py'

%%writefile {_trainer_module_file}

from typing import List
from absl import logging
import tensorflow as tf
from tensorflow import keras
from tensorflow_transform.tf_metadata import schema_utils

from tfx import v1 as tfx
from tfx_bsl.public import tfxio
from tensorflow_metadata.proto.v0 import schema_pb2

_FEATURE_KEYS = [
    'culmen_length_mm', 'culmen_depth_mm', 'flipper_length_mm', 'body_mass_g'
]
_LABEL_KEY = 'species'

_TRAIN_BATCH_SIZE = 20
_EVAL_BATCH_SIZE = 10

# Since we're not generating or creating a schema, you will instead create
# a feature spec.  Since there are a fairly small number of features this is
# manageable for this dataset.
_FEATURE_SPEC = {
    **{
        feature: tf.io.FixedLenFeature(shape=[1], dtype=tf.float32)
           for feature in _FEATURE_KEYS
       },
    _LABEL_KEY: tf.io.FixedLenFeature(shape=[1], dtype=tf.int64)
}


def _input_fn(file_pattern: List[str],
              data_accessor: tfx.components.DataAccessor,
              schema: schema_pb2.Schema,
              batch_size: int = 200) -> tf.data.Dataset:
  """Generates features and label for training.

  Args:
    file_pattern: List of paths or patterns of input tfrecord files.
    data_accessor: DataAccessor for converting input to RecordBatch.
    schema: schema of the input data.
    batch_size: representing the number of consecutive elements of returned
      dataset to combine in a single batch

  Returns:
    A dataset that contains (features, indices) tuple where features is a
      dictionary of Tensors, and indices is a single Tensor of label indices.
  """
  return data_accessor.tf_dataset_factory(
      file_pattern,
      tfxio.TensorFlowDatasetOptions(
          batch_size=batch_size, label_key=_LABEL_KEY),
      schema=schema).repeat()


def _build_keras_model() -> tf.keras.Model:
  """Creates a DNN Keras model for classifying penguin data.

  Returns:
    A Keras Model.
  """
  # The model below is built with Functional API, please refer to
  # https://www.tensorflow.org/guide/keras/overview for all API options.
  inputs = [keras.layers.Input(shape=(1,), name=f) for f in _FEATURE_KEYS]
  d = keras.layers.concatenate(inputs)
  for _ in range(2):
    d = keras.layers.Dense(8, activation='relu')(d)
  outputs = keras.layers.Dense(3)(d)

  model = keras.Model(inputs=inputs, outputs=outputs)
  model.compile(
      optimizer=keras.optimizers.Adam(1e-2),
      loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
      metrics=[keras.metrics.SparseCategoricalAccuracy()])

  model.summary(print_fn=logging.info)
  return model


# TFX Trainer will call this function.
def run_fn(fn_args: tfx.components.FnArgs):
  """Train the model based on given args.

  Args:
    fn_args: Holds args used to train the model as name/value pairs.
  """

  # This schema is usually either an output of SchemaGen or a manually-curated
  # version provided by pipeline author. A schema can also derived from TFT
  # graph if a Transform component is used. In the case when either is missing,
  # `schema_from_feature_spec` could be used to generate schema from very simple
  # feature_spec, but the schema returned would be very primitive.
  schema = schema_utils.schema_from_feature_spec(_FEATURE_SPEC)

  train_dataset = _input_fn(
      fn_args.train_files,
      fn_args.data_accessor,
      schema,
      batch_size=_TRAIN_BATCH_SIZE)
  eval_dataset = _input_fn(
      fn_args.eval_files,
      fn_args.data_accessor,
      schema,
      batch_size=_EVAL_BATCH_SIZE)

  model = _build_keras_model()
  model.fit(
      train_dataset,
      steps_per_epoch=fn_args.train_steps,
      validation_data=eval_dataset,
      validation_steps=fn_args.eval_steps)

  # The result of the training should be saved in `fn_args.serving_model_dir`
  # directory.
  model.save(fn_args.serving_model_dir, save_format='tf')

"""
Explanation: You should be able to see five values. species is one of 0, 1 or 2, and all
other features should have values between 0 and 1.
Create a pipeline
TFX pipelines are defined using Python APIs. You will define a pipeline which
consists of following three components.
- CsvExampleGen: Reads in data files and convert them to TFX internal format
for further processing. There are multiple
ExampleGens for various
formats. In this tutorial, you will use CsvExampleGen which takes CSV file input.
- Trainer: Trains an ML model.
Trainer component requires a
model definition code from users. You can use TensorFlow APIs to specify how to
train a model and save it in a saved_model format.
- Pusher: Copies the trained model outside of the TFX pipeline.
Pusher component can be thought
of an deployment process of the trained ML model.
Before actually define the pipeline, you need to write a model code for the
Trainer component first.
Write model training code
You will create a simple DNN model for classification using TensorFlow Keras
API. This model training code will be saved to a separate file.
In this tutorial you will use
Generic Trainer
of TFX which support Keras-based models. You need to write a Python file
containing run_fn function, which is the entrypoint for the Trainer
component.
End of explanation
"""


def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     module_file: str, serving_model_dir: str,
                     metadata_path: str) -> tfx.dsl.Pipeline:
  """Creates a three component penguin pipeline with TFX."""
  # Brings data into the pipeline.
  example_gen = tfx.components.CsvExampleGen(input_base=data_root)

  # Uses user-provided Python function that trains a model.
  trainer = tfx.components.Trainer(
      module_file=module_file,
      examples=example_gen.outputs['examples'],
      train_args=tfx.proto.TrainArgs(num_steps=100),
      eval_args=tfx.proto.EvalArgs(num_steps=5))

  # Pushes the model to a filesystem destination.
  pusher = # TODO 2: Your code goes here

  # Following three components will be included in the pipeline.
  components = [
      example_gen,
      trainer,
      pusher,
  ]

  return tfx.dsl.Pipeline(
      pipeline_name=pipeline_name,
      pipeline_root=pipeline_root,
      metadata_connection_config=tfx.orchestration.metadata
      .sqlite_metadata_connection_config(metadata_path),
      components=components)

"""
Explanation: Now you have completed all preparation steps to build a TFX pipeline.
Write a pipeline definition
You define a function to create a TFX pipeline. A Pipeline object
represents a TFX pipeline which can be run using one of pipeline
orchestration systems that TFX supports.
End of explanation
"""


tfx.orchestration.LocalDagRunner().run(
  _create_pipeline(
      pipeline_name=PIPELINE_NAME,
      pipeline_root=PIPELINE_ROOT,
      data_root=DATA_ROOT,
      module_file=_trainer_module_file,
      serving_model_dir=SERVING_MODEL_DIR,
      metadata_path=METADATA_PATH))

"""
Explanation: Run the pipeline
TFX supports multiple orchestrators to run pipelines.
In this tutorial you will use LocalDagRunner which is included in the TFX
Python package and runs pipelines on local environment.
You often call TFX pipelines "DAGs" which stands for directed acyclic graph.
LocalDagRunner provides fast iterations for developemnt and debugging.
TFX also supports other orchestrators including Kubeflow Pipelines and Apache
Airflow which are suitable for production use cases.
See
TFX on Cloud AI Platform Pipelines
or
TFX Airflow Tutorial
to learn more about other orchestration systems.
Now you create a LocalDagRunner and pass a Pipeline object created from the
function you already defined.
The pipeline runs directly and you can see logs for the progress of the pipeline including ML model training.
End of explanation
"""


# List files in created model directory.
# TODO 3: Your code goes here


"""
Explanation: You should see "INFO:absl:Component Pusher is finished." at the end of the
logs if the pipeline finished successfully. Because Pusher component is the
last component of the pipeline.
The pusher component pushes the trained model to the SERVING_MODEL_DIR which
is the serving_model/penguin-simple directory if you did not change the
variables in the previous steps. You can see the result from the file browser
in the left-side panel in Colab, or using the following command:
End of explanation
"""


