{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"Thanks for the public notebooks below:  \nhttps://www.kaggle.com/code/hoyso48/1st-place-solution-training  \nhttps://www.kaggle.com/code/irohith/aslfr-ctc-based-on-prev-comp-1st-place  \nhttps://www.kaggle.com/code/markwijkhuizen/aslfr-transformer-training-inference  \nThis is the 3rd place solution training code, you could refer the solution here:  \nhttps://www.kaggle.com/competitions/asl-fingerspelling/discussion/434393  ","metadata":{}},{"cell_type":"markdown","source":"# Install libs","metadata":{}},{"cell_type":"code","source":"try:\n  from icecream import ic\n  import pymp\nexcept Exception:\n  !pip install -q icecream --no-index --find-links=file:///kaggle/input/icecream\n  !pip install -q pymp-pypi --no-index --find-links=file:///kaggle/input/pymp-pypi/pymp-pypi-0.4.5/dist","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:42:17.499683Z","iopub.execute_input":"2023-08-27T00:42:17.500802Z","iopub.status.idle":"2023-08-27T00:42:50.058436Z","shell.execute_reply.started":"2023-08-27T00:42:17.500764Z","shell.execute_reply":"2023-08-27T00:42:50.056931Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Import libs","metadata":{}},{"cell_type":"code","source":"import sys, os\nimport numpy as np\nimport pandas as pd\nimport json\nimport re\nimport six\nimport glob\nimport traceback\nimport inspect\nfrom typing import Union\nfrom collections import Counter, OrderedDict, defaultdict\nfrom collections.abc import Iterable\nfrom multiprocessing import cpu_count\nfrom tqdm.notebook import tqdm\nfrom icecream import ic\nimport pymp\nimport transformers\nimport tensorflow as tf\nimport torch\nfrom torch import nn, einsum\nimport torch.nn.functional as F\nic(tf.__version__, torch.__version__)","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:42:50.061606Z","iopub.execute_input":"2023-08-27T00:42:50.062009Z","iopub.status.idle":"2023-08-27T00:43:04.855065Z","shell.execute_reply.started":"2023-08-27T00:42:50.061974Z","shell.execute_reply":"2023-08-27T00:43:04.853466Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Flags","metadata":{}},{"cell_type":"code","source":"class FLAGS(object):\n  # for tfrecords args, you could ignore\n  seed = 1024\n  batch_parse = False\n  sparse_to_dense = True\n  eval_keys = []\n  incl_keys = []\n  excl_keys = []\n  recount_tfrecords = False  \n  batch_sizes = []\n  buffer_size = 1024\n  buckets = None\n  drop_remainder = None\n  shard_by_files = True\n  shuffle_batch = None\n  shuffle_files = None\n  num_dataset_threads = 0\n  num_prefetch_batches = 1024\n  repeat_then_shuffle = False\n  length_index = 1\n  length_key = None\n  dynamic_pad = True\n  cache = False\n  cache_after_map = False\n  fixed_random = False\n  parallel_read_files = True\n  padding_idx = 0\n  dataset_keys = []\n  dataset_excl_keys = []\n  exclude_varlen_keys = False\n  prefetch = None\n  dataset_ordered = False\n    \n  torch = True\n  keras = False\n    \n  # online==False means using n-fold split and train on fold 1,2, folds-1 while valid on fold 0\n  # online==True means using all train data but still will valid on fold 0\n  online = False  \n  folds = 4\n  fold = 0\n  fold_seed = 1229\n  root = '../input/asl-fingerspelling'\n  working = '/kaggle/working'\n  use_z = True  # use x,y,z if True\n  norm_frames = True # norm frames using x - mean / std\n  concat_frames = True # concat original and normalized frames\n  add_pos = True # add abs frame pos, like 1/1000., 2/1000.\n  sup_weight = 0.1 # for supplement dataset assigin weight 0.1\n  \n  train_files = []\n  valid_files = []\n      \n  mix_sup = True # train & sup dataset\n  vie = 5 # valid interval epochs \n  lr = 2e-3\n  epochs = 400 \n  batch_size = 128\n  eval_batch_size = 256\n  awp = True\n  adv_start_epoch = None\n  adv_lr = 0.2\n  adv_eps = 0\n  fp16 = False # notice fp16 could not be set True if using awp here, otherwise nan\n  optimizer = 'Adam'\n  opt_eps = 1e-6 \n  scheduler = 'linear'\n  # for model related configs\n  encoder_layers = 17\n  encoder_units = 200 \n  n_frames = 320  \n  distributed = False\n    \ndef load_json(filename):\n  with open(filename) as fh:\n    obj = json.load(fh)\n  return obj","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:43:04.857596Z","iopub.execute_input":"2023-08-27T00:43:04.859467Z","iopub.status.idle":"2023-08-27T00:43:04.879828Z","shell.execute_reply.started":"2023-08-27T00:43:04.859401Z","shell.execute_reply":"2023-08-27T00:43:04.878151Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Common configs","metadata":{}},{"cell_type":"code","source":"LPOSE = [13, 15, 17, 19, 21]\nRPOSE = [14, 16, 18, 20, 22]\nPOSE = LPOSE + RPOSE\n\nLIP = [\n    61, 185, 40, 39, 37, 0, 267, 269, 270, 409,\n    291, 146, 91, 181, 84, 17, 314, 405, 321, 375,\n    78, 191, 80, 81, 82, 13, 312, 311, 310, 415,\n    95, 88, 178, 87, 14, 317, 402, 318, 324, 308,\n]\nic(len(LIP))\nLLIP = [84,181,91,146,61,185,40,39,37,87,178,88,95,78,191,80,81,82]\nRLIP = [314,405,321,375,291,409,270,269,267,317,402,318,324,308,415,310,311,312]\nMID_LIP = [i for i in LIP if i not in LLIP + RLIP]\nic(len(LLIP), len(RLIP), len(MID_LIP))\n\nNOSE=[\n    1,2,98,327\n]\nLNOSE = [98]\nRNOSE = [327]\nMID_NOSE = [i for i in NOSE if i not in LNOSE + RNOSE]\n\nLEYE = [\n    263, 249, 390, 373, 374, 380, 381, 382, 362,\n    466, 388, 387, 386, 385, 384, 398,\n]\nREYE = [\n    33, 7, 163, 144, 145, 153, 154, 155, 133,\n    246, 161, 160, 159, 158, 157, 173,\n]\n\nN_HAND_POINTS = 21\nN_POSE_POINTS = len(LPOSE)\nN_LIP_POINTS = len(LLIP)\nN_EYE_POINTS = len(LEYE)\nN_NOSE_POINTS = len(LNOSE)\nN_MID_POINTS = len(MID_LIP + MID_NOSE)\n\nSEL_COLS = []\nfor i in range(N_HAND_POINTS):\n  SEL_COLS.extend([f'x_left_hand_{i}', f'y_left_hand_{i}', f'z_left_hand_{i}'])\nfor i in range(N_HAND_POINTS):\n  SEL_COLS.extend([f'x_right_hand_{i}', f'y_right_hand_{i}', f'z_right_hand_{i}'])\nfor i in LPOSE:\n  SEL_COLS.extend([f'x_pose_{i}', f'y_pose_{i}', f'z_pose_{i}'])\nfor i in RPOSE:\n  SEL_COLS.extend([f'x_pose_{i}', f'y_pose_{i}', f'z_pose_{i}'])\nfor i in LLIP:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\nfor i in RLIP:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\n\nfor i in LEYE:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\nfor i in REYE:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\n  \nfor i in LNOSE:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\nfor i in RNOSE:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\n  \nfor i in MID_LIP:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\nfor i in MID_NOSE:\n  SEL_COLS.extend([f'x_face_{i}', f'y_face_{i}', f'z_face_{i}'])\n    \nN_COLS = len(SEL_COLS)\nic(N_COLS)\n    \nCHAR2IDX = load_json(f'../input/asl-fingerspelling/character_to_prediction_index.json')\nCHAR2IDX = {k: v + 1 for k, v in CHAR2IDX.items()}\nN_CHARS = len(CHAR2IDX)\nic(N_CHARS)\n\nPAD_IDX = 0\nSOS_IDX = PAD_IDX # Start Of Sentence\nEOS_IDX = N_CHARS + 1 # End Of Sentence\nic(PAD_IDX, SOS_IDX, EOS_IDX)\n\nPAD_TOKEN = '<PAD>'\nSOS_TOKEN = PAD_TOKEN\nEOS_TOKEN = '<EOS>'\n\nCHAR2IDX[PAD_TOKEN] = PAD_IDX\nCHAR2IDX[EOS_TOKEN] = EOS_IDX \n\nADDRESS_TOKEN = '<ADDRESS>'\nURL_TOKEN = '<URL>'\nPHONE_TOKEN = '<PHONE>'\nSUP_TOKEN = '<SUP>'\n\nVOCAB_SIZE = len(CHAR2IDX)\nIDX2CHAR = {v: k for k, v in CHAR2IDX.items()}\nic(VOCAB_SIZE)\nic(len(IDX2CHAR))\n\nSTATS = {}\nCLASSES = [\n  'address', \n  'url', \n  'phone', \n  'sup',\n  ]\nPHRASE_TYPES = dict(zip(CLASSES, range(len(CLASSES))))\nN_TYPES = len(CLASSES)\nMAX_PHRASE_LEN = 32\n\ndef get_vocab_size():\n  vocab_size = VOCAB_SIZE\n  return vocab_size\n\ndef get_n_cols(no_motion=False, use_z=None):\n  n_cols = N_COLS\n  if use_z is None:\n    use_z = FLAGS.use_z\n  \n  if FLAGS.concat_frames:\n    assert FLAGS.norm_frames\n    n_cols += N_COLS\n  \n  if not use_z:\n    n_cols = n_cols // 3 * 2\n    \n  if FLAGS.add_pos:\n    n_cols += 1\n  \n  return n_cols","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:43:04.882157Z","iopub.execute_input":"2023-08-27T00:43:04.882655Z","iopub.status.idle":"2023-08-27T00:43:05.397105Z","shell.execute_reply.started":"2023-08-27T00:43:04.88262Z","shell.execute_reply":"2023-08-27T00:43:05.396484Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Tfrecord dataset","metadata":{}},{"cell_type":"code","source":"def gen_inputs(files, \n           decode_fn, \n           batch_size=64,\n           post_decode_fn=None,\n           num_epochs = None, \n           num_threads=None, \n           buffer_size = 15000, #change from 1000 to 15000\n           dynamic_pad=True,\n           shuffle=True,\n           shuffle_batch=None,\n           shuffle_files=None,\n           ordered=None,\n           min_after_dequeue=None, #depreciated\n           seed=None, \n           enqueue_many=False,  #depreciated\n           fixed_random=False, \n           drop_remainder=False, \n           num_prefetch_batches=None, \n           bucket_boundaries=None,\n           length_index=None,\n           length_key=None,\n           length_fn=None,\n           bucket_batch_sizes=None,\n           repeat=True,\n           initializable=False,\n           filter_fn=None,\n           balance_pos_neg=False,\n           pos_filter_fn=None,\n           neg_filter_fn=None,\n           count_fn=None,\n           return_iterator=False,\n           Dataset=None,\n           batch_parse=False, #by default will be line parse\n           hvd_shard=True,\n           shard_by_files=False,\n           training=False,\n           simple_parse=False,\n           repeat_then_shuffle=False,\n           cache=False,\n           cache_file='',\n           cache_after_map=False,\n           device=None,\n           world_size=1,\n           rank=0,\n           parallel_read_files=False,\n           use_feed_dict=False,\n           feed_name=None,\n           padding_values=None,\n           distribute_strategy=None,\n           torch=False,\n           keras=False,\n           subset=None,\n           return_numpy=False,\n           name='input'):\n  Dataset = Dataset or tf.data.TFRecordDataset\n  AUTO = tf.data.AUTOTUNE\n  use_horovod = False\n\n  def shard(d):\n    return d.shard(hvd.size(), hvd.rank())\n\n  # Choose to use cpu outside input function like in dataset.py\n  #with tf.device('/cpu:0'):\n  if isinstance(files, str):\n    files = gezi.list_files(files)\n  assert len(files) > 0\n\n  if not num_threads:\n    num_threads = 8\n\n  if 'batch_size' in inspect.getfullargspec(decode_fn).args:\n    decode_fn_ = decode_fn\n    def decode_function(example):\n      return decode_fn_(example, batch_size)\n    decode_fn = decode_function\n    \n  if not num_epochs: \n    num_epochs = None\n\n  if shuffle:\n    if shuffle_files is None:\n      shuffle_files = True\n    if shuffle_batch is None:\n      shuffle_batch = True\n  else:\n    if shuffle_files is None:\n      shuffle_files = False\n    if shuffle_batch is None:\n      shuffle_batch = False\n    # TDO 并行读取就会打乱顺序？\n    if not shuffle_files:\n      parallel_read_files = False\n\n  if fixed_random:\n    if seed is None:\n      seed = 1024\n  else:\n    pass\n\n  num_files = len(files)\n  if use_feed_dict and feed_name:\n    files = tf.compat.v1.placeholder(tf.string, [None], feed_name)\n    gezi.set_global(feed_name, files)\n\n  if not num_prefetch_batches:\n    #num_prefetch_batches = num_threads + 3\n    if buffer_size:\n      num_prefetch_batches = int(buffer_size / batch_size)\n    # else:\n    #   num_prefetch_batches = 100\n  \n  if not buffer_size and num_prefetch_batches:\n    buffer_size = num_prefetch_batches * batch_size\n    \n  options = tf.data.Options()\n  try:\n    options.threading.private_threadpool_size = num_threads\n    options.threading.max_intra_op_parallelism = 1\n  except Exception:\n    options.experimental_threading.private_threadpool_size = num_threads\n    options.experimental_threading.max_intra_op_parallelism = 1\n\n  options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.DATA\n  options.experimental_deterministic = True\n\n  if shuffle and not fixed_random:\n    options.experimental_deterministic = False\n\n  if not ordered:\n    options.experimental_deterministic = False\n\n  if not parallel_read_files or num_files == 1:\n    d = Dataset(files)\n    d = d.with_options(options)\n    if use_horovod and hvd_shard:\n      d = shard(d)\n    if not use_horovod and world_size > 1:\n      d = d.shard(world_size, rank)\n  else:\n    try:\n      if shffle_files and (use_horovod or world_size > 1):\n        assert seed\n      d = tf.data.Dataset.list_files(files, shuffle=shuffle_files, seed=seed)\n      d = d.with_options(options)\n    except Exception:\n      d = tf.data.Dataset.from_tensor_slices(files)\n      d = d.with_options(options)\n    # here shard by files, not work good, especially for text line dataset with horovod\n    if use_horovod and shard_by_files:\n      d = shard(d)\n    elif world_size > 1 and shard_by_files:\n      d = d.shard(world_size, rank)\n\n    d = d.interleave(Dataset,\n                  #  cycle_length=min(len(files), 1000),  # in tf 1.14 must set and can not set as AUTOTUNE for tf 2.1 with default as AUTOTUNE\n                  block_length=1,\n                  num_parallel_calls=AUTO)\n\n    if world_size > 1 and not shard_by_files:\n      d = d.shard(world_size, rank)\n\n  if repeat and repeat_then_shuffle:\n    d = d.repeat(num_epochs)\n\n  if cache and (not FLAGS.cache_after_map):\n    d = d.cache(cache_file)\n    \n  # must batch then map if use pyfunc which you might use batch_parse, here batch_parse means batch parse otherwise slower but simple and powerfull...\n  if not batch_parse:\n    d = d.map(decode_fn, num_parallel_calls=AUTO)\n    if cache and cache_after_map:\n      d = d.cache(cache_file)\n  \n  if filter_fn is not None and not batch_parse:\n    d = d.filter(filter_fn)\n\n  if shuffle_batch:\n    d = d.shuffle(buffer_size=buffer_size, seed=seed, reshuffle_each_iteration=True)\n\n  # shuffle then repeat\n  if repeat and not repeat_then_shuffle:\n    d = d.repeat(num_epochs)\n  \n  if dynamic_pad:\n    if not batch_parse: \n      d = d.padded_batch(batch_size, drop_remainder=drop_remainder)\n    else:\n      d = d.batch(batch_size, drop_remainder=drop_remainder)\n  else:\n    d = d.batch(batch_size, drop_remainder=drop_remainder)\n\n  if batch_parse:\n    d = d.map(decode_fn, num_parallel_calls=AUTO)\n    if filter_fn is not None:\n      try:\n        d = d.unbatch()\n        d = d.filter(filter_fn)\n      except Exception:\n        d = d.unbatch()\n\n      d = d.batch(batch_size, drop_remainder=drop_remainder)\n\n  if post_decode_fn is not None:\n    d = d.map(post_decode_fn, num_parallel_calls=AUTO)\n\n  if cache and FLAGS.cache_after_map:\n    logging.debug('Cache datase after map')\n    d = d.cache(cache_file)\n\n  d = d.prefetch(FLAGS.prefetch or AUTO)\n\n  if not return_numpy:    \n    return d\n  else:\n    return d.as_numpy_iterator()\n","metadata":{"_kg_hide-input":true,"_kg_hide-output":false,"execution":{"iopub.status.busy":"2023-08-27T00:43:05.402394Z","iopub.execute_input":"2023-08-27T00:43:05.402836Z","iopub.status.idle":"2023-08-27T00:43:05.438275Z","shell.execute_reply.started":"2023-08-27T00:43:05.402803Z","shell.execute_reply":"2023-08-27T00:43:05.437042Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def decode_example(x):\n  if tf.executing_eagerly():\n    x = x.numpy()\n  x = tf.train.Example.FromString(x).features.feature\n  features = {}\n  for key in x.keys():\n    typenames = ['bytes_list', 'float_list', 'int64_list']\n    dtypes = [object, np.float32, np.int64]\n    for typename, dtype in zip(typenames, dtypes):\n      value = getattr(x[key], typename).value\n      if value:\n        features[key] = np.array(value, dtype=dtype)\n  return features\n\ndef first_example(record_file):\n  if isinstance(record_file, (list, tuple)):\n    record_file = record_file[0]\n  if tf.executing_eagerly():\n    for item in tf.data.TFRecordDataset(record_file):\n      x = decode_example(item)\n      return x\n  else:\n    for item in tf.compat.v1.python_io.tf_record_iterator(record_file):\n      x = decode_example(item)\n      return x\n\ndef npdtype2tfdtype(dtype, large=False):\n  if dtype == np.float32:\n    return tf.float32\n  if dtype == np.int32:\n    if not large:\n      return tf.int32\n    else:\n      return tf.int64\n  if dtype == np.int64:\n    return tf.int64\n  if dtype == np.float64:\n    return tf.float32\n  return tf.string\n\ndef sparse_tensor_to_dense(input_tensor, default_value=0):  \n  return tf.sparse.to_dense(input_tensor, default_value=default_value, validate_indices=False)\n\ndef sparse2dense(features, key=None, default_value=0):\n\n  def sparse2dense_(features, key, default_value):\n    val = features[key]\n    if val.values.dtype == tf.string:\n      default_value = None\n    val = sparse_tensor_to_dense(val, default_value)\n    features[key] = val\n\n  modified = False\n  if key:\n    sparse2dense_(features, key)\n    modified = True\n  else:\n    from tensorflow.python.framework.sparse_tensor import SparseTensor\n    for key, val in features.items():\n      if isinstance(val, SparseTensor):\n        sparse2dense_(features, key, default_value)\n        modified = True\n  return modified\n\ndef get_num_records_single(tf_record_file, recount=False):\n  if not recount:\n    filename = os.path.basename(tf_record_file)\n    filename = filename.replace('-', '.').replace('_', '.')\n    l = filename.split('.')\n\n    for item in reversed(l):\n      if item.isdigit():\n        return int(item)\n\n  # try:\n  return sum(\n      1 for _ in tf.compat.v1.python_io.tf_record_iterator(tf_record_file))\n  # except Exception:\n  #   return 0\n\n\ndef get_num_records(files, recount=False):\n  if isinstance(files, str):\n    files = gezi.list_files(files)\n  res = sum([\n      get_num_records_single(file, recount=recount)\n      for file in tqdm(files, ascii=False, desc='get_num_records', leave=False)\n  ])\n  return res\n","metadata":{"_kg_hide-input":true,"execution":{"iopub.status.busy":"2023-08-27T00:43:05.439955Z","iopub.execute_input":"2023-08-27T00:43:05.44028Z","iopub.status.idle":"2023-08-27T00:43:05.461294Z","shell.execute_reply.started":"2023-08-27T00:43:05.440253Z","shell.execute_reply":"2023-08-27T00:43:05.460302Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# A wrapper base class for tfrecords related dataset \nclass TfrecordsDataset(object):\n  def __init__(self, \n               subset='valid',\n               batch_size=None,\n               Type=None, \n               files=None,\n               num_instances=None,\n               batch_parse=None,\n               sparse_to_dense=None,\n               hvd_shard=True,\n               use_int32=True,\n               is_info=False,\n               eval_keys=[],\n               incl_keys=[],\n               excl_keys=[],\n               str_keys=[],\n               varlen_keys=[],\n               use_tpu=False,\n               recount=None):\n    self.subset = subset\n    self.filter_fn = None\n    self.pos_filter_fn = None\n    self.neg_filter_fn = None \n    self.count_fn = None\n    self.Type = Type\n    self.batch_parse = batch_parse if batch_parse is not None else FLAGS.batch_parse\n    self.sparse_to_dense = sparse_to_dense if sparse_to_dense is not None else FLAGS.sparse_to_dense\n    self.use_post_decode = None\n    # if self.batch_parse:\n    #   self.sparse_to_dense = False\n    self.batch_size = batch_size or FLAGS.batch_size\n    self.hvd_shard = hvd_shard\n    self.indexes = {'train': -1, 'valid': -1, 'test': -1}\n    self.is_info = is_info\n    self.eval_keys = eval_keys or FLAGS.eval_keys\n    if subset == 'test':\n      self.eval_keys = gezi.get('test_keys') or self.eval_keys\n    self.show_keys = set()  # 如果用户不指定eval_keys 可以用self.show_keys所有非变成以及长度为0,1的key 前提需要使用.adds不能自己外部定义\n    self.excl_keys = excl_keys or FLAGS.excl_keys\n    self.incl_keys = incl_keys or FLAGS.incl_keys\n    self.str_keys = str_keys\n    self.varlen_keys = varlen_keys\n\n    self.parse_fn = tf.io.parse_single_example if not self.batch_parse else tf.io.parse_example\n\n    self.features_dict = {}\n    self.has_varlen_feats = False\n    self.use_tpu = use_tpu\n    try:\n      # TPU detection. No parameters necessary if TPU_NAME environment variable is\n      # set: this is always the case on Kaggle.\n      tpu = tf.distribute.cluster_resolver.TPUClusterResolver()\n      # print('Running on TPU ', tpu.master())\n    except ValueError:\n      tpu = None\n    if tpu is not None:\n      self.use_tpu = True\n    self.use_int32 = use_int32\n    if self.use_tpu:\n      self.use_int32 = True\n\n    self.num_instances_ = num_instances\n    self.files_ = files\n\n    # self.use_post_decode = use_post_decode\n    self.recount = recount or FLAGS.recount_tfrecords\n\n    assert self.subset in ['train', 'valid', 'test'], \\\n          'subset is {} but should in [train, valid, test]'.format(self.subset)\n\n  @staticmethod\n  def get_filenames_(subset=None, shuffle=False):\n    try:\n      if subset in ['train', 'valid', 'test']:\n        if subset == 'train':\n          return FLAGS.train_files\n        elif subset == 'valid':\n          return FLAGS.valid_files\n      else:\n        raise ValueError('Invalid data subset \"%s\"' % subset)\n    except Exception:\n      return None\n\n  def get_filenames(self, subset=None, shuffle=False):\n    subset = subset or self.subset\n    return TfrecordsDataset.get_filenames_(subset, shuffle=False)\n\n  def basic_parse(self, example):\n    self.auto_parse(keys=self.incl_keys, exclude_keys=self.excl_keys)\n    if self.varlen_keys:\n      self.adds_varlens(self.varlen_keys)\n    fe = self.parse_(serialized=example)\n    return fe\n  \n  # override this\n  def parse(self, example):\n    return self.basic_parse(example)\n\n  def decode(self, example):\n    l = self.parse(example)\n    \n    if isinstance(l, (list, tuple)):\n      features = l[0]\n    else:\n      features = l\n    # self.use_tpu = True\n    if isinstance(features, dict):\n      if self.use_tpu:\n        def decode_label(label):\n          label = tf.io.decode_raw(label, tf.uint8)  # tf.string -> [tf.uint8]\n          label = tf.reshape(label, [])  # label is a scalar\n          return tf.cast(label, tf.int32) \n        for key in features.keys():\n          if features[key].dtype in [tf.int64, tf.uint8, tf.uint16, tf.uint32]:\n            features[key] = tf.cast(features[key], tf.int32)\n\n        if not self.is_info:\n          keys = list(features.keys())\n          for key in keys:\n            if features[key].dtype ==tf.string:\n              del features[key]\n\n              if key in self.eval_keys:\n                FLAGS.use_info_dataset = True  # 因为训练model的dataset不再包含eval的某个信息 需要依赖再遍历一遍info_dataset\n              # features[key] = tf.ones_like(features[key], tf.int32)\n              # features[key] = decode_label(features[key]) ## not work TODO\n\n      else:\n        def _cast_dict(features):\n          for key in features:\n            if isinstance(features[key], dict):\n              _cast_dict(features[key])\n            else:\n              # tf.print(key, features[key])\n              if features[key].dtype == tf.int64 and self.use_int32:\n                features[key] = tf.cast(features[key], tf.int32)\n        _cast_dict(features)\n \n      # is_info 只在tf2 keras模式下生效, 都创建 但是有可能不用 只有 FLAGS.use_info_dataset = True 才使用\n      if self.is_info:\n        keys = list(features.keys())\n        if not FLAGS.predict_on_batch:\n          if not self.eval_keys:\n            for key in keys:\n              dim = 1 if self.batch_parse else 0\n              if not (len(features[key].shape) == dim or features[key].shape[dim] == 1):\n                del features[key]\n              else:\n                self.show_keys.add(key)\n          else:\n            for key in keys:\n              if key not in self.eval_keys:\n                del features[key]\n      else:\n        keys = list(features.keys())\n        for key in keys:\n          if key in self.excl_keys:\n            del features[key]\n\n    return l\n\n  def adjust(self, result):\n    return result\n\n  def parse_(self, serialized, features=None):\n    features = features or self.features_dict\n    # ic(features)\n    features = self.parse_fn(serialized=serialized, features=features)\n    if FLAGS.exclude_varlen_keys:\n      from tensorflow.python.framework.sparse_tensor import SparseTensor\n      sparse_keys = [key for key in features if isinstance(key, SparseTensor)]\n      for key in sparse_keys:\n        del features[key]\n    else:\n      if self.sparse_to_dense:\n        modified = sparse2dense(features, default_value=FLAGS.padding_idx)\n        self.has_varlen_feats = modified\n    self.features = features\n    return features\n  \n  def gen_example(self, files=None):\n    if not files:\n      files = self.get_filenames()\n    if not isinstance(files, (list, tuple)):\n      files = [files]\n    example = {}\n    if files:\n      for file in files:\n        try:\n          example = first_example(file)\n        except Exception:\n          ic(traceback.format_exc())\n          ic('bad tfrecord:', file)\n        if example:\n          self.example = example\n          break\n    self.example = example\n    return example\n\n  def gen_input(self, files=None):\n    example = self.gen_example().copy()\n    for key in example:\n      example[key] = np.asarray([example[key]])\n    return example\n\n  def first_input(self, files=None):\n    return self.gen_input(files)\n\n  def add(self, key, dtype=None, length=None, features_dict=None):\n    features_dict = features_dict or self.features_dict\n    dtype_ = dtype\n    if key in self.example:\n      dtype = dtype_ or self.example[key].dtype \n      if length is None:\n        features_dict[key] = tf.io.VarLenFeature(dtype)\n      elif length > 0:\n        features_dict[key] = tf.io.FixedLenFeature([length], dtype)\n      else:\n        features_dict[key] = tf.io.FixedLenFeature([], dtype)\n    \n  def adds(self, keys, dtype=None, length=None, features_dict=None):\n    features_dict = features_dict or self.features_dict\n    dtype_ = dtype\n    for key in keys:\n      if key in self.example:\n        dtype = dtype_ or self.example[key].dtype \n        if length is None:\n          features_dict[key] = tf.io.VarLenFeature(dtype)\n        elif length > 0:\n          features_dict[key] = tf.io.FixedLenFeature([length], dtype)\n        else:\n          features_dict[key] = tf.io.FixedLenFeature([], dtype)\n\n  def auto_parse(self, keys=[], exclude_keys=[], features_dict=None):\n    keys = keys or FLAGS.dataset_keys or self.example.keys()\n    exclude_keys = exclude_keys or FLAGS.dataset_excl_keys\n    keys = [key for key in keys if key not in exclude_keys]\n\n    for key in keys:\n      if key not in self.example:\n        continue\n      length = self.example[key].shape[0]\n      \n      if length == 1:\n        # just to (bs,), tf keras will auto change to (bs,1), also for string 0 is ok\n        length = 0 \n\n      dtype = npdtype2tfdtype(self.example[key].dtype)\n      # print(key, dtype, length, self.example[key])\n      self.adds([key], dtype, length, features_dict)\n\n  def adds_varlens(self, keys=[], exclude_keys=[], features_dict=None):\n    keys = keys or self.example.keys()\n    keys = [key for key in keys if key not in exclude_keys]\n\n    for key in keys:\n      if not key in self.example:\n        continue\n      length = self.example[key].shape[0]\n      dtype = npdtype2tfdtype(self.example[key].dtype)\n      length = None\n      if dtype == tf.string:\n        length = 1\n      self.adds([key], dtype, length, features_dict)  \n  \n  def make_batch(self, \n                 batch_size=None, \n                 filenames=None,\n                 subset=None,\n                 initializable=False,\n                 repeat=None,\n                 shuffle=None,\n                 return_iterator=True,\n                 hvd_shard=None,\n                 simple_parse=False,\n                 num_epochs=None,\n                 cache=False,\n                 cache_file='',\n                 buffer_size=None,\n                 batch_sizes=None,\n                 buckets=None,\n                 drop_remainder=None,\n                 world_size=1,\n                 rank=0,\n                 shard_by_files=None,\n                 distribute_strategy=None,\n                 return_numpy=False):\n    # with tf.device('/cpu:0'):\n    subset = subset or self.subset\n    hvd_shard = hvd_shard if hvd_shard is not None else self.hvd_shard\n    if batch_size is None:\n      is_test = True\n    else:\n      is_test = False\n    batch_size = batch_size or self.batch_size\n    self.batch_size = batch_size\n    batch_sizes = batch_sizes if batch_sizes is not None else FLAGS.batch_sizes\n    buffer_size = buffer_size if buffer_size is not None else FLAGS.buffer_size\n    buckets = buckets if buckets is not None else FLAGS.buckets\n    drop_remainder = drop_remainder if drop_remainder is not None else FLAGS.drop_remainder\n    shard_by_files = shard_by_files if shard_by_files is not None else FLAGS.shard_by_files\n\n    self.return_numpy = return_numpy\n\n    filenames = filenames or self.files_ or self.get_filenames(subset)\n    \n    self.gen_example(filenames)\n\n    is_eager = tf.executing_eagerly()\n\n    self.files_ = filenames\n\n    self.indexes[self.subset] += 1\n    \n    if repeat is None:\n      num_gpus = 1\n      # if subset == 'train' or num_gpus > 1:\n      if subset == 'train':\n        repeat = True\n      else:\n        repeat = False\n      if is_eager and num_gpus == 1 and tf.__version__ < '2':\n        # let tf eager similary to pytorch\n        repeat = False\n\n    if shuffle is None:\n      if subset == 'train':\n        shuffle = FLAGS.shuffle \n      else:\n        shuffle = FLAGS.shuffle_valid \n\n    if drop_remainder is None:\n      if subset == 'train':\n        drop_remainder = True\n      else:\n        drop_remainder = False\n\n    balance_pos_neg=False\n    ic(self.subset, repeat, drop_remainder)\n\n    seed = FLAGS.seed \n    if seed is not None:\n      FLAGS.seed += 1\n\n    ## put on cpu or dummy\n    with tf.device('/cpu'):\n      result = gen_inputs(\n        filenames, \n        decode_fn=self.decode,\n        batch_size=batch_size,\n        post_decode_fn=self.post_decode if hasattr(self, 'post_decode') and self.use_post_decode != False else None,\n        shuffle=shuffle,\n        shuffle_batch=FLAGS.shuffle_batch,\n        shuffle_files=FLAGS.shuffle_files,\n        ordered=FLAGS.dataset_ordered if subset == 'train' else True,\n        num_threads=FLAGS.num_dataset_threads,\n        buffer_size=buffer_size,\n        num_prefetch_batches=FLAGS.num_prefetch_batches,\n        initializable=initializable,\n        repeat=repeat,\n        repeat_then_shuffle=FLAGS.repeat_then_shuffle,\n        drop_remainder=drop_remainder,\n        bucket_boundaries=buckets,\n        bucket_batch_sizes=batch_sizes,\n        length_index=FLAGS.length_index,\n        length_key=FLAGS.length_key,\n        seed=seed,\n        return_iterator=return_iterator,\n        filter_fn=self.filter_fn,  # inside filter_fn judge subset train or valid or test\n        balance_pos_neg=balance_pos_neg,\n        pos_filter_fn=self.pos_filter_fn if subset == 'train' else None,\n        neg_filter_fn=self.neg_filter_fn if subset == 'train' else None,\n        count_fn=self.count_fn if subset == 'train' else None,\n        name=subset,\n        Dataset=self.Type,\n        batch_parse=self.batch_parse,\n        hvd_shard=hvd_shard,\n        shard_by_files=shard_by_files,\n        training=subset == 'train',\n        simple_parse=simple_parse,\n        num_epochs=num_epochs,\n        dynamic_pad=FLAGS.dynamic_pad, #如果有varlen feats才需要 padded_batch 同时batch_parse模式其实也不需要因为sparse2dense就可以自动padd\n        cache=cache,\n        cache_file=cache_file,\n        cache_after_map=FLAGS.cache_after_map,\n        device='/gpu:0',\n        world_size=world_size,\n        rank=rank,\n        fixed_random=FLAGS.fixed_random,\n        parallel_read_files=FLAGS.parallel_read_files,\n        #use_feed_dict=FLAGS.train_loop and FLAGS.rounds > 1 and not is_eager and FLAGS.feed_dataset and tf.__version__ < '2',\n        feed_name=f'{self.subset}_{self.indexes[self.subset]}' if not is_test else None,\n        padding_values=FLAGS.padding_idx, \n        distribute_strategy=distribute_strategy,\n        torch=FLAGS.torch,\n        keras=FLAGS.keras,\n        subset=self.subset,\n        return_numpy=return_numpy,\n        ) \n      \n    result = self.adjust(result)\n    return result\n    \n  @staticmethod\n  def num_examples_per_epoch(subset, dir=None):\n    if subset == 'train':\n      num_examples = get_num_records(FLAGS.train_files)\n    elif subset == 'valid':\n      num_examples = get_num_records(FLAGS.valid_files)\n    else:\n      raise ValueError('Invalid data subset \"%s\"' % subset)\n    \n    assert num_examples\n    return num_examples\n\n  @staticmethod\n  def num_examples(subset, dir=None):\n    return Dataset.num_examples_per_epoch(subset, dir)\n\n  @property\n  def num_instances(self):\n    if self.num_instances_:\n      return self.num_instances_\n    assert self.files_\n    self.num_instances_ = get_num_records(self.files_, recount=self.recount)\n    return self.num_instances_\n\n  @property\n  def files(self):\n    return self.files_\n\n  @property\n  def records(self):\n    return self.files_\n\n  def __len__(self):\n    return self.num_instances or Dataset.num_examples_per_epoch(self.subset)\n\n  @property\n  def num_steps(self):\n    return -(-len(self) // self.batch_size)\n","metadata":{"_kg_hide-input":true,"_kg_hide-output":true,"execution":{"iopub.status.busy":"2023-08-27T00:43:05.463251Z","iopub.execute_input":"2023-08-27T00:43:05.463689Z","iopub.status.idle":"2023-08-27T00:43:05.539781Z","shell.execute_reply.started":"2023-08-27T00:43:05.463657Z","shell.execute_reply":"2023-08-27T00:43:05.538605Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Generate means.npy and stds.npy","metadata":{}},{"cell_type":"code","source":"records_pattern = f'../input/3rd-place-step1-gen-tfrecords-for-train/tfrecords/train/*.tfrec'\nrecord_files = glob.glob(records_pattern)\nic(record_files[:2])\ndataset = TfrecordsDataset('valid', \n                      files=record_files, \n                      incl_keys=['frames', 'n_frames'],\n                      varlen_keys=['frames'],\n                          )\ndatas = dataset.make_batch(1024, \n                           shuffle=False, \n                           drop_remainder=False, \n                           return_numpy=True)\nic(dataset.features_dict)\nic(dataset.num_instances)\nnum_steps = dataset.num_steps\nic(num_steps)","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:43:05.541305Z","iopub.execute_input":"2023-08-27T00:43:05.541647Z","iopub.status.idle":"2023-08-27T00:43:10.196679Z","shell.execute_reply.started":"2023-08-27T00:43:05.541618Z","shell.execute_reply":"2023-08-27T00:43:10.195299Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# https://stackoverflow.com/questions/5543651/computing-standard-deviation-in-a-stream\nclass OnlineVariance(object):\n    \"\"\"\n    Welford's algorithm computes the sample variance incrementally.\n    \"\"\"\n\n    def __init__(self, iterable=None, ddof=1):\n        self.ddof, self.n, self.mean_, self.M2 = ddof, 0, 0.0, 0.0\n        if iterable is not None:\n            for datum in iterable:\n                self.include(datum)\n\n    def add(self, datum):\n        self.n += 1\n        self.delta = datum - self.mean\n        self.mean_ += self.delta / self.n\n        self.M2 += self.delta * (datum - self.mean_)\n\n    @property\n    def variance(self):\n        return self.M2 / (self.n - self.ddof)\n\n    @property\n    def std(self):\n        return np.sqrt(self.variance)\n    \n    @property\n    def mean(self):\n      return self.mean_","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:43:10.198791Z","iopub.execute_input":"2023-08-27T00:43:10.199683Z","iopub.status.idle":"2023-08-27T00:43:10.209934Z","shell.execute_reply.started":"2023-08-27T00:43:10.199629Z","shell.execute_reply":"2023-08-27T00:43:10.208808Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ic(N_COLS)\nmeans = np.zeros([N_COLS], dtype=np.float32)\nstds = np.zeros([N_COLS], dtype=np.float32)\novs = [OnlineVariance() for _ in range(N_COLS)]\n# using streaming mean due to memory limit, notice since the last batch size might not be 1024, so maybe a bit different from all in cache mean results, however should not diff much\nfor x in tqdm(datas, total=num_steps, desc='Loop-dataset'):\n  batch_frames = x['frames']\n  batch_n_frames = x['n_frames']\n  batch_frames = batch_frames.reshape(batch_frames.shape[0], -1, N_COLS)\n  l = []\n  for frames, n_frames in zip(batch_frames, batch_n_frames):\n    frames = frames[:n_frames]\n    l.append(frames)\n  frames = np.concatenate(l)\n  for col, v in enumerate(frames.reshape([-1, N_COLS]).T):\n    v = v[~np.isnan(v)]\n    ovs[col].add(v.astype(np.float32).mean())\n        \nfor i, ov in enumerate(ovs):\n  means[i] = ov.mean\n  # very important, other wise keras and tflite results diff...\n  if ov.std >= 1e-6:\n    stds[i] = ov.std\n  else:\n    stds[i] = 1.","metadata":{"execution":{"iopub.status.busy":"2023-08-27T00:43:10.211427Z","iopub.execute_input":"2023-08-27T00:43:10.211795Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"np.save(f'{FLAGS.working}/means.npy', means)\nnp.save(f'{FLAGS.working}/stds.npy', stds)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}