(self, num_inputs, batch_size, input_dim, seed)
| 2265 | seed=st.integers(1, 10), |
| 2266 | ) |
| 2267 | def testBlobWeightedSum(self, num_inputs, batch_size, input_dim, seed): |
| 2268 | |
| 2269 | def get_blob_weighted_sum(): |
| 2270 | weights = [] |
| 2271 | for i in range(num_inputs): |
| 2272 | w_blob_name = 'blob_weighted_sum/w_{0}'.format(i) |
| 2273 | assert workspace.HasBlob(w_blob_name), ( |
| 2274 | "cannot fine blob {}".format(w_blob_name) |
| 2275 | ) |
| 2276 | w = workspace.FetchBlob(w_blob_name) |
| 2277 | weights.append(w) |
| 2278 | |
| 2279 | result = np.sum([ |
| 2280 | input_data[idx] * weights[idx] for idx in range(num_inputs) |
| 2281 | ], axis=0) |
| 2282 | return result |
| 2283 | |
| 2284 | np.random.seed(seed) |
| 2285 | expected_output_schema = schema.Scalar((np.float32, (input_dim,))) |
| 2286 | input_schema = schema.Tuple( |
| 2287 | *[expected_output_schema for _ in range(num_inputs)] |
| 2288 | ) |
| 2289 | input_data = [ |
| 2290 | np.random.random((batch_size, input_dim)).astype(np.float32) |
| 2291 | for _ in range(num_inputs) |
| 2292 | ] |
| 2293 | input_record = self.new_record(input_schema) |
| 2294 | schema.FeedRecord(input_record, input_data) |
| 2295 | |
| 2296 | # test output schema |
| 2297 | ws_output = self.model.BlobWeightedSum(input_record) |
| 2298 | self.assertEqual(len(self.model.layers), 1) |
| 2299 | assert schema.equal_schemas(ws_output, expected_output_schema) |
| 2300 | |
| 2301 | # test train net |
| 2302 | train_init_net, train_net = self.get_training_nets() |
| 2303 | workspace.RunNetOnce(train_init_net) |
| 2304 | workspace.RunNetOnce(train_net) |
| 2305 | output = workspace.FetchBlob(ws_output()) |
| 2306 | npt.assert_almost_equal(get_blob_weighted_sum(), output, decimal=5) |
| 2307 | |
| 2308 | self.run_train_net_forward_only() |
| 2309 | output = workspace.FetchBlob(ws_output()) |
| 2310 | npt.assert_almost_equal(get_blob_weighted_sum(), output, decimal=5) |
| 2311 | |
| 2312 | # test eval net |
| 2313 | eval_net = self.get_eval_net() |
| 2314 | workspace.RunNetOnce(eval_net) |
| 2315 | output = workspace.FetchBlob(ws_output()) |
| 2316 | npt.assert_almost_equal(get_blob_weighted_sum(), output, decimal=5) |
| 2317 | |
| 2318 | # test pred net |
| 2319 | pred_net = self.get_predict_net() |
| 2320 | workspace.RunNetOnce(pred_net) |
| 2321 | output = workspace.FetchBlob(ws_output()) |
| 2322 | npt.assert_almost_equal(get_blob_weighted_sum(), output, decimal=5) |
| 2323 | |
| 2324 | def testFeatureSparseToDenseGetAccessedFeatures(self): |
nothing calls this directly
no test coverage detected