def compute_group(cls, data, scales, **params):
n = len(data)
if n < 3:
return pd.DataFrame()
weight = data.get('weight')
if params['trim']:
range_y = data['y'].min(), data['y'].max()
else:
range_y = scales.y.dimension()
dens = compute_density(data['y'], weight, range_y, **params)
dens['y'] = dens['x']
dens['x'] = np.mean([data['x'].min(), data['x'].max()])
# Compute width if x has multiple values
if len(np.unique(data['x'])) > 1:
dens['width'] = np.ptp(data['x']) * 0.9
return dens
python类ptp()的实例源码
def draw_group(data, panel_params, coord, ax, **params):
data = coord.transform(data, panel_params)
fill = to_rgba(data['fill'], data['alpha'])
color = to_rgba(data['color'], data['alpha'])
ranges = coord.range(panel_params)
# For perfect circles the width/height of the circle(ellipse)
# should factor in the dimensions of axes
bbox = ax.get_window_extent().transformed(
ax.figure.dpi_scale_trans.inverted())
ax_width, ax_height = bbox.width, bbox.height
factor = ((ax_width/ax_height) *
np.ptp(ranges.y)/np.ptp(ranges.x))
size = data.loc[0, 'binwidth'] * params['dotsize']
offsets = data['stackpos'] * params['stackratio']
if params['binaxis'] == 'x':
width, height = size, size*factor
xpos, ypos = data['x'], data['y'] + height*offsets
elif params['binaxis'] == 'y':
width, height = size/factor, size
xpos, ypos = data['x'] + width*offsets, data['y']
circles = []
for xy in zip(xpos, ypos):
patch = mpatches.Ellipse(xy, width=width, height=height)
circles.append(patch)
coll = mcoll.PatchCollection(circles,
edgecolors=color,
facecolors=fill)
ax.add_collection(coll)
def fit(self, X, y=None):
"""Fit it.
Parameters
----------
X : array, shape (n_epochs, n_times)
The data for one channel.
y : None
Redundant. Necessary to be compatible with sklearn
API.
"""
deltas = np.ptp(X, axis=1)
self.deltas_ = deltas
keep = deltas <= self.thresh
# XXX: actually go over all the folds before setting the min
# in skopt. Otherwise, may confuse skopt.
if self.thresh < np.min(np.ptp(X, axis=1)):
assert np.sum(keep) == 0
keep = deltas <= np.min(np.ptp(X, axis=1))
self.mean_ = _slicemean(X, keep, axis=0)
return self
def _vote_bad_epochs(self, epochs):
"""Each channel votes for an epoch as good or bad.
Parameters
----------
epochs : instance of mne.Epochs
The epochs object for which bad epochs must be found.
"""
n_epochs = len(epochs)
picks = _handle_picks(info=epochs.info, picks=self.picks)
drop_log = np.zeros((n_epochs, len(epochs.ch_names)))
bad_sensor_counts = np.zeros((len(epochs), ))
ch_names = [epochs.ch_names[p] for p in picks]
deltas = np.ptp(epochs.get_data()[:, picks], axis=-1).T
threshes = [self.threshes_[ch_name] for ch_name in ch_names]
for ch_idx, (delta, thresh) in enumerate(zip(deltas, threshes)):
bad_epochs_idx = np.where(delta > thresh)[0]
# TODO: combine for different ch types
bad_sensor_counts[bad_epochs_idx] += 1
drop_log[bad_epochs_idx, picks[ch_idx]] = 1
return drop_log, bad_sensor_counts
def extend_limits(values, fraction=0.10, tolerance=1e-2):
""" Extend the values of a list by a fractional amount """
values = np.array(values)
finite_indices = np.isfinite(values)
if np.sum(finite_indices) == 0:
raise ValueError("no finite values provided")
lower_limit, upper_limit = np.min(values[finite_indices]), np.max(values[finite_indices])
ptp_value = np.ptp([lower_limit, upper_limit])
new_limits = lower_limit - fraction * ptp_value, ptp_value * fraction + upper_limit
if np.abs(new_limits[0] - new_limits[1]) < tolerance:
if np.abs(new_limits[0]) < tolerance:
# Arbitrary limits, since we"ve just been passed zeros
offset = 1
else:
offset = np.abs(new_limits[0]) * fraction
new_limits = new_limits[0] - offset, offset + new_limits[0]
return np.array(new_limits)
def calculate_fractional_overlap(interest_range, comparison_range):
"""
Calculate how much of the range of interest overlaps with the comparison
range.
"""
if not (interest_range[-1] >= comparison_range[0] \
and comparison_range[-1] >= interest_range[0]):
return 0.0 # No overlap
elif (interest_range[0] >= comparison_range[0] \
and interest_range[-1] <= comparison_range[-1]):
return 1.0 # Total overlap
else:
# Some overlap. Which side?
if interest_range[0] < comparison_range[0]:
# Left hand side
width = interest_range[-1] - comparison_range[0]
else:
# Right hand side
width = comparison_range[-1] - interest_range[0]
return width/np.ptp(interest_range) # Fractional overlap
def update_roi_xy_size(self):
""" Update the cursor size showing the optimizer scan area for the XY image.
"""
hpos = self.roi_xy.pos()[0]
vpos = self.roi_xy.pos()[1]
hsize = self.roi_xy.size()[0]
vsize = self.roi_xy.size()[1]
hcenter = hpos + 0.5 * hsize
vcenter = vpos + 0.5 * vsize
if self.adjust_cursor_roi:
newsize = self._optimizer_logic.refocus_XY_size
else:
viewrange = self.xy_image.getViewBox().viewRange()
newsize = np.sqrt(np.sum(np.ptp(viewrange, axis=1)**2)) / 20
self.roi_xy.setSize([newsize, newsize])
self.roi_xy.setPos([hcenter - newsize / 2, vcenter - newsize / 2])
def update_roi_depth_size(self):
""" Update the cursor size showing the optimizer scan area for the X-depth image.
"""
hpos = self.roi_depth.pos()[0]
vpos = self.roi_depth.pos()[1]
hsize = self.roi_depth.size()[0]
vsize = self.roi_depth.size()[1]
hcenter = hpos + 0.5 * hsize
vcenter = vpos + 0.5 * vsize
if self.adjust_cursor_roi:
newsize_h = self._optimizer_logic.refocus_XY_size
newsize_v = self._optimizer_logic.refocus_Z_size
else:
viewrange = self.depth_image.getViewBox().viewRange()
newsize = np.sqrt(np.sum(np.ptp(viewrange, axis=1)**2)) / 20
newsize_h = newsize
newsize_v = newsize
self.roi_depth.setSize([newsize_h, newsize_v])
self.roi_depth.setPos([hcenter - newsize_h / 2, vcenter - newsize_v / 2])
def plane_fit(points, tolerance=None):
'''
Given a set of points, find an origin and normal using least squares
Arguments
---------
points: (n,3)
tolerance: how non-planar the result can be without raising an error
Returns
---------
C: (3) point on the plane
N: (3) normal vector
'''
C = points[0]
x = points - C
M = np.dot(x.T, x)
N = np.linalg.svd(M)[0][:,-1]
if not (tolerance is None):
normal_range = np.ptp(np.dot(N, points.T))
if normal_range > tol.planar:
log.error('Points have peak to peak of %f', normal_range)
raise ValueError('Plane outside tolerance!')
return C, N
def plot_epipolar_line(p1, p2, F, show_epipole=False):
""" Plot the epipole and epipolar line F*x=0
in an image given the corresponding points.
F is the fundamental matrix and p2 are the point in the other image.
"""
lines = np.dot(F, p2)
pad = np.ptp(p1, 1) * 0.01
mins = np.min(p1, 1)
maxes = np.max(p1, 1)
# epipolar line parameter and values
xpts = np.linspace(mins[0] - pad[0], maxes[0] + pad[0], 100)
for line in lines.T:
ypts = np.asarray([(line[2] + line[0] * p) / (-line[1]) for p in xpts])
valid_idx = ((ypts >= mins[1] - pad[1]) & (ypts <= maxes[1] + pad[1]))
plt.plot(xpts[valid_idx], ypts[valid_idx], linewidth=1)
plt.plot(p1[0], p1[1], 'ro')
if show_epipole:
epipole = compute_epipole(F)
plt.plot(epipole[0] / epipole[2], epipole[1] / epipole[2], 'r*')
def startModulation(self,
radiusInMilliRad,
frequencyInHz,
centerInMilliRad):
self._origTargetPosition= centerInMilliRad
self.stopModulation()
periodInSec= 1./ frequencyInHz
assert np.ptp(self._ctrl.getWaveGeneratorTableRate()) == 0, \
"wave generator table rate must be the same for every table"
wgtr= self._ctrl.getWaveGeneratorTableRate()[0]
timestep= self._ctrl.getServoUpdateTimeInSeconds() * wgtr
lengthInPoints= periodInSec/ timestep
peakOfTheSineCurve= self._milliRadToGcsUnits(
self.getTargetPosition() + radiusInMilliRad)
offsetOfTheSineCurve= self._milliRadToGcsUnits(
self.getTargetPosition() - radiusInMilliRad)
amplitudeOfTheSineCurve= peakOfTheSineCurve - offsetOfTheSineCurve
wavelengthOfTheSineCurveInPoints= periodInSec/ timestep
startPoint= np.array([0, 0.25])* wavelengthOfTheSineCurveInPoints
curveCenterPoint= 0.5* wavelengthOfTheSineCurveInPoints
self._ctrl.clearWaveTableData([1, 2, 3])
self._ctrl.setSinusoidalWaveform(
1, WaveformGenerator.CLEAR, lengthInPoints,
amplitudeOfTheSineCurve[0], offsetOfTheSineCurve[0],
wavelengthOfTheSineCurveInPoints, startPoint[0], curveCenterPoint)
self._ctrl.setSinusoidalWaveform(
2, WaveformGenerator.CLEAR, lengthInPoints,
amplitudeOfTheSineCurve[1], offsetOfTheSineCurve[1],
wavelengthOfTheSineCurveInPoints, startPoint[1], curveCenterPoint)
self._ctrl.setConnectionOfWaveTableToWaveGenerator([1, 2], [1, 2])
self._ctrl.setWaveGeneratorStartStopMode([1, 1, 0])
self._modulationEnabled= True
def compute_group(cls, data, scales, **params):
labels = ['x', 'y']
X = np.array(data[labels])
res = boxplot_stats(X, whis=params['coef'], labels=labels)[1]
try:
n = data['weight'].sum()
except KeyError:
n = len(data['y'])
if len(np.unique(data['x'])) > 1:
width = np.ptp(data['x']) * 0.9
else:
width = params['width']
if pdtypes.is_categorical(data['x']):
x = data['x'].iloc[0]
else:
x = np.mean([data['x'].min(), data['x'].max()])
d = {'ymin': res['whislo'],
'lower': res['q1'],
'middle': [res['med']],
'upper': res['q3'],
'ymax': res['whishi'],
'outliers': [res['fliers']],
'notchupper': res['med']+1.58*res['iqr']/np.sqrt(n),
'notchlower': res['med']-1.58*res['iqr']/np.sqrt(n),
'x': x,
'width': width,
'relvarwidth': np.sqrt(n)}
return pd.DataFrame(d)
def test_ptp(self):
a = [3, 4, 5, 10, -3, -5, 6.0]
assert_equal(np.ptp(a, axis=0), 15.0)
def _phampcheck(self, pha, amp, axis):
"""Check phase and amplitude values."""
# Shape checking :
if pha.ndim != amp.ndim:
raise ValueError("pha and amp must have the same number of "
"dimensions.")
# Force phase / amplitude to be at least (1, N) :
if (pha.ndim == 1) and (amp.ndim == 1):
pha = pha.reshape(1, -1)
amp = amp.reshape(1, -1)
axis = 1
# Check if the phase is in radians :
if np.ptp(pha) > 2 * np.pi:
raise ValueError("Your phase is probably in degrees and should be"
" converted in radians using either np.degrees or"
" np.deg2rad.")
# Check if the phase/amplitude have the same number of points on axis:
if pha.shape[axis] != amp.shape[axis]:
phan, ampn = pha.shape[axis], amp.shape[axis]
raise ValueError("The phase (" + str(phan) + ") and the amplitude "
"(" + str(ampn) + ") do not have the same number"
" of points on the specified axis (" +
str(axis) + ").")
# Force the phase to be in [-pi, pi] :
pha = (pha + np.pi) % (2 * np.pi) - np.pi
return pha, amp, axis
###########################################################################
# PROPERTIES
###########################################################################
# ----------- IDPAC -----------
def _postprocess_contours(self, index, times, freqs, salience):
"""Remove contours that are too short.
Parameters
----------
index : np.array
array of contour numbers
times : np.array
array of contour times
freqs : np.array
array of contour frequencies
salience : np.array
array of contour salience values
Returns
-------
index_pruned : np.array
Pruned array of contour numbers
times_pruned : np.array
Pruned array of contour times
freqs_pruned : np.array
Pruned array of contour frequencies
salience_pruned : np.array
Pruned array of contour salience values
"""
keep_index = np.ones(times.shape).astype(bool)
for i in set(index):
this_idx = (index == i)
if np.ptp(times[this_idx]) <= self.min_contour_len:
keep_index[this_idx] = False
return (index[keep_index], times[keep_index],
freqs[keep_index], salience[keep_index])
def test_ptp(self):
a = [3, 4, 5, 10, -3, -5, 6.0]
assert_equal(np.ptp(a, axis=0), 15.0)
def fit(self, X, y=None):
"""Fit it."""
if self.n_channels is None or self.n_times is None:
raise ValueError('Cannot fit without knowing n_channels'
' and n_times')
X = X.reshape(-1, self.n_channels, self.n_times)
deltas = np.array([np.ptp(d, axis=1) for d in X])
epoch_deltas = deltas.max(axis=1)
keep = epoch_deltas <= self.thresh
self.mean_ = _slicemean(X, keep, axis=0)
return self
def _get_epochs_interpolation(self, epochs, drop_log,
ch_type, verbose='progressbar'):
"""Interpolate the bad epochs."""
# 1: bad segment, # 2: interpolated
fix_log = drop_log.copy()
ch_names = epochs.ch_names
non_picks = np.setdiff1d(range(epochs.info['nchan']), self.picks)
interp_channels = list()
n_interpolate = self.n_interpolate[ch_type]
for epoch_idx in range(len(epochs)):
n_bads = drop_log[epoch_idx, self.picks].sum()
if n_bads == 0:
continue
else:
if n_bads <= n_interpolate:
interp_chs_mask = drop_log[epoch_idx] == 1
else:
# get peak-to-peak for channels in that epoch
data = epochs[epoch_idx].get_data()[0]
peaks = np.ptp(data, axis=-1)
peaks[non_picks] = -np.inf
# find channels which are bad by rejection threshold
interp_chs_mask = drop_log[epoch_idx] == 1
# ignore good channels
peaks[~interp_chs_mask] = -np.inf
# find the ordering of channels amongst the bad channels
sorted_ch_idx_picks = np.argsort(peaks)[::-1]
# then select only the worst n_interpolate channels
interp_chs_mask[
sorted_ch_idx_picks[n_interpolate:]] = False
fix_log[epoch_idx][interp_chs_mask] = 2
interp_chs = np.where(interp_chs_mask)[0]
interp_chs = [ch_name for idx, ch_name in enumerate(ch_names)
if idx in interp_chs]
interp_channels.append(interp_chs)
return interp_channels, fix_log
def normalizeData(X):
mean = []
data_range = []
mean.append(np.mean(X[:,1]))
mean.append(np.mean(X[:,2]))
data_range = np.ptp(X,axis=0)[-2:]
#print(mean,data_range)
for i in range(len(X)):
X[:,1][i] = (X[:,1][i] - float(mean[0]))/float(data_range[0])
X[:,2][i] = (X[:,2][i] - float(mean[1]))/float(data_range[1])
return X
def normalizeData(X):
mean = []
data_range = []
mean.append(np.mean(X[:,1]))
mean.append(np.mean(X[:,2]))
data_range = np.ptp(X,axis=0)[-2:]
#print(mean,data_range)
for i in range(len(X)):
X[:,1][i] = (X[:,1][i] - float(mean[0]))/float(data_range[0])
X[:,2][i] = (X[:,2][i] - float(mean[1]))/float(data_range[1])
return X
def FeatureScaling(X):
mean = []
data_range = []
X1 = np.zeros((len(X),X.shape[1]))
mean.append(np.mean(X[:,1]))
mean.append(np.mean(X[:,2]))
data_range = np.ptp(X,axis=0)[-2:]
#print(mean)
print(data_range)
for i in range(len(X)):
X1[:,0][i] = (X[:,0][i] - mean[0])/data_range[0]
X1[:,1][i] = (X[:,1][i] - mean[1])/data_range[1]
return X1
def neighbours(self, effective_temperature, surface_gravity, metallicity, N,
scales=None):
"""
Return indices of the `N`th-nearest neighbours in the grid. The three
parameters are scaled by the peak-to-peak range in the grid, unless
`scales` are indicates.
:param effective_temperature:
The effective temperature of the star.
:param surface_gravity:
The surface gravity of the star.
:param metallicity:
The metallicity of the star.
:param N:
The number of neighbouring indices to return.
:returns:
An array of length `N` that contains the indices of the closest
neighbours in the grid.
"""
point = np.array([effective_temperature, surface_gravity, metallicity])
if scales is None:
scales = np.ptp(self._grid, axis=0)
distance = np.sum(((self._grid - point)/scales)**2, axis=1)
return np.argsort(distance)[:N]
def nearest_neighbours(self, point, n):
"""
Return the indices of the n nearest neighbours to the point.
"""
stellar_parameters = _recarray_to_array(self.stellar_parameters)
distances = np.sum(((point - stellar_parameters) \
/ np.ptp(stellar_parameters, axis=0))**2, axis=1)
return distances.argsort()[:n]
def figure_mouse_pick(self, event):
"""
Trigger for when the mouse is used to select an item in the figure.
:param event:
The matplotlib event.
"""
ycol = "abundance"
xcol = {
self.ax_excitation_twin: "expot",
self.ax_line_strength_twin: "reduced_equivalent_width"
}[event.inaxes]
xscale = np.ptp(event.inaxes.get_xlim())
yscale = np.ptp(event.inaxes.get_ylim())
try:
distance = np.sqrt(
((self._state_transitions[ycol] - event.ydata)/yscale)**2 \
+ ((self._state_transitions[xcol] - event.xdata)/xscale)**2)
except AttributeError:
# Stellar parameters have not been measured yet
return None
index = np.nanargmin(distance)
# Because the state transitions are linked to the parent source model of
# the table view, we will have to get the proxy index.
proxy_index = self.table_view.model().mapFromSource(
self.proxy_spectral_models.sourceModel().createIndex(index, 0)).row()
self.table_view.selectRow(proxy_index)
return None
def normalize(vec):
"""
Given an input vector normalize the vector
Parameters
==========
vec : array_like
input vector to normalize
Returns
=======
out : array_like
normalized vector
Examples
========
>>> import spacepy.toolbox as tb
>>> tb.normalize([1,2,3])
[0.0, 0.5, 1.0]
"""
# check to see if vec is numpy array, this is fastest
if isinstance(vec, np.ndarray):
out = (vec - vec.min())/np.ptp(vec)
else:
vecmin = np.min(vec)
ptp = np.ptp(vec)
out = [(val - vecmin)/ptp for val in vec]
return out
test_analytics.py 文件源码
项目：PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 28
收藏 0
点赞 0
评论 0
def test_ptp(self):
N = 1000
arr = np.random.randn(N)
ser = Series(arr)
self.assertEqual(np.ptp(ser), np.ptp(arr))
# GH11163
s = Series([3, 5, np.nan, -3, 10])
self.assertEqual(s.ptp(), 13)
self.assertTrue(pd.isnull(s.ptp(skipna=False)))
mi = pd.MultiIndex.from_product([['a', 'b'], [1, 2, 3]])
s = pd.Series([1, np.nan, 7, 3, 5, np.nan], index=mi)
expected = pd.Series([6, 2], index=['a', 'b'], dtype=np.float64)
self.assert_series_equal(s.ptp(level=0), expected)
expected = pd.Series([np.nan, np.nan], index=['a', 'b'])
self.assert_series_equal(s.ptp(level=0, skipna=False), expected)
with self.assertRaises(ValueError):
s.ptp(axis=1)
s = pd.Series(['a', 'b', 'c', 'd', 'e'])
with self.assertRaises(TypeError):
s.ptp()
with self.assertRaises(NotImplementedError):
s.ptp(numeric_only=True)
def _get_indice(cls, w, flux, blue, red, band=None, unit='ew', degree=1,
**kwargs):
""" compute spectral index after continuum subtraction
Parameters
----------
w: ndarray (nw, )
array of wavelengths in AA
flux: ndarray (N, nw)
array of flux values for different spectra in the series
blue: tuple(2)
selection for blue continuum estimate
red: tuple(2)
selection for red continuum estimate
band: tuple(2), optional
select region in this band only.
default is band = (min(blue), max(red))
unit: str
`ew` or `mag` wether equivalent width or magnitude
degree: int (default 1)
degree of the polynomial fit to the continuum
Returns
-------
ew: ndarray (N,)
equivalent width array
"""
wi, fi = cls.continuum_normalized_region_around_line(w, flux, blue,
red, band=band,
degree=degree)
if unit in (0, 'ew', 'EW'):
return np.trapz(1. - fi, wi, axis=-1)
else:
m = np.trapz(fi, wi, axis=-1)
m = -2.5 * np.log10(m / np.ptp(wi))
return m
def test_basic(self):
a = [3, 4, 5, 10, -3, -5, 6.0]
assert_equal(np.ptp(a, axis=0), 15.0)
b = [[3, 6.0, 9.0],
[4, 10.0, 5.0],
[8, 3.0, 2.0]]
assert_equal(np.ptp(b, axis=0), [5.0, 7.0, 7.0])
assert_equal(np.ptp(b, axis=-1), [6.0, 6.0, 6.0])
def plot_cdf(x, copy=True, fractional=True, **kwargs):
"""
Add a log-log CCDF plot to the current axes.
Arguments
---------
x : array_like
The data to plot
copy : boolean
copy input array in a new object before sorting it. If data is a *very*
large, the copy can avoided by passing False to this parameter.
fractional : boolean
compress the data by means of fractional ranking. This collapses the
ranks from multiple, identical observations into their midpoint, thus
producing smaller figures. Note that the resulting plot will NOT be the
exact CCDF function, but an approximation.
Additional keyword arguments are passed to `matplotlib.pyplot.loglog`.
Returns a matplotlib axes object.
"""
N = float(len(x))
if copy:
x = x.copy()
x.sort()
if fractional:
t = []
for x, chunk in groupby(enumerate(x, 1), itemgetter(1)):
xranks, _ = zip(*list(chunk))
t.append((float(x), xranks[0] + np.ptp(xranks) / 2.0))
t = np.asarray(t)
else:
t = np.c_[np.asfarray(x), np.arange(N) + 1]
if 'ax' not in kwargs:
ax = plt.gca()
else:
ax = kwargs.pop('ax')
ax.loglog(t[:, 0], (N - t[:, 1]) / N, 'ow', **kwargs)
return ax
def test_integrate():
subslice = slice(100,200)
wvln = np.linspace(1000., 4000., 1024)
flux = np.zeros_like(wvln)
flux[subslice] = 1./np.ptp(wvln[subslice]) # so the integral is 1
s = Spectrum(wvln*u.angstrom, flux*u.erg/u.cm**2/u.angstrom)
# the integration grid is a sub-section of the full wavelength array
wvln_grid = s.wavelength[subslice]
i_flux = s.integrate(wvln_grid)
assert np.allclose(i_flux.value, 1.) # "close" because this is float comparison
