File size: 2,877 Bytes
fa1a600 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
# Copyright 2024 Big Vision Authors.
#
# Licensed 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.
"""Preprocessing builder."""
from absl import logging
from big_vision.pp import registry
import tensorflow as tf
def get_preprocess_fn(pp_pipeline, log_data=True):
"""Transform an input string into the preprocessing function.
The minilanguage is as follows:
fn1|fn2(arg, arg2,...)|...
And describes the successive application of the various `fn`s to the input,
where each function can optionally have one or more arguments, which are
either positional or key/value, as dictated by the `fn`.
The output preprocessing function expects a dictionary as input. This
dictionary should have a key "image" that corresponds to a 3D tensor
(height x width x channel).
Args:
pp_pipeline: A string describing the pre-processing pipeline. If empty or
None, no preprocessing will be executed.
log_data: Whether to log the data before and after preprocessing. Can also
be a string to show in the log for debugging, for example dataset name.
Returns:
preprocessing function.
Raises:
ValueError: if preprocessing function name is unknown
"""
names, ops = [], []
if pp_pipeline:
for op_spec in pp_pipeline.split("|"):
if not op_spec: continue # Skip empty section instead of error.
try:
ops.append(registry.Registry.lookup(f"preprocess_ops.{op_spec}")())
names.append(registry.parse_name(op_spec)[0])
except SyntaxError as err:
raise ValueError(f"Syntax error on: {op_spec}") from err
def _preprocess_fn(data):
"""The preprocessing function that is returned."""
nonlocal log_data
# Apply all the individual steps in sequence.
if log_data:
logging.info("Data before pre-processing (%s):\n%s", log_data, data)
for name, op in zip(names, ops):
with tf.name_scope(name):
data = op(data)
# Validate input
if not isinstance(data, dict):
raise ValueError("Argument `data` must be a dictionary, "
"not %s" % str(type(data)))
if log_data:
logging.info("Data after pre-processing (%s):\n%s", log_data, data)
log_data = False # For eager&pygrain: only log first one of each pipeline.
return data
return _preprocess_fn
|