Untitled.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "254bc3d3-e23b-449f-91a1-ad1819ec1457",
   "metadata": {},
   "outputs": [],
   "source": [
    "kfp_endpoint='http://localhost:8080'\n",
    "\n",
    "from typing import NamedTuple\n",
    "\n",
    "import kfp\n",
    "import kfp.dsl as dsl\n",
    "from kfp.components import InputPath, InputTextFile, OutputPath, OutputTextFile\n",
    "from kfp.components import func_to_container_op"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "7a36dc0f-20a7-4a52-9c46-5400f1134b21",
   "metadata": {},
   "outputs": [],
   "source": [
    "#def generate_tagset(changeset_path: InputPath(str)) -> str:\n",
    "def generate_tagset(changeset_path: InputPath(str), train_type_path: OutputPath(str)):\n",
    "    '''generate tagset from the changeset'''\n",
    "    import tagset_gen\n",
    "    tag_dict_gen = tagset_gen.run()\n",
    "    import json\n",
    "    import pickle\n",
    "    #return 'iterative'\n",
    "    train_type = 'iterative'\n",
    "    with open(train_type_path, 'wb') as writer:\n",
    "        pickle.dump(train_type, writer)\n",
    "generate_tagset_op = kfp.components.create_component_from_func(generate_tagset, output_component_file='generate_tagset_component.yaml', base_image=\"lkorver/praxi-columbus-base:0.03\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "94378a9e-8ebd-4e0e-a1c7-b473220c46b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def iterative_training(tagset_path: InputPath(str), prediction_path: OutputPath(str)):\n",
    "    ''''''\n",
    "    import main\n",
    "    import os\n",
    "    import json\n",
    "    import pickle\n",
    "    import time\n",
    "    args = main.get_inputs()\n",
    "    data_loaded = []\n",
    "    with open(tagset_path, 'r') as stream:\n",
    "        for line in stream:\n",
    "            temp = json.loads(line)\n",
    "            if (type(temp) != None):\n",
    "                data_loaded.append(temp)\n",
    "    model = main.iterative_train(data_loaded, args)\n",
    "    pred = main.test(model, data_loaded, args)\n",
    "    \n",
    "    with open(prediction_path, 'wb') as writer:\n",
    "        pickle.dump(pred, writer)\n",
    "generate_ittrain_op = kfp.components.create_component_from_func(iterative_training, output_component_file='generate_ittrain_component.yaml', base_image=\"lkorver/praxi-vw-base:0.075\")\n",
    "\n",
    "def multilabel_training(tagset_path: InputPath(str), prediction_path: OutputPath(str)):\n",
    "    import main\n",
    "    import os\n",
    "    import json\n",
    "    import pickle\n",
    "    import time\n",
    "    args = main.get_inputs()\n",
    "    data_loaded = []\n",
    "    with open(tagset_path, 'r') as stream:\n",
    "        for line in stream:\n",
    "            temp = json.loads(line)\n",
    "            if (type(temp) != None):\n",
    "                data_loaded.append(temp)\n",
    "    model = main.multilabel_train(data_loaded, data_loaded, args) #temporarily passing same train/test data\n",
    "    pred = main.test(model, data_loaded, args)\n",
    "    print(\"output\", pred)\n",
    "    with open(prediction_path, 'wb') as writer:\n",
    "        pickle.dump(pred,writer)\n",
    "generate_multilabel_op = kfp.components.create_component_from_func(multilabel_training, output_component_file='generate_multilabel_component.yaml', base_image=\"lkorver/praxi-vw-base:0.075\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "25c9575a-f818-43fd-a4d9-d70db9180e47",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<a href=\"http://localhost:8080/#/experiments/details/1ba78fb5-531e-4732-af47-57649a27a88a\" target=\"_blank\" >Experiment details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<a href=\"http://localhost:8080/#/runs/details/f707e0d8-9809-475b-a5de-76d262c02d08\" target=\"_blank\" >Run details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "RunPipelineResult(run_id=f707e0d8-9809-475b-a5de-76d262c02d08)"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Reading bigger data\n",
    "@func_to_container_op\n",
    "def print_text(text_path: InputPath()): # The \"text\" input is untyped so that any data can be printed\n",
    "    '''Print text'''\n",
    "    with open(text_path, 'rb') as reader:\n",
    "        for line in reader:\n",
    "            print(line, end = '')\n",
    "\n",
    "def praxi_pipeline():\n",
    "    vop = dsl.VolumeOp(\n",
    "        name=\"snapshot-claim-4\",\n",
    "        resource_name=\"snapshot-claim-4\",\n",
    "        size=\"1Gi\",\n",
    "        modes=dsl.VOLUME_MODE_RWM,\n",
    "        volume_name=\"snapshot-volume\",\n",
    "        storage_class=\"manual\",\n",
    "        generate_unique_name=False,\n",
    "        action='apply',\n",
    "        set_owner_reference=True\n",
    "    )\n",
    "    # dir_to_scan = \"\"\n",
    "    # changeset = generate_changeset(dir_to_scan).add_pvolumes({\"/fake-snapshot\": vop.volume})\n",
    "    # tagset = generate_tagset(changeset.outputs[\"output_text\"]).add_pvolumes({\"/fake-snapshot\": vop.volume})\n",
    "    tagset = generate_tagset_op(\"/\").add_pvolumes({\"/fake-snapshot\": vop.volume})\n",
    "    tagset.execution_options.caching_strategy.max_cache_staleness = \"P0D\"\n",
    "    #with dsl.Condition(tagset.outputs[\"train_type\"] != 'iterative'):\n",
    "    with dsl.Condition(tagset.outputs[\"train_type\"] != 'iterative'):\n",
    "        mul_train = generate_multilabel_op(tagset.outputs[\"train_type\"]).add_pvolumes({\"/fake-snapshot\": vop.volume})\n",
    "        mul_train.execution_options.caching_strategy.max_cache_staleness = \"P0D\"\n",
    "        print_text(mul_train.output)\n",
    "\n",
    "    #with dsl.Condition(tagset.outputs[\"train_type\"] == 'iterative'):\n",
    "    with dsl.Condition(tagset.outputs[\"train_type\"] == 'iterative'):\n",
    "        it_train = generate_ittrain_op(tagset.outputs[\"train_type\"]).add_pvolumes({\"/fake-snapshot\": vop.volume})\n",
    "        it_train.execution_options.caching_strategy.max_cache_staleness = \"P0D\"\n",
    "        print_text(it_train.output)\n",
    "\n",
    "kfp.Client(host=kfp_endpoint).create_run_from_pipeline_func(praxi_pipeline, arguments={})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b2b83d9d-3a10-4f52-a3a2-7f60e8fb7551",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}