def deltasCSVWriter(self, name='ant'):
"Changes"
header = array([h.name[1:] for h in self.test.headers[:-2]])
oldRows = [r for r, p in zip(self.test._rows, self.pred) if p > 0]
delta = array([self.delta(t) for t in oldRows])
y = median(delta, axis=0)
yhi, ylo = percentile(delta, q=[75, 25], axis=0)
dat1 = sorted(
[(h, a, b, c) for h, a, b, c in zip(header, y, ylo, yhi)], key=lambda F: F[1])
dat = asarray([(d[0], n, d[1], d[2], d[3])
for d, n in zip(dat1, range(1, 21))])
with open('/Users/rkrsn/git/GNU-Plots/rkrsn/errorbar/%s.csv' % (name), 'w') as csvfile:
writer = csv.writer(csvfile, delimiter=' ')
for el in dat[()]:
writer.writerow(el)
# new = [self.newRow(t) for t in oldRows]
python类median()的实例源码
def find_history_data(self, row, history_dict=None,):
start_district_id = row.iloc[0]
time_id = row.iloc[1]
index = ['history_mean','history_median','history_mode','history_plus_mean','history_plus_median', 'history_plus_mode']
min_list = self.__get_historylist_from_dict(history_dict, start_district_id, time_id)
plus_list1 = self.__get_historylist_from_dict(history_dict, start_district_id, time_id-1)
plus_list2 = self.__get_historylist_from_dict(history_dict, start_district_id, time_id-2)
plus_list = np.array((plus_list1 + plus_list2 + min_list))
min_list = np.array(min_list)
res =pd.Series([min_list.mean(), np.median(min_list), mode(min_list)[0][0], plus_list.mean(), np.median(plus_list),mode(plus_list)[0][0]], index = index)
return res
return pd.Series(res, index = ['history_mean', 'history_mode', 'history_median'])
def get_7d_all(user_id):
data_paths = [
'./features/tensorflow_model/np_tiny_7_model/',
'./features/tensorflow_model/np_tiny_7_exp_model/',
'./features/tensorflow_model/np_tiny_7_filtered_model/',
'./features/tensorflow_model/np_tiny_7_filtered_exp_model/',
'./features/tensorflow_model/np_tiny_7_f2_model/',
'./features/tensorflow_model/np_tiny_7_f2_exp_model/',
]
def get_predict_val(dataset):
return dataset['y_p#%d'%user_id][-1]
def get_mid_val(day):
all_dataset = map(lambda path:pd.DataFrame.from_csv(path+'%d.csv'%day),data_paths)
val_list = map(get_predict_val,all_dataset)
val = np.median(val_list)
print (user_id,day,val)
return val
return map(get_mid_val,range(1,32))
def _nanmedian1d(arr1d, overwrite_input=False):
"""
Private function for rank 1 arrays. Compute the median ignoring NaNs.
See nanmedian for parameter usage
"""
c = np.isnan(arr1d)
s = np.where(c)[0]
if s.size == arr1d.size:
warnings.warn("All-NaN slice encountered", RuntimeWarning)
return np.nan
elif s.size == 0:
return np.median(arr1d, overwrite_input=overwrite_input)
else:
if overwrite_input:
x = arr1d
else:
x = arr1d.copy()
# select non-nans at end of array
enonan = arr1d[-s.size:][~c[-s.size:]]
# fill nans in beginning of array with non-nans of end
x[s[:enonan.size]] = enonan
# slice nans away
return np.median(x[:-s.size], overwrite_input=True)
def test_out(self):
mat = np.random.rand(3, 3)
nan_mat = np.insert(mat, [0, 2], np.nan, axis=1)
resout = np.zeros(3)
tgt = np.median(mat, axis=1)
res = np.nanmedian(nan_mat, axis=1, out=resout)
assert_almost_equal(res, resout)
assert_almost_equal(res, tgt)
# 0-d output:
resout = np.zeros(())
tgt = np.median(mat, axis=None)
res = np.nanmedian(nan_mat, axis=None, out=resout)
assert_almost_equal(res, resout)
assert_almost_equal(res, tgt)
res = np.nanmedian(nan_mat, axis=(0, 1), out=resout)
assert_almost_equal(res, resout)
assert_almost_equal(res, tgt)
def test_basic(self):
a0 = np.array(1)
a1 = np.arange(2)
a2 = np.arange(6).reshape(2, 3)
assert_equal(np.median(a0), 1)
assert_allclose(np.median(a1), 0.5)
assert_allclose(np.median(a2), 2.5)
assert_allclose(np.median(a2, axis=0), [1.5, 2.5, 3.5])
assert_equal(np.median(a2, axis=1), [1, 4])
assert_allclose(np.median(a2, axis=None), 2.5)
a = np.array([0.0444502, 0.0463301, 0.141249, 0.0606775])
assert_almost_equal((a[1] + a[3]) / 2., np.median(a))
a = np.array([0.0463301, 0.0444502, 0.141249])
assert_equal(a[0], np.median(a))
a = np.array([0.0444502, 0.141249, 0.0463301])
assert_equal(a[-1], np.median(a))
# check array scalar result
assert_equal(np.median(a).ndim, 0)
a[1] = np.nan
with warnings.catch_warnings(record=True) as w:
warnings.filterwarnings('always', '', RuntimeWarning)
assert_equal(np.median(a).ndim, 0)
assert_(w[0].category is RuntimeWarning)
def test_axis_keyword(self):
a3 = np.array([[2, 3],
[0, 1],
[6, 7],
[4, 5]])
for a in [a3, np.random.randint(0, 100, size=(2, 3, 4))]:
orig = a.copy()
np.median(a, axis=None)
for ax in range(a.ndim):
np.median(a, axis=ax)
assert_array_equal(a, orig)
assert_allclose(np.median(a3, axis=0), [3, 4])
assert_allclose(np.median(a3.T, axis=1), [3, 4])
assert_allclose(np.median(a3), 3.5)
assert_allclose(np.median(a3, axis=None), 3.5)
assert_allclose(np.median(a3.T), 3.5)
def getPreferenceList(preference,nSamplesOri,data_array):
"""
Input preference should be a numeric scalar, or a string of 'min' / 'median', or a list/np 1D array(length of samples).
Return preference list(same length as samples)
"""
# numeric value
if isinstance(preference, float) or isinstance(preference, int) or isinstance(preference, long):
preference_list=[float(preference)]*nSamplesOri
# str/unicode min/mean
elif isinstance(preference, basestring):
if str(preference)=='min':
preference=data_array.min()
elif str(preference)=='median':
preference=np.median(data_array)
else: #other string
raise ValueError("Preference should be a numeric scalar, or a string of 'min' / 'median',\
or a list/np 1D array(length of samples).\n Your input preference is: {0})".format(str(prefernce)))
preference_list=[preference]*nSamplesOri
# list or numpy array
elif (isinstance(preference, list) or isinstance(preference, np.ndarray)) and len(preference)==nSamplesOri:
preference_list=preference
else:
raise ValueError("Preference should be a numeric scalar, or a str of 'min' / 'median',\
or a list/np 1D array(length of samples).\n Your input preference is: {0})".format(str(prefernce)))
return preference_list
def medfilt(x, k):
'''
Apply a length-k median filter to a 1D array x.
Boundaries are extended by repeating endpoints.
Args:
x (numpy.array)
k (int)
Returns:
numpy.array
'''
assert k % 2 == 1, 'Median filter length must be odd.'
assert x.ndim == 1, 'Input must be one-dimensional.'
k2 = (k - 1) // 2
y = np.zeros((len(x), k), dtype=x.dtype)
y[:, k2] = x
for i in range(k2):
j = k2 - i
y[j:, i] = x[:-j]
y[:j, i] = x[0]
y[:-j, -(i+1)] = x[j:]
y[-j:, -(i+1)] = x[-1]
return np.median(y, axis=1)
def prctile(data, p_vals=[0, 25, 50, 75, 100], sorted_=False):
"""``prctile(data, 50)`` returns the median, but p_vals can
also be a sequence.
Provides for small samples better values than matplotlib.mlab.prctile,
however also slower.
"""
ps = [p_vals] if isscalar(p_vals) else p_vals
if not sorted_:
data = sorted(data)
n = len(data)
d = []
for p in ps:
fi = p * n / 100 - 0.5
if fi <= 0: # maybe extrapolate?
d.append(data[0])
elif fi >= n - 1:
d.append(data[-1])
else:
i = int(fi)
d.append((i + 1 - fi) * data[i] + (fi - i) * data[i + 1])
return d[0] if isscalar(p_vals) else d
def plotmatchdisthist(M, mas=True, nbins=100, doclf=True, color='b', **kwa):
import pylab as plt
if doclf:
plt.clf()
R = np.sqrt(M.dra_arcsec**2 + M.ddec_arcsec**2)
if mas:
R *= 1000.
rng = [0, M.rad*1000.]
else:
rng = [0, M.rad]
print 'Match distances: median', np.median(R), 'arcsec'
n,b,p = plt.hist(R, nbins, range=rng, histtype='step', color=color, **kwa)
if mas:
plt.xlabel('Match distance (mas)')
else:
plt.xlabel('Match distance (arcsec)')
plt.xlim(*rng)
return n,b,p
def analyze(self):
self.neighborgrid()
# just looking at up and left to avoid needless doubel calculations
slopes=np.concatenate((np.abs(self.left - self.center),np.abs(self.up - self.center)))
return '\n'.join(["%-15s: %.3f"%t for t in [
('height average', np.average(self.center)),
('height median', np.median(self.center)),
('height max', np.max(self.center)),
('height min', np.min(self.center)),
('height std', np.std(self.center)),
('slope average', np.average(slopes)),
('slope median', np.median(slopes)),
('slope max', np.max(slopes)),
('slope min', np.min(slopes)),
('slope std', np.std(slopes))
]]
)
def GetTransitTimes(file = 'ttv_kruse.dat'):
'''
'''
planet, _, time, dtime = np.loadtxt(os.path.join(TRAPPIST_DAT, file), unpack = True)
transit_times = [None for i in range(7)]
if file == 'ttv_kruse.dat':
for i in range(7):
inds = np.where(planet == i + 1)[0]
transit_times[i] = time[inds] + (2455000 - 2454833)
elif file == 'ttv_agol.dat':
for i in range(6):
inds = np.where(planet == i + 1)[0]
transit_times[i] = time[inds] + (2450000 - 2454833)
# Append a few extra for padding
pad = [transit_times[i][-1] + np.median(np.diff(transit_times[i])),
transit_times[i][-1] + 2 * np.median(np.diff(transit_times[i])),
transit_times[i][-1] + 3 * np.median(np.diff(transit_times[i]))]
transit_times[i] = np.append(transit_times[i], pad)
return PlanetProperty(transit_times)
graph_service.py 文件源码
项目:oss-github-analysis-project
作者: itu-oss-project-team
项目源码
文件源码
阅读 27
收藏 0
点赞 0
评论 0
def __reduce_vertex_metrics(self, metrics, is_property_map = True):
"""
Calculate mean, min, max and median of given vertex metrics
:param metrics: Metric values of vertexes
:param is_property_map: Is metrics PropertyMap (list otherwise)
:return: Dict of reduced metrics
"""
statistics = collections.OrderedDict()
if is_property_map:
metrics = metrics.get_array() # Get a numpy.ndarray subclass (PropertyArray)
statistics["mean"] = metrics.mean()
statistics["min"] = metrics.min()
statistics["max"] = metrics.max()
statistics["median"] = np.median(metrics)
return statistics
def reject_outliers(data, m = 2.):
d = np.abs(data - np.median(data))
mdev = np.median(d)
s = d/mdev if mdev else 0.
return data[s<m]
def obs_callback(self, msg):
cn0 = np.array([obs.cn0/4.0 for obs in msg.obs])
m = SignalStatus()
m.header.stamp = rospy.Time.now()
m.mean_cn0 = np.mean(cn0)
m.median_cn0 = np.median(cn0)
m.robust_mean_cn0 = np.mean(reject_outliers(cn0))
m.num_sats = len(msg.obs)
self.signal_pub.publish(m)
def median_color(obs, color, frame_coordinates=None):
color = np.array(color)
if frame_coordinates is not None:
(r1, r2), (c1, c2) = frame_coordinates
obs = obs[r1:r2, c1:c2]
indices = (obs == color).all(2).nonzero()
indices = np.array(indices)
if indices.size:
med = np.median(indices, axis=1)
return med.astype(np.int32)
return None
def median_color(obs, color, frame_coordinates=None):
color = np.array(color)
if frame_coordinates is not None:
(r1, r2), (c1, c2) = frame_coordinates
obs = obs[r1:r2, c1:c2]
indices = (obs == color).all(2).nonzero()
indices = np.array(indices)
if indices.size:
med = np.median(indices, axis=1)
return med.astype(np.int32)
return None
def median_color(obs, color, frame_coordinates=None):
color = np.array(color)
if frame_coordinates is not None:
(r1, r2), (c1, c2) = frame_coordinates
obs = obs[r1:r2, c1:c2]
indices = (obs == color).all(2).nonzero()
indices = np.array(indices)
if indices.size:
med = np.median(indices, axis=1)
return med.astype(np.int32)
return None