{"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":"\n","metadata":{}},{"cell_type":"code","source":"\"\"\"\nTo prepare training dataset, step-by-step:\n\n1. List all .npy data files.\n2. Shuffle file names.\n3. Load the training label file, match .npy file with its label.\n4. Split data files to blocks.\n4. Create a TFDataset for each block.\n5. Load numpy array from file, cast data to bfloat16 to reduce disk storage/memory usage.\n6. Save data blocks to separated directories.\n\"\"\"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"import libraries\"\"\"\n\nimport random\nimport tensorflow as tf\nimport numpy as np\nfrom functools import partial\nfrom pathlib import Path","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"\ntensorflow data loading functions\n\"\"\"\n@tf.function\ndef load_numpy(filename, return_id = False):\n    \"\"\"Load numpy file inside tf data pipeline. Also return ident if needed.\"\"\"\n    s = tf.io.read_file(filename)\n    data_len = 8 * 4096 * 3 # shape 3 x 4096, dtype=float64\n    s = tf.strings.substr(s, tf.strings.length(s) - data_len, data_len, unit='BYTE') \n    a = tf.io.decode_raw(s, tf.float64, little_endian=True, fixed_length=data_len)\n    a = a * (1e20) # the absolute value of a is very small (~1e-20)\n    a = tf.cast(a, tf.bfloat16) # convert float64 -> bfloat16 to reduce memory usage\n    a = tf.reshape(a, (3, 4096))\n    if return_id:\n        ident = tf.strings.substr(filename, -14, 10)\n        return ident, a\n    else:\n        return a\n\n\n# load time series data which is seperated from label data\ndef load_numpy_dataset(file_pattern, return_id=False):\n    data=tf.data.Dataset.list_files(file_pattern, False)\n    data = data.map(partial(load_numpy, return_id=return_id), num_parallel_calls=tf.data.AUTOTUNE)\n    return data ","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"getting list of all training data files\"\"\"\ndata_dir = Path('/kaggle/input/g2net-gravitational-wave-detection/')\ntrain_data_dir = data_dir / 'train'\ntest_data_dir = data_dir / 'test'\n\ntrain_data_files = sorted(train_data_dir.glob('*/*/*/*.npy'))\nrandom.Random(42).shuffle(train_data_files) # random shuffle all training records\n\n# check random shuffling\nprint(train_data_files[:10])\n\ndef load_training_labels():\n    file_name = '/kaggle/input/g2net-gravitational-wave-detection/training_labels.csv'\n    labels = np.loadtxt(file_name, delimiter=',', skiprows=1, usecols=(1,), dtype=np.int32)\n    ids = np.loadtxt(file_name, delimiter=',', skiprows=1, usecols=(0,), dtype=str)\n    return dict(zip(ids, labels))\n\nlabels_ = load_training_labels()\ntrain_data = [ (fn, labels_[fn.stem]) for fn in train_data_files ]","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"\n1. Split training records to multiple `data blocks` for later use in training.\n2. Save data blocks to separated directories.\n\"\"\"\nnum_blocks = 10\nsize_per_block = len(train_data) // num_blocks\ndata_blocks = [ train_data[i * size_per_block : (i+1) * size_per_block] for i in range(num_blocks) ]\ndef gen(data_block):\n    for fn, label in data_block:\n        yield str(fn), label\n\nfor i in range(num_blocks):\n    tfdata = tf.data.Dataset.from_generator(\n        partial(gen, data_block=data_blocks[i]), \n        output_signature = (tf.TensorSpec(shape=(), dtype=tf.string), tf.TensorSpec(shape=(), dtype=tf.int32))\n    )\n    tfdata = tfdata.map(lambda x, y: (load_numpy(x), y))\n    file_name = f'./train_data_{i:02d}'\n    print(f\"saving data block to {file_name}\")\n    tf.data.experimental.save(tfdata, file_name)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# save test data to disk\ntest_dataset = load_numpy_dataset('/kaggle/input/g2net-gravitational-wave-detection/test/*/*/*/*.npy', return_id=True)\ntf.data.experimental.save(test_dataset, './test_data')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"\nAn example code showing how to load the prepared data.\n\"\"\"\n\nimport tensorflow as tf\n\nnum_blocks = 10\n\n# load all training data blocks\ndata_blocks = [\n    tf.data.experimental.load(\n        f'train_data_{i:02d}', \n        (tf.TensorSpec(shape=(3, 4096), dtype=tf.bfloat16), tf.TensorSpec(shape=(), dtype=tf.int32))\n    ) for i in range(num_blocks)\n]\n    \n# create train/val/test split\nval_dataset = data_blocks[0]\ntest_dataset = data_blocks[1]\ntrain_dataset = data_blocks[2]\n# concatenate the remain data blocks into the train dataset.\nfor i in range(3, 10):\n    train_dataset = train_dataset.concatenate(data_blocks[i])\n\n\n# load data for final prediction\nsubmission_dataset = tf.data.experimental.load(\n    'test_data', \n    (tf.TensorSpec(shape=(), dtype=tf.string), tf.TensorSpec(shape=(3, 4096), dtype=tf.bfloat16))\n)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}