<jupyter_start><jupyter_text>Image Transformation PipelineIn this notebook we're going to explore the concepts and processes for defining a simple set of operators that will perform image pre-processing before they are feed into our model(s).<jupyter_code>!pip3 install kfp --upgrade
import kfp.components as comp
import kfp.dsl as dsl
from kfp import compiler

import tensorflow as tf
import numpy as np
tf.enable_eager_execution()
import IPython.display as display
from os import listdir
from os.path import isfile, join
read_datasets_op = comp.func_to_container_op(read_records)
create_tf_datasets_op = comp.func_to_container_op(create_tf_datasets)
parse_images_op = comp.func_to_container_op(parse_datasets)
@dsl.python_component(
    name="img_preprocessing_pipline",
    description="Image preprocessing pipeline for learning to steer")

def test_nfs(server_ip: str) -> str:
    

def preprocess(dataset_path: str, num_examples: int, output_dir: str,
               batch_size: int) -> str:
    """
    Image preprocessing pipeline
    """
    import json
    import numpy as np
    import tensorflow as tf
    from tensorflow.python.lib.io import file_io
    from sklearn.utils import shuffle
    
    # TODO Move configuration related vars out to config file
    record_dir = 'tfrecords/2148-09-02-2019/tfrecords'
    
    # Create a dictionary describing the features.
    # TODO : define in config file? might be too kludgey from a UX perspective?
    image_feature_description = {
        'timestamp': tf.FixedLenFeature([], tf.float32),
        'image': tf.FixedLenFeature([], tf.string),
        'steering_theta': tf.FixedLenFeature([], tf.float32),
        'accelerator': tf.FixedLenFeature([], tf.float32),
        'height': tf.FixedLenFeature([], tf.int64),
        'width': tf.FixedLenFeature([], tf.int64),
        'capture_height': tf.FixedLenFeature([], tf.int64),
        'capture_width': tf.FixedLenFeature([], tf.int64),
        'capture_fps': tf.FixedLenFeature([], tf.int64),
        'num_channels': tf.FixedLenFeature([], tf.int64)
    }
    
    def read_records(directory: str) -> list:
        """
        Read all files in a directory and return a list containing all 
        of the filenames.
        """
        filenames = [f for f in listdir(directory) if isfile(join(directory, f))]
        return filenames

    def create_tf_datasets(filenames: list) -> list:
        raw_image_datasets = []

        for record in read_records(record_dir):
            raw_image_datasets.append(tf.data.TFRecordDataset(f'{record_dir}/{record}'))

        return raw_image_datasets

    # raw_image_datasets = []

    # for record in tfrecords:
    #     raw_image_datasets.append(tf.data.TFRecordDataset(f'tfrecords/tfrecords/{record}'))

    def parse_image_function(example_proto):
      # Parse the input tf.Example proto using the dictionary above.
        return tf.parse_single_example(example_proto, image_feature_description)

    def parse_datasets(datasets: list) -> list:
        parsed_datasets = []

        for raw_image_dataset in raw_image_datasets:
            parsed_datasets.append(r.map(_parsed_image_function()))

        return parsed_datasets
preprocess_pipeline = preprocess
pipeline_filename = preprocess_pipeline.__name__ = ".image.preprocess.tar.gz"
import kfp.compiler as compiler
compiler.Compiler().compile(preprocess_pipeline, pipeline_filename)

EXPERIMENT_NAME = "LEARN_TO_STEER"
import kfp
client = kfp.Client()
experiment = client.create_experiment(EXPERIMENT_NAME)

run_name = preprocess_pipeline.__name__ + ' run'
run_result = client.run_pipeline(experiment.id, run_name, pipeline_filename)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09296004349007883, 'fraction_numerical': 0.008426202772492526, 'mean_word_length': 3.8612945838837516, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '196123', 'license': 'MIT', 'n_tokens_mistral': 1139, 'n_tokens_neox': 1113, 'n_words': 280, 'path': 'Image Transformation Pipeline.ipynb', 'repo_name': 'edhenry/bot', 'size': '3679'}