Merge pull request #7 from ChunweiYan/feature/link_sdk_with_server

server/visualdl/visual_dl.py

@@ -13,6 +13,7 @@ from flask import Response
 from visualdl.log import logger
 import visualdl.mock.data as mock_data
 import visualdl.mock.tags as mock_tags
+import storage
 
 app = Flask(__name__, static_url_path="")
 
@@ -31,7 +32,14 @@ def option_parser():
         default=8040,
         action="store",
         dest="port",
-        help="rest api service port")
+        help="api service port")
+    parser.add_option(
+        "-t",
+        "--host",
+        type=str,
+        default="0.0.0.0",
+        action="store",
+        help="api service ip")
     parser.add_option(
         "--logdir", action="store", dest="logdir", help="log file directory")
     return parser.parse_args()
@@ -42,6 +50,8 @@ server_path = os.path.abspath(os.path.dirname(sys.argv[0]))
 static_file_path = "./frontend/dist/"
 mock_data_path = "./mock_data/"
 
+storage = storage.StorageReader(options.logdir)
+
 
 # return data
 # status, msg, data
@@ -85,7 +95,22 @@ def runs():
 @app.route("/data/plugin/scalars/tags")
 def tags():
     is_debug = bool(request.args.get('debug'))
-    result = gen_result(0, "", mock_tags.data())
+    tag = request.args.get('tag')
+    if is_debug:
+        result = mock_tags.data()
+    else:
+        result = {}
+        print 'modes', storage.modes()
+        for mode in storage.modes():
+            result[mode] = {}
+            reader = storage.as_mode(mode)
+            for tag in reader.tags("scalar"):
+                result[mode][tag] = {
+                    'displayName': reader.scalar(tag).caption(),
+                    'description': ""
+                }
+    print 'tags', result
+    result = gen_result(0, "", result)
     return Response(json.dumps(result), mimetype='application/json')
 
 
@@ -94,10 +119,22 @@ def scalars():
     run = request.args.get('run')
     tag = request.args.get('tag')
     is_debug = bool(request.args.get('debug'))
-    result = gen_result(0, "", mock_data.sequence_data())
+    if is_debug:
+        result = gen_result(0, "", mock_data.sequence_data())
+    else:
+        reader = storage.as_mode(run)
+        scalar = reader.scalar(tag)
+
+        records = scalar.records()
+        ids = scalar.ids()
+        timestamps = scalar.timestamps()
+
+        result = zip(timestamps, ids, records)
+        result = gen_result(0, "", result)
+
     return Response(json.dumps(result), mimetype='application/json')
 
 
 if __name__ == '__main__':
     logger.info(" port=" + str(options.port))
-    app.run(debug=False, host="0.0.0.0", port=options.port)
+    app.run(debug=False, host=options.host, port=options.port)

visualdl/logic/pybind.cc

@@ -38,11 +38,13 @@ PYBIND11_PLUGIN(core) {
     return vs::components::ScalarReader<T>(std::move(tablet));          \
   })
   py::class_<vs::Reader>(m, "Reader")
-      .def(
-          "__init__",
-          [](vs::Reader& instance,
-             const std::string& mode,
-             const std::string& dir) { new (&instance) vs::Reader(mode, dir); })
+      .def("__init__",
+           [](vs::Reader& instance, const std::string& dir) {
+             new (&instance) vs::Reader(dir);
+           })
+      .def("as_mode", &vs::Reader::AsMode)
+      .def("modes", [](vs::Reader& self) { return self.storage().modes(); })
+      .def("tags", &vs::Reader::tags)
       // clang-format off
     ADD_SCALAR(float)
     ADD_SCALAR(double)
