You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

374 lines
14 KiB

  1. # -*- coding: utf-8 -*-
  2. import nilmdb.server
  3. import nilmdb.client
  4. import nilmdb.client.numpyclient
  5. from nilmdb.utils.printf import *
  6. from nilmdb.utils import timestamper
  7. from nilmdb.client import ClientError, ServerError
  8. import datetime_tz
  9. from nose.plugins.skip import SkipTest
  10. from nose.tools import *
  11. from nose.tools import assert_raises
  12. import itertools
  13. import distutils.version
  14. from testutil.helpers import *
  15. import numpy as np
  16. testdb = "tests/numpyclient-testdb"
  17. testurl = "http://localhost:32180/"
  18. def setup_module():
  19. global test_server, test_db
  20. # Clear out DB
  21. recursive_unlink(testdb)
  22. # Start web app on a custom port
  23. test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(
  24. testdb, bulkdata_args = { "file_size" : 16384,
  25. "files_per_dir" : 3 } )
  26. test_server = nilmdb.server.Server(test_db, host = "127.0.0.1",
  27. port = 32180, stoppable = False,
  28. fast_shutdown = True,
  29. force_traceback = True)
  30. test_server.start(blocking = False)
  31. def teardown_module():
  32. global test_server, test_db
  33. # Close web app
  34. test_server.stop()
  35. test_db.close()
  36. class TestNumpyClient(object):
  37. def test_numpyclient_01_basic(self):
  38. # Test basic connection
  39. client = nilmdb.client.numpyclient.NumpyClient(url = testurl)
  40. version = client.version()
  41. eq_(distutils.version.LooseVersion(version),
  42. distutils.version.LooseVersion(test_server.version))
  43. # Verify subclassing
  44. assert(isinstance(client, nilmdb.client.Client))
  45. # Layouts
  46. for layout in "int8_t", "something_8", "integer_1":
  47. with assert_raises(ValueError):
  48. for x in client.stream_extract_numpy("/foo", layout=layout):
  49. pass
  50. for layout in "int8_1", "uint8_30", "int16_20", "float64_100":
  51. with assert_raises(ClientError) as e:
  52. for x in client.stream_extract_numpy("/foo", layout=layout):
  53. pass
  54. in_("No such stream", str(e.exception))
  55. with assert_raises(ClientError) as e:
  56. for x in client.stream_extract_numpy("/foo"):
  57. pass
  58. in_("can't get layout for path", str(e.exception))
  59. client.close()
  60. def test_numpyclient_02_extract(self):
  61. client = nilmdb.client.numpyclient.NumpyClient(url = testurl)
  62. # Insert some data as text
  63. client.stream_create("/newton/prep", "float32_8")
  64. testfile = "tests/data/prep-20120323T1000"
  65. start = nilmdb.utils.time.parse_time("20120323T1000")
  66. rate = 120
  67. data = timestamper.TimestamperRate(testfile, start, rate)
  68. result = client.stream_insert("/newton/prep", data,
  69. start, start + 119999777)
  70. # Extract Numpy arrays
  71. array = None
  72. pieces = 0
  73. for chunk in client.stream_extract_numpy("/newton/prep", maxrows=1000):
  74. pieces += 1
  75. if array is not None:
  76. array = np.vstack((array, chunk))
  77. else:
  78. array = chunk
  79. eq_(array.shape, (14400, 9))
  80. eq_(pieces, 15)
  81. # Try structured
  82. s = list(client.stream_extract_numpy("/newton/prep", structured = True))
  83. assert(np.array_equal(np.c_[s[0]['timestamp'], s[0]['data']], array))
  84. # Compare. Will be close but not exact because the conversion
  85. # to and from ASCII was lossy.
  86. data = timestamper.TimestamperRate(testfile, start, rate)
  87. actual = np.fromstring(" ".join(data), sep=' ').reshape(14400, 9)
  88. assert(np.allclose(array, actual))
  89. client.close()
  90. def test_numpyclient_03_insert(self):
  91. client = nilmdb.client.numpyclient.NumpyClient(url = testurl)
  92. # Limit _max_data just to get better coverage
  93. old_max_data = nilmdb.client.numpyclient.StreamInserterNumpy._max_data
  94. nilmdb.client.numpyclient.StreamInserterNumpy._max_data = 100000
  95. client.stream_create("/test/1", "uint16_1")
  96. client.stream_insert_numpy("/test/1",
  97. np.array([[0, 1],
  98. [1, 2],
  99. [2, 3],
  100. [3, 4]]))
  101. # Wrong number of dimensions
  102. with assert_raises(ValueError) as e:
  103. client.stream_insert_numpy("/test/1",
  104. np.array([[[0, 1],
  105. [1, 2]],
  106. [[3, 4],
  107. [4, 5]]]))
  108. in_("wrong number of dimensions", str(e.exception))
  109. # Wrong number of fields
  110. with assert_raises(ValueError) as e:
  111. client.stream_insert_numpy("/test/1",
  112. np.array([[0, 1, 2],
  113. [1, 2, 3],
  114. [3, 4, 5],
  115. [4, 5, 6]]))
  116. in_("wrong number of fields", str(e.exception))
  117. # Unstructured
  118. client.stream_create("/test/2", "float32_8")
  119. client.stream_insert_numpy(
  120. "/test/2",
  121. client.stream_extract_numpy(
  122. "/newton/prep", structured = False, maxrows = 1000))
  123. # Structured, and specifying layout
  124. client.stream_create("/test/3", "float32_8")
  125. client.stream_insert_numpy(
  126. path = "/test/3", layout = "float32_8",
  127. data = client.stream_extract_numpy(
  128. "/newton/prep", structured = True, maxrows = 1000))
  129. # Structured, specifying wrong layout
  130. client.stream_create("/test/4", "float32_8")
  131. with assert_raises(ValueError) as e:
  132. client.stream_insert_numpy(
  133. "/test/4", layout = "uint16_1",
  134. data = client.stream_extract_numpy(
  135. "/newton/prep", structured = True, maxrows = 1000))
  136. in_("wrong dtype", str(e.exception))
  137. # Unstructured, and specifying wrong layout
  138. client.stream_create("/test/5", "float32_8")
  139. with assert_raises(ClientError) as e:
  140. client.stream_insert_numpy(
  141. "/test/5", layout = "uint16_8",
  142. data = client.stream_extract_numpy(
  143. "/newton/prep", structured = False, maxrows = 1000))
  144. # timestamps will be screwy here, because data will be parsed wrong
  145. in_("error parsing input data", str(e.exception))
  146. # Make sure the /newton/prep copies are identical
  147. a = np.vstack(client.stream_extract_numpy("/newton/prep"))
  148. b = np.vstack(client.stream_extract_numpy("/test/2"))
  149. c = np.vstack(client.stream_extract_numpy("/test/3"))
  150. assert(np.array_equal(a,b))
  151. assert(np.array_equal(a,c))
  152. # Make sure none of the files are greater than 16384 bytes as
  153. # we configured with the bulkdata_args above.
  154. datapath = os.path.join(testdb, "data")
  155. for (dirpath, dirnames, filenames) in os.walk(datapath):
  156. for f in filenames:
  157. fn = os.path.join(dirpath, f)
  158. size = os.path.getsize(fn)
  159. if size > 16384:
  160. raise AssertionError(sprintf("%s is too big: %d > %d\n",
  161. fn, size, 16384))
  162. nilmdb.client.numpyclient.StreamInserterNumpy._max_data = old_max_data
  163. client.close()
  164. def test_numpyclient_04_context(self):
  165. # Like test_client_context, but with Numpy data
  166. client = nilmdb.client.numpyclient.NumpyClient(testurl)
  167. client.stream_create("/context/test", "uint16_1")
  168. with client.stream_insert_numpy_context("/context/test") as ctx:
  169. # override _max_rows to trigger frequent server updates
  170. ctx._max_rows = 2
  171. ctx.insert([[1000, 1]])
  172. ctx.insert([[1010, 1], [1020, 1], [1030, 1]])
  173. ctx.insert([[1040, 1], [1050, 1]])
  174. ctx.finalize()
  175. ctx.insert([[1070, 1]])
  176. ctx.update_end(1080)
  177. ctx.finalize()
  178. ctx.update_start(1090)
  179. ctx.insert([[1100, 1]])
  180. ctx.insert([[1110, 1]])
  181. ctx.send()
  182. ctx.insert([[1120, 1], [1130, 1], [1140, 1]])
  183. ctx.update_end(1160)
  184. ctx.insert([[1150, 1]])
  185. ctx.update_end(1170)
  186. ctx.insert([[1160, 1]])
  187. ctx.update_end(1180)
  188. ctx.insert([[1170, 123456789.0]])
  189. ctx.finalize()
  190. ctx.insert(np.zeros((0,2)))
  191. with assert_raises(ClientError):
  192. with client.stream_insert_numpy_context("/context/test",
  193. 1000, 2000) as ctx:
  194. ctx.insert([[1180, 1]])
  195. with assert_raises(ClientError):
  196. with client.stream_insert_numpy_context("/context/test",
  197. 2000, 3000) as ctx:
  198. ctx._max_rows = 2
  199. ctx.insert([[3180, 1]])
  200. ctx.insert([[3181, 1]])
  201. with client.stream_insert_numpy_context("/context/test",
  202. 2000, 3000) as ctx:
  203. # make sure our override wasn't permanent
  204. ne_(ctx._max_rows, 2)
  205. ctx.insert([[2250, 1]])
  206. ctx.finalize()
  207. with assert_raises(ClientError):
  208. with client.stream_insert_numpy_context("/context/test",
  209. 3000, 4000) as ctx:
  210. ctx.insert([[3010, 1]])
  211. ctx.insert([[3020, 2]])
  212. ctx.insert([[3030, 3]])
  213. ctx.insert([[3040, 4]])
  214. ctx.insert([[3040, 4]]) # non-monotonic after a few lines
  215. ctx.finalize()
  216. eq_(list(client.stream_intervals("/context/test")),
  217. [ [ 1000, 1051 ],
  218. [ 1070, 1080 ],
  219. [ 1090, 1180 ],
  220. [ 2000, 3000 ] ])
  221. client.stream_remove("/context/test")
  222. client.stream_destroy("/context/test")
  223. client.close()
  224. def test_numpyclient_05_emptyintervals(self):
  225. # Like test_client_emptyintervals, with insert_numpy_context
  226. client = nilmdb.client.numpyclient.NumpyClient(testurl)
  227. client.stream_create("/empty/test", "uint16_1")
  228. def info():
  229. result = []
  230. for interval in list(client.stream_intervals("/empty/test")):
  231. result.append((client.stream_count("/empty/test", *interval),
  232. interval))
  233. return result
  234. eq_(info(), [])
  235. # Insert a region with just a few points
  236. with client.stream_insert_numpy_context("/empty/test") as ctx:
  237. ctx.update_start(100)
  238. ctx.insert([[140, 1]])
  239. ctx.insert([[150, 1]])
  240. ctx.insert([[160, 1]])
  241. ctx.update_end(200)
  242. ctx.finalize()
  243. eq_(info(), [(3, [100, 200])])
  244. # Delete chunk, which will leave one data point and two intervals
  245. client.stream_remove("/empty/test", 145, 175)
  246. eq_(info(), [(1, [100, 145]),
  247. (0, [175, 200])])
  248. # Try also creating a completely empty interval from scratch,
  249. # in a few different ways.
  250. client.stream_insert("/empty/test", "", 300, 350)
  251. client.stream_insert("/empty/test", [], 400, 450)
  252. with client.stream_insert_numpy_context("/empty/test", 500, 550):
  253. pass
  254. # If enough timestamps aren't provided, empty streams won't be created.
  255. client.stream_insert("/empty/test", [])
  256. with client.stream_insert_numpy_context("/empty/test"):
  257. pass
  258. client.stream_insert("/empty/test", [], start = 600)
  259. with client.stream_insert_numpy_context("/empty/test", start = 700):
  260. pass
  261. client.stream_insert("/empty/test", [], end = 850)
  262. with client.stream_insert_numpy_context("/empty/test", end = 950):
  263. pass
  264. # Equal start and end is OK as long as there's no data
  265. with assert_raises(ClientError) as e:
  266. with client.stream_insert_numpy_context("/empty/test",
  267. start=9, end=9) as ctx:
  268. ctx.insert([[9, 9]])
  269. ctx.finalize()
  270. in_("have data to send, but invalid start/end times", str(e.exception))
  271. with client.stream_insert_numpy_context("/empty/test",
  272. start=9, end=9) as ctx:
  273. pass
  274. # reusing a context object is bad
  275. with assert_raises(Exception) as e:
  276. ctx.insert([[9, 9]])
  277. # Try various things that might cause problems
  278. with client.stream_insert_numpy_context("/empty/test",
  279. 1000, 1050) as ctx:
  280. ctx.finalize() # inserts [1000, 1050]
  281. ctx.finalize() # nothing
  282. ctx.finalize() # nothing
  283. ctx.insert([[1100, 1]])
  284. ctx.finalize() # inserts [1100, 1101]
  285. ctx.update_start(1199)
  286. ctx.insert([[1200, 1]])
  287. ctx.update_end(1250)
  288. ctx.finalize() # inserts [1199, 1250]
  289. ctx.update_start(1299)
  290. ctx.finalize() # nothing
  291. ctx.update_end(1350)
  292. ctx.finalize() # nothing
  293. ctx.update_start(1400)
  294. ctx.insert(np.zeros((0,2)))
  295. ctx.update_end(1450)
  296. ctx.finalize()
  297. ctx.update_start(1500)
  298. ctx.insert(np.zeros((0,2)))
  299. ctx.update_end(1550)
  300. ctx.finalize()
  301. ctx.insert(np.zeros((0,2)))
  302. ctx.insert(np.zeros((0,2)))
  303. ctx.insert(np.zeros((0,2)))
  304. ctx.finalize()
  305. # Check everything
  306. eq_(info(), [(1, [100, 145]),
  307. (0, [175, 200]),
  308. (0, [300, 350]),
  309. (0, [400, 450]),
  310. (0, [500, 550]),
  311. (0, [1000, 1050]),
  312. (1, [1100, 1101]),
  313. (1, [1199, 1250]),
  314. (0, [1400, 1450]),
  315. (0, [1500, 1550]),
  316. ])
  317. # Clean up
  318. client.stream_remove("/empty/test")
  319. client.stream_destroy("/empty/test")
  320. client.close()