wldetection2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# model definition and training\n",
    "import chainer\n",
    "from PIL import Image\n",
    "import numpy as np\n",
    "import argparse\n",
    "import glob\n",
    "from time import strftime\n",
    "import chainer\n",
    "import chainer.functions as F\n",
    "import chainer.links as L\n",
    "from chainer.backends import cuda\n",
    "from sklearn.metrics import mean_squared_error\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "from chainer import training\n",
    "from chainer.training import extensions\n",
    "\n",
    "class WLClassifier(chainer.Chain):\n",
    "    def __init__(self, predictor):\n",
    "        super(WLClassifier, self).__init__()\n",
    "        with self.init_scope():\n",
    "            self.predictor = predictor\n",
    "    \n",
    "    def __call__(self, *args):\n",
    "        x = args[0]\n",
    "        t = args[-1]\n",
    "        self.y = None\n",
    "        self.loss = None\n",
    "        self.accuracy = None\n",
    "        self.y = self.predictor(x)\n",
    "        self.loss = self.lossfun(self.y, t)\n",
    "        chainer.report({'loss': self.loss}, self)\n",
    "        #print(np.sqrt(mean_squared_error(t[0][0], self.y.data[0][0])))\n",
    "        return self.loss\n",
    "        \n",
    "    def lossfun(self, y, t):\n",
    "        return F.sigmoid_cross_entropy(y,t)\n",
    "\n",
    "class MLP(chainer.Chain):\n",
    "\n",
    "    def __init__(self, n_units1, n_units2):\n",
    "        super(MLP, self).__init__()\n",
    "        with self.init_scope():\n",
    "            initializer = chainer.initializers.HeNormal()\n",
    "            self.l1 = L.Convolution2D(3, n_units1, 5, stride=1, pad=2, initialW=initializer)\n",
    "            #self.l2 = L.Convolution2D(None, n_units2, 5, stride=1, pad=2, initialW=initializer)\n",
    "            self.l3 = L.Convolution2D(None, 1, 1)\n",
    "            \n",
    "    def __call__(self, *args):\n",
    "\n",
    "        #assert len(args) >= 2\n",
    "        x = args[0]\n",
    "        #t = args[-1]\n",
    "        #t = x\n",
    "        #self.y = None\n",
    "        #self.loss = None\n",
    "        #self.accuracy = None\n",
    "        self.y = self.forward(x)\n",
    "        #self.loss = self.lossfun(self.y, t)\n",
    "        #chainer.report({'loss': self.loss}, self)\n",
    "        #print(np.sqrt(mean_squared_error(t[0][0], self.y.data[0][0])))\n",
    "        return self.y\n",
    " \n",
    "    def forward(self, x):\n",
    "        #print(x.data.shape)\n",
    "        h1 = F.relu(self.l1(x))\n",
    "        #h2 = F.relu(self.l2(h1))\n",
    "        #print(h1.data.shape)\n",
    "        h3 = self.l3(h1)\n",
    "        return h3\n",
    "\n",
    "    def lossfun(self, y, t):\n",
    "        #return F.sigmoid_cross_entropy(y,t)\n",
    "        return (y * (t - (y >= 0)) - F.log1p(F.exp(-F.abs(y))))\n",
    "        #return chainer.functions.mean_squared_error(F.tanh(y), (-t*2+1).astype(np.float32))\n",
    "        \n",
    "   \n",
    "imw = 320\n",
    "imh = 240\n",
    "\n",
    "\n",
    "model = MLP(4,2)\n",
    "x = np.zeros((0, 3, imh, imw), dtype=np.float32)\n",
    "t = np.zeros((0, 1, imh, imw), dtype=np.int32)\n",
    "classifier = WLClassifier(model)\n",
    "\n",
    "import os\n",
    "jpgfiles = glob.glob('white-line2020/*.jpg')\n",
    "print(jpgfiles)\n",
    "for f in jpgfiles:\n",
    "    img = Image.open(f).resize((imw, imh))\n",
    "    plt.imshow(np.array(img))\n",
    "    a = np.asarray(img).transpose(2,0,1).astype(np.float32)/255.\n",
    "    a1 = np.expand_dims(a,axis=0)\n",
    "    x = np.append(x, a1, axis=0)\n",
    "    \n",
    "    lfile = os.path.splitext(f)[0] + '_label.png'\n",
    "    print(lfile)\n",
    "    img = Image.open(lfile).resize((imw, imh))\n",
    "    img = img.convert('L')\n",
    "    a = np.asarray(img).astype(np.float32)/254.\n",
    "    a = a.astype(np.int32)\n",
    "    print(np.sum(a))\n",
    "    a1 = np.expand_dims(a,axis=0)\n",
    "    a1 = np.expand_dims(a1,axis=0)\n",
    "    t = np.append(t, a1, axis=0)\n",
    "\n",
    "optimizer = chainer.optimizers.Adam()\n",
    "optimizer.setup(classifier)\n",
    "model.to_gpu()\n",
    "\n",
    "for rep in range(20):\n",
    "    train = chainer.datasets.TupleDataset(x, t)\n",
    "    train_iter = chainer.iterators.SerialIterator(train, 4)\n",
    "    #test_iter = chainer.iterators.SerialIterator(test, 1, repeat=False, shuffle=False)\n",
    "\n",
    "    # Set up a trainer\n",
    "    updater = training.updaters.StandardUpdater(\n",
    "        train_iter, optimizer, device=0)\n",
    "    trainer = training.Trainer(updater, (500, 'epoch'), out='test1')\n",
    "\n",
    "    # Evaluate the model with the test dataset for each epoch\n",
    "    #trainer.extend(extensions.Evaluator(test_iter, model, device=args.gpu))\n",
    "\n",
    "    # Dump a computational graph from 'loss' variable at the first iteration\n",
    "    # The \"main\" refers to the target link of the \"main\" optimizer.\n",
    "    # trainer.extend(extensions.dump_graph('main/loss'))\n",
    "\n",
    "    # Write a log of evaluation statistics for each epoch\n",
    "    trainer.extend(extensions.LogReport())\n",
    "    trainer.extend(extensions.PrintReport(\n",
    "        ['epoch', 'main/loss', 'validation/main/loss',\n",
    "         'main/accuracy', 'validation/main/accuracy', 'elapsed_time']))\n",
    "\n",
    "    # Print a progress bar to stdout\n",
    "    #trainer.extend(extensions.ProgressBar())\n",
    "    trainer.run()\n",
    "    \n",
    "    \n",
    "chainer.serializers.save_npz('model_' + strftime(\"%Y%m%d_%H%M%S\") + '.npz', model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "def eval_image(fname, thre):\n",
    "    model.to_cpu()\n",
    "    testx = np.zeros((0, 3, imh, imw), dtype = np.float32)\n",
    "\n",
    "    y2 = cuda.to_cpu(model.y.data)\n",
    "    #plt.imshow(y2[0][0])\n",
    "    img = Image.open(fname).convert('RGB').resize((imw,imh))\n",
    "    a = np.asarray(img).transpose(2,0,1).astype(np.float32)/255.\n",
    "    a1 = np.expand_dims(a,axis=0)\n",
    "    #print(a1.shape)\n",
    "    testx = np.append(testx, a1, axis=0)\n",
    "\n",
    "    import time\n",
    "    t0 = time.time()\n",
    "    with chainer.using_config('train', False), \\\n",
    "            chainer.using_config('enable_backprop', False):\n",
    "        testy = model.forward(testx)\n",
    "    print('forward time [s]: ' + str(time.time()-t0))\n",
    "\n",
    "    #fig, axs = plt.subplots(1,3)\n",
    "    #plt.imshow(img, ax=axs[0])\n",
    "    imd = Image.new('RGB', (imw*2, imh))\n",
    "    imd.paste(img)\n",
    "    thimg = (testy.data[0][0]> thre)\n",
    "    thimg = thimg.astype(np.uint8) * 255\n",
    "\n",
    "    thimg = Image.fromarray(thimg, 'L')\n",
    "    plt.imshow(thimg)\n",
    "\n",
    "    imd.paste(thimg, (imw, 0))\n",
    "    plt.imshow(imd)\n",
    "    return testy.data[0][0]\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "chainer.serializers.load_npz(\"model_20200107_070457.npz\", model)\n",
    "a = eval_image('white-line2020/000206.jpg', 0)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}