我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用numpy.loads()。
def get(self, ds_name, default = None): if self.reference_name is None: return default if ds_name in self.cache: return self.cache[ds_name] url = _url_path.format(self.reference_name, ds_name) try: resp = urllib.request.urlopen(url) except urllib.error.HTTPError: raise YTNoOldAnswer(url) else: for this_try in range(3): try: data = resp.read() except: time.sleep(0.01) else: # We were succesful break else: # Raise error if all tries were unsuccessful raise YTCloudError(url) # This is dangerous, but we have a controlled S3 environment rv = cPickle.loads(data) self.cache[ds_name] = rv return rv
def compare_image_lists(new_result, old_result, decimals): fns = [] for i in range(2): tmpfd, tmpname = tempfile.mkstemp(suffix='.png') os.close(tmpfd) fns.append(tmpname) num_images = len(old_result) assert(num_images > 0) for i in range(num_images): mpimg.imsave(fns[0], np.loads(zlib.decompress(old_result[i]))) mpimg.imsave(fns[1], np.loads(zlib.decompress(new_result[i]))) results = compare_images(fns[0], fns[1], 10**(-decimals)) if results is not None: if os.environ.get("JENKINS_HOME") is not None: tempfiles = [line.strip() for line in results.split('\n') if line.endswith(".png")] for fn in tempfiles: sys.stderr.write("\n[[ATTACHMENT|{}]]".format(fn)) sys.stderr.write('\n') assert_equal(results, None, results) for fn in fns: os.remove(fn)
def test_roundtrip(self): import pickle carray = np.array([[2, 9], [7, 0], [3, 8]]) DATA = [ carray, np.transpose(carray), np.array([('xxx', 1, 2.0)], dtype=[('a', (str, 3)), ('b', int), ('c', float)]) ] for a in DATA: assert_equal(a, pickle.loads(a.dumps()), err_msg="%r" % a)
def _loads(self, obj): if sys.version_info[0] >= 3: return np.loads(obj, encoding='latin1') else: return np.loads(obj) # version 0 pickles, using protocol=2 to pickle # version 0 doesn't have a version field
def pull_parameters(self): @gen.coroutine def _pull_parameters_sync(): if self._websock is None: raise TypeError('The websock is not initialized') pull_request = {'worker_id':self._id, 'worker_op':'pull'} self._websock.write_message(json.dumps(pull_request)) param_meta_msg = yield self._websock.read_message() # self._ioloop.add_future(param_meta_future, apply_parameters) param_meta = json.loads(param_meta_msg) return_id = param_meta['worker_id'] if return_id != self._id: raise ValueError('The returned ID does not match the current worker ID') param_version = param_meta['param_version'] if param_version <= self._last_version: return parameters = param_meta['parameters'] if parameters is None: raise ValueError('Parameters from server is None') # deserialized = {key:np.loads(parameters[key]) for key in parameters} # sutil.apply_parameters(self._sess, deserialized) np_parameters = {key:np.array(parameters[key]) for key in parameters} sutil.apply_parameters(self._sess, np_parameters) self._last_version = param_version self._ioloop.run_sync(_pull_parameters_sync)
def deserialize(self, serialized): return np.loads(serialized)
def loads(s): elements = s.split(';') shape = elements[0].split(',') width = int(shape[0]) height = int(shape[1]) flat = [] for e in elements[1:]: if e[0] == 'x': num = int(e[1:]) for i in range(0,num): flat.append(0) else: flat.append(float(e)) return np.reshape(flat, (width, height))
def get_block(self, block_name): return np.loads(self.rdb.get(block_name))