@@ -59,8 +61,7 @@ PYBIND11_PLUGIN(core) {
   py::class_<vs::Writer>(m, "Writer")
       .def("__init__",
            [](vs::Writer& instance, const std::string& dir, int sync_cycle) {
-             new (&instance) vs::Writer(dir);
-             instance.storage().meta.cycle = sync_cycle;
+             new (&instance) vs::Writer(dir, sync_cycle);
            })
       .def("as_mode", &vs::Writer::AsMode)
       // clang-format off

visualdl/logic/sdk.h

@@ -6,16 +6,24 @@
 #include "visualdl/utils/string.h"
 namespace visualdl {
 
+const static std::string kDefaultMode{"default"};
+
 class Writer {
 public:
-  Writer(const std::string& dir) {
+  Writer(const std::string& dir, int sync_cycle) {
     storage_.SetDir(dir);
+    storage_.meta.cycle = sync_cycle;
+  }
+  Writer(const Writer& other) {
+    storage_ = other.storage_;
+    mode_ = other.mode_;
   }
 
-  Writer& AsMode(const std::string& mode) {
-    mode_ = mode;
+  Writer AsMode(const std::string& mode) {
+    Writer writer = *this;
     storage_.AddMode(mode);
-    return *this;
+    writer.mode_ = mode;
+    return writer;
   }
 
   Tablet AddTablet(const std::string& tag) {
@@ -31,13 +39,18 @@ public:
 
 private:
   Storage storage_;
-  std::string mode_;
+  std::string mode_{kDefaultMode};
 };
 
 class Reader {
 public:
-  Reader(const std::string& mode, const std::string& dir)
-      : mode_(mode), reader_(dir) {}
+  Reader(const std::string& dir) : reader_(dir) {}
+
+  Reader AsMode(const std::string& mode) {
+    auto tmp = *this;
+    tmp.mode_ = mode;
+    return tmp;
+  }
 
   TabletReader tablet(const std::string& tag) {
     auto tmp = mode_ + "/" + tag;
@@ -45,9 +58,45 @@ public:
     return reader_.tablet(tmp);
   }
 
+  std::vector<std::string> all_tags() {
+    auto tags = reader_.all_tags();
+    auto it =
+        std::remove_if(tags.begin(), tags.end(), [&](const std::string& tag) {
+          return !TagMatchMode(tag);
+        });
+    tags.erase(it + 1);
+    return tags;
+  }
+
+  std::vector<std::string> tags(const std::string& component) {
+    auto type = Tablet::type(component);
+    auto tags = reader_.tags(type);
+    CHECK(!tags.empty());
+    std::vector<std::string> res;
+    for (const auto& tag : tags) {
+      if (TagMatchMode(tag)) {
+        res.push_back(GenReadableTag(tag));
+      }
+    }
+    return res;
+  }
+
+  StorageReader& storage() { return reader_; }
+
+protected:
+  bool TagMatchMode(const std::string& tag) {
+    if (tag.size() <= mode_.size()) return false;
+    return tag.substr(0, mode_.size()) == mode_;
+  }
+  std::string GenReadableTag(const std::string& tag) {
+    auto tmp = tag;
+    string::TagDecode(tmp);
+    return tmp.substr(mode_.size() + 1);  // including `/`
+  }
+
 private:
   StorageReader reader_;
-  std::string mode_{"default"};
+  std::string mode_{kDefaultMode};
 };
 
 namespace components {
@@ -84,14 +133,13 @@ struct ScalarReader {
   std::vector<T> ids() const;
   std::vector<T> timestamps() const;
   std::string caption() const;
-  size_t total_records() {return reader_.total_records();}
+  size_t total_records() { return reader_.total_records(); }
   size_t size() const;
 
 private:
   TabletReader reader_;
 };
 
-
 }  // namespace components
 }  // namespace visualdl
 

visualdl/logic/sdk_test.cc

@@ -6,18 +6,19 @@ namespace visualdl {
 
 TEST(Scalar, write) {
   const auto dir = "./tmp/sdk_test";
-  Storage storage;
+  Writer writer__(dir, 1);
+  auto writer = writer__.AsMode("train");
   // write disk every time
-  storage.meta.cycle = 1;
-  storage.SetDir(dir);
-  auto tablet = storage.AddTablet("scalar0");
+  auto tablet = writer.AddTablet("scalar0");
   components::Scalar<int> scalar(tablet);
-  scalar.SetCaption("train");
   scalar.AddRecord(0, 12);
-  storage.PersistToDisk();
+  auto tablet1 = writer.AddTablet("model/layer/min");
+  components::Scalar<float> scalar1(tablet1);
+  scalar1.SetCaption("customized caption");
 
   // read from disk
-  StorageReader reader(dir);
+  Reader reader_(dir);
+  auto reader = reader_.AsMode("train");
   auto tablet_reader = reader.tablet("scalar0");
   auto scalar_reader = components::ScalarReader<int>(std::move(tablet_reader));
   auto captioin = scalar_reader.caption();
@@ -27,6 +28,16 @@ TEST(Scalar, write) {
   ASSERT_EQ(record.size(), 1);
   // check the first entry of first record
   ASSERT_EQ(record.front(), 12);
+
+  // check tags
+  ASSERT_EQ(reader.all_tags().size(), 1);
+  auto tags = reader.tags("scalar");
+  ASSERT_EQ(tags.size(), 2);
+  ASSERT_EQ(tags.front(), "scalar0");
+  ASSERT_EQ(tags[1], "model/layer/min");
+  components::ScalarReader<float> scalar_reader1(
+      reader.tablet("model/layer/min"));
+  ASSERT_EQ(scalar_reader1.caption(), "customized caption");
 }
 
 }  // namespace visualdl

visualdl/python/storage.py

@@ -9,8 +9,19 @@ dtypes = ("float", "double", "int32", "int64")
 
 class StorageReader(object):
 
-    def __init__(self, mode, dir):
-        self.reader = core.Reader(mode, dir)
+    def __init__(self, dir, reader=None):
+        self.dir = dir
+        self.reader = reader if reader else core.Reader(dir)
+
+    def as_mode(self, mode):
+        tmp = StorageReader(dir, self.reader.as_mode(mode))
+        return tmp
+
+    def modes(self):
+        return self.reader.modes()
+
+    def tags(self, kind):
+        return self.reader.tags(kind)
 
     def scalar(self, tag, type='float'):
         type2scalar = {
@@ -23,12 +34,14 @@ class StorageReader(object):
 
 class StorageWriter(object):
 
-    def __init__(self, dir, sync_cycle):
-        self.writer = core.Writer(dir, sync_cycle)
+    def __init__(self, dir, sync_cycle, writer=None):
+        self.dir = dir
+        self.sync_cycle = sync_cycle
+        self.writer = writer if writer else core.Writer(dir, sync_cycle)
 
     def as_mode(self, mode):
-        self.writer = self.writer.as_mode(mode)
-        return self
+        tmp = StorageWriter(self.dir, self.sync_cycle, self.writer.as_mode(mode))
+        return tmp
 
     def scalar(self, tag, type='float'):
         type2scalar = {

visualdl/python/test_read_service.py

-import summary
-import numpy as np
-import unittest
-import time
-
-
-class StorageTester(unittest.TestCase):
-    def test_storage(self):
-        summary.set_writable_storage("./tmp_dir")
-        time.sleep(5)
-        summary.stop_service()
-
-
-if __name__ == '__main__':
-    unittest.main()

visualdl/python/test_storage.py

@@ -4,20 +4,22 @@ import unittest
 import random
 import time
 
+
 class StorageTest(unittest.TestCase):
     def setUp(self):
         self.dir = "./tmp/storage_test"
 
     def test_read(self):
         print 'test write'
-        self.writer = storage.StorageWriter(self.dir, sync_cycle=1).as_mode("train")
+        self.writer = storage.StorageWriter(
+            self.dir, sync_cycle=1).as_mode("train")
         scalar = self.writer.scalar("model/scalar/min")
         # scalar.set_caption("model/scalar/min")
         for i in range(10):
             scalar.add_record(i, float(i))
 
         print 'test read'
-        self.reader = storage.StorageReader("train", self.dir)
+        self.reader = storage.StorageReader(self.dir).as_mode("train")
         scalar = self.reader.scalar("model/scalar/min")
         self.assertEqual(scalar.caption(), "train")
         records = scalar.records()

visualdl/python/test_write_service.py

-import summary
-import numpy as np
-import unittest
-import time
-
-
-class StorageTester(unittest.TestCase):
-    def test_read_storage(self):
-        summary.set_readable_storage("./tmp")
-        time.sleep(1)
-        scalar = summary.read_scalar('tag01')
-        time.sleep(5)
-        summary.stop_service()
-
-
-if __name__ == '__main__':
-    unittest.main()

visualdl/storage/storage.h

@@ -2,14 +2,15 @@
 #define VISUALDL_STORAGE_STORAGE_H
 
 #include <glog/logging.h>
-#include <vector>
+#include <algorithm>
 #include <set>
+#include <vector>
 
 #include "visualdl/logic/im.h"
-#include "visualdl/utils/guard.h"
 #include "visualdl/storage/storage.pb.h"
 #include "visualdl/storage/tablet.h"
 #include "visualdl/utils/filesystem.h"
+#include "visualdl/utils/guard.h"
 
 namespace visualdl {
 
@@ -41,29 +42,34 @@ struct Storage {
 
   mutable SimpleSyncMeta meta;
 
-  Storage() { data_ = std::make_shared<storage::Storage>(); }
-  Storage(const std::shared_ptr<storage::Storage>& x) : data_(x) {
+  Storage() {
+    data_ = std::make_shared<storage::Storage>();
+    tablets_ = std::make_shared<std::map<std::string, storage::Tablet>>();
+    modes_ = std::make_shared<std::set<std::string>>();
     time_t t;
     time(&t);
     data_->set_timestamp(t);
   }
+  Storage(const Storage& other)
+      : data_(other.data_), tablets_(other.tablets_), modes_(other.modes_) {}
 
   // write operations
   void AddMode(const std::string& x) {
     // avoid duplicate modes.
-    if (modes_.count(x) != 0) return;
+    if (modes_->count(x) != 0) return;
     *data_->add_modes() = x;
-    modes_.insert(x);
+    modes_->insert(x);
     WRITE_GUARD
   }
 
   Tablet AddTablet(const std::string& x) {
-    CHECK(tablets_.count(x) == 0) << "tablet [" << x << "] has existed";
-    tablets_[x] = storage::Tablet();
+    CHECK(tablets_->count(x) == 0) << "tablet [" << x << "] has existed";
+    (*tablets_)[x] = storage::Tablet();
     AddTag(x);
     LOG(INFO) << "really add tag " << x;
-    WRITE_GUARD
-    return Tablet(&tablets_[x], this);
+    // WRITE_GUARD
+    PersistToDisk();
+    return Tablet(&(*tablets_)[x], this);
   }
 
   void SetDir(const std::string& dir) { dir_ = dir; }
@@ -78,8 +84,8 @@ struct Storage {
 
     fs::SerializeToFile(*data_, meta_path(dir));
     for (auto tag : data_->tags()) {
-      auto it = tablets_.find(tag);
-      CHECK(it != tablets_.end()) << "tag " << tag << " not exist.";
+      auto it = tablets_->find(tag);
+      CHECK(it != tablets_->end()) << "tag " << tag << " not exist.";
       fs::SerializeToFile(it->second, tablet_path(dir, tag));
     }
   }
@@ -89,13 +95,16 @@ struct Storage {
 protected:
   void AddTag(const std::string& x) {
     *data_->add_tags() = x;
+    LOG(INFO) << "add tag " << x;
+    LOG(INFO) << "tag.size " << data_->tags_size();
+    WRITE_GUARD
   }
 
 private:
   std::string dir_;
-  std::map<std::string, storage::Tablet> tablets_;
+  std::shared_ptr<std::map<std::string, storage::Tablet>> tablets_;
   std::shared_ptr<storage::Storage> data_;
-  std::set<std::string> modes_;
+  std::shared_ptr<std::set<std::string>> modes_;
 };
 
 /*
@@ -105,13 +114,23 @@ struct StorageReader {
   StorageReader(const std::string& dir) : dir_(dir) {}
 
   // read operations
-  std::vector<std::string> Tags() {
+  std::vector<std::string> all_tags() {
     storage::Storage storage;
     Reload(storage);
     return std::vector<std::string>(storage.tags().begin(),
                                     storage.tags().end());
   }
-  std::vector<std::string> Modes() {
+  std::vector<std::string> tags(Tablet::Type component) {
+    auto tags = all_tags();
+    auto it =
+        std::remove_if(tags.begin(), tags.end(), [&](const std::string& tag) {
+          auto tb = tablet(tag);
+          return tb.type() != component;
+        });
+    tags.resize(it - tags.begin());
+    return tags;
+  }
+  std::vector<std::string> modes() {
     storage::Storage storage;
     Reload(storage);
     return std::vector<std::string>(storage.modes().begin(),

visualdl/storage/storage_test.cc

@@ -25,7 +25,7 @@ TEST_F(StorageTest, main) {
   entry.Set(12);
 
   StorageReader reader("./tmp/storage_test");
-  auto modes = reader.Modes();
+  auto modes = reader.modes();
 
   ASSERT_EQ(modes.size(), 2);
   ASSERT_EQ(modes[0], "train");

visualdl/storage/tablet.h

 #ifndef VISUALDL_TABLET_H
 #define VISUALDL_TABLET_H
 
+#include <glog/logging.h>
+
 #include "visualdl/logic/im.h"
 #include "visualdl/storage/record.h"
 #include "visualdl/storage/storage.pb.h"
@@ -18,6 +20,19 @@ struct Tablet {
 
   Tablet(storage::Tablet* x, Storage* parent) : data_(x), x_(parent) {}
 
+  static Type type(const std::string& name) {
+    if (name == "scalar") {
+      return kScalar;
+    }
+    if (name == "histogram") {
+      return kHistogram;
+    }
+    if (name == "image") {
+      return kImage;
+    }
+    LOG(ERROR) << "unknown component: " << name;
+  }
+
   // write operations.
   void SetNumSamples(int x) {
     data_->set_num_samples(x);