Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support for optional edge weights #390

Merged
merged 3 commits into from
Jun 23, 2017
Merged

Support for optional edge weights #390

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eefe169
Add support for optional edge weights
jbcrail Jun 23, 2017
dcf7d2f
Fix flake8 warnings
jbcrail Jun 23, 2017
0a881a3
Fix future warnings
jbcrail Jun 23, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 28 additions & 27 deletions datashader/bundling.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ def resample_edge(segments, min_segment_length, max_segment_length):
change, total_resamples = calculate_length(segments, min_segment_length, max_segment_length)
if not change:
return segments
resampled = np.empty((total_resamples, 2))
resampled = np.empty((total_resamples, 3))
resample_segment(segments, resampled, min_segment_length, max_segment_length)
return resampled

Expand Down Expand Up @@ -166,7 +166,7 @@ def draw_to_surface(edge_segments, bandwidth, accuracy):
img = np.zeros((accuracy + 1, accuracy + 1))
for segments in edge_segments:
for point in segments:
img[int(point[0] * accuracy), int(point[1] * accuracy)] += 1
img[int(point[0] * accuracy), int(point[1] * accuracy)] += point[2]
return gaussian(img, sigma=bandwidth / 2)


Expand All @@ -188,9 +188,10 @@ def _convert_graph_to_edge_segments(nodes, edges):
Merge graph dataframes into a list of edge segments.

Given a graph defined as a pair of dataframes (nodes and edges), the
nodes (id, coordinates) and edges (id, source, target) are joined by
node id to create a single dataframe with each source/target of an
edge replaced with the respective coordinates.
nodes (id, coordinates) and edges (id, source, target, weight) are
joined by node id to create a single dataframe with each source/target
of an edge (including its optional weight) replaced with the respective
coordinates.

All node points are normalized to the range (0, 1) using min-max
scaling.
Expand All @@ -209,11 +210,14 @@ def minmax_scale(series):
df = pd.merge(nodes, df, left_index=True, right_on=['target'])
df = df.rename(columns={'x': 'dst_x', 'y': 'dst_y'})

df = df.filter(items=['src_x', 'src_y', 'dst_x', 'dst_y'])
if 'weight' not in edges:
df['weight'] = 1

df = df.filter(items=['src_x', 'src_y', 'dst_x', 'dst_y', 'weight'])

edge_segments = []
for edge in df.get_values():
segments = [[edge[0], edge[1]], [edge[2], edge[3]]]
segments = [[edge[0], edge[1], edge[4]], [edge[2], edge[3], edge[4]]]
edge_segments.append(np.array(segments))
return edge_segments

Expand All @@ -231,14 +235,13 @@ def _convert_edge_segments_to_dataframe(edge_segments):
def edge_iterator():
for edge in edge_segments:
yield edge
yield np.array([[np.nan, np.nan]])
yield np.array([[np.nan, np.nan, np.nan]])

df = DataFrame(np.concatenate(list(edge_iterator())))
df.columns = ['x', 'y']
df.columns = ['x', 'y', 'weight']
return df



class directly_connect_edges(param.ParameterizedFunction):
"""
Convert a graph into paths suitable for datashading.
Expand All @@ -251,7 +254,7 @@ class directly_connect_edges(param.ParameterizedFunction):
def __call__(self, nodes, edges):
"""
Convert a graph data structure into a path structure for plotting

Given a set of nodes (as a dataframe with a unique ID for each
node) and a set of edges (as a dataframe with with columns for the
source and destination IDs for each edge), returns a dataframe
Expand All @@ -271,7 +274,7 @@ class hammer_bundle(directly_connect_edges):
Breaks each edge into a path with multiple line segments, and
iteratively curves this path to bundle edges into groups.
"""

initial_bandwidth = param.Number(default=0.05,bounds=(0.0,None),doc="""
Initial value of the bandwidth....""")

Expand All @@ -284,7 +287,6 @@ class hammer_bundle(directly_connect_edges):
batch_size = param.Integer(default=20000,bounds=(1,None),doc="""
Number of edges to process together""")


tension = param.Number(default=0.3,bounds=(0,None),precedence=-0.5,doc="""
Exponential smoothing factor to use when smoothing""")

Expand All @@ -300,54 +302,53 @@ class hammer_bundle(directly_connect_edges):
max_segment_length = param.Number(default=0.016,bounds=(0,None),precedence=-0.5,doc="""
Maximum length (in data space?) for an edge segment""")


def __call__(self, nodes, edges, **params):
p = param.ParamOverrides(self,params)
p = param.ParamOverrides(self, params)

# Convert graph into list of edge segments
edges = _convert_graph_to_edge_segments(nodes, edges)

# This is simply to let the work split out over multiple cores
edge_batches = list(batches(edges, p.batch_size))

# This gets the edges split into lots of small segments
# Doing this inside a delayed function lowers the transmission overhead
edge_segments = [resample_edges(batch, p.min_segment_length, p.max_segment_length) for batch in edge_batches]

for i in range(p.iterations):
# Each step, the size of the 'blur' shrinks
bandwidth = p.initial_bandwidth * p.decay**(i + 1) * p.accuracy

# If it's this small, there won't be a change anyway
if bandwidth < 2:
break

# Draw the density maps and combine them
images = [draw_to_surface(segment, bandwidth, p.accuracy) for segment in edge_segments]
overall_image = sum(images)

gradients = get_gradients(overall_image)

# Move edges along the gradients and resample when necessary
# This could include smoothing to adjust the amount a graph can change
edge_segments = [advect_resample_all(gradients, segment, p.advect_iterations, p.accuracy, p.min_segment_length, p.max_segment_length)
for segment in edge_segments]

# Do a final resample to a smaller size for nicer rendering
edge_segments = [resample_edges(segment, p.min_segment_length, p.max_segment_length) for segment in edge_segments]

# Finally things can be sent for computation
edge_segments = compute(*edge_segments)

# Smooth out the graph
for i in range(10):
for batch in edge_segments:
smooth(batch, p.tension)

# Flatten things
new_segs = []
for batch in edge_segments:
new_segs.extend(batch)

# Convert list of edge segments to Pandas dataframe
return _convert_edge_segments_to_dataframe(new_segs)
38 changes: 22 additions & 16 deletions datashader/tests/test_bundling.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,31 +28,37 @@ def edges():
return edges_df


def assert_eq(a, b):
assert a.equals(b)


def test_directly_connect(nodes, edges):
# Expect four lines starting at center (0.5, 0.5) and terminating
# at a different corner and NaN
data = pd.DataFrame({'x': [0.5, 0.0, np.nan, 0.5, 1.0, np.nan,
0.5, 0.0, np.nan, 0.5, 1.0, np.nan],
'y': [0.5, 1.0, np.nan, 0.5, 1.0, np.nan,
0.5, 0.0, np.nan, 0.5, 0.0, np.nan]})
expected = pd.DataFrame(data)
data = pd.DataFrame({'x':
[0.5, 0.0, np.nan, 0.5, 1.0, np.nan,
0.5, 0.0, np.nan, 0.5, 1.0, np.nan],
'y':
[0.5, 1.0, np.nan, 0.5, 1.0, np.nan,
0.5, 0.0, np.nan, 0.5, 0.0, np.nan],
'weight':
[1.0, 1.0, np.nan, 1.0, 1.0, np.nan,
1.0, 1.0, np.nan, 1.0, 1.0, np.nan]})
expected = pd.DataFrame(data, columns=['x', 'y', 'weight'])

given = directly_connect_edges(nodes, edges)
assert_eq(given, expected)
assert given.equals(expected)


def test_hammer_bundle(nodes, edges):
# Expect four lines starting at center (0.5, 0.5) and terminating
# with NaN
data = pd.DataFrame({'x': [0.5, np.nan, 0.5, np.nan,
0.5, np.nan, 0.5, np.nan],
'y': [0.5, np.nan, 0.5, np.nan,
0.5, np.nan, 0.5, np.nan]})
expected = pd.DataFrame(data)
data = pd.DataFrame({'x':
[0.5, np.nan, 0.5, np.nan,
0.5, np.nan, 0.5, np.nan],
'y':
[0.5, np.nan, 0.5, np.nan,
0.5, np.nan, 0.5, np.nan],
'weight':
[1.0, np.nan, 1.0, np.nan,
1.0, np.nan, 1.0, np.nan]})
expected = pd.DataFrame(data, columns=['x', 'y', 'weight'])

df = hammer_bundle(nodes, edges)

Expand All @@ -62,4 +68,4 @@ def test_hammer_bundle(nodes, edges):
given.sort_index(inplace=True)
given.reset_index(drop=True, inplace=True)

assert_eq(given, expected)
assert given.equals(expected)
26 changes: 13 additions & 13 deletions datashader/tests/test_dask.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,18 +67,18 @@ def test_any():


def test_sum():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.sum('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.sum('i64')), out)
out = xr.DataArray(np.nansum(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nansum(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.sum('f32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.sum('f64')), out)


def test_min():
out = xr.DataArray(df.i64.reshape((2, 2, 5)).min(axis=2).astype('f8').T,
out = xr.DataArray(df.i64.values.reshape((2, 2, 5)).min(axis=2).astype('f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.min('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.min('i64')), out)
Expand All @@ -87,7 +87,7 @@ def test_min():


def test_max():
out = xr.DataArray(df.i64.reshape((2, 2, 5)).max(axis=2).astype('f8').T,
out = xr.DataArray(df.i64.values.reshape((2, 2, 5)).max(axis=2).astype('f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.max('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.max('i64')), out)
Expand All @@ -96,33 +96,33 @@ def test_max():


def test_mean():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).mean(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).mean(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.mean('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.mean('i64')), out)
out = xr.DataArray(np.nanmean(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanmean(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.mean('f32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.mean('f64')), out)


def test_var():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).var(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).var(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.var('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.var('i64')), out)
out = xr.DataArray(np.nanvar(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanvar(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.var('f32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.var('f64')), out)


def test_std():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).std(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).std(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.std('i32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.std('i64')), out)
out = xr.DataArray(np.nanstd(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanstd(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(ddf, 'x', 'y', ds.std('f32')), out)
assert_eq(c.points(ddf, 'x', 'y', ds.std('f64')), out)
Expand All @@ -147,9 +147,9 @@ def test_multiple_aggregates():
i32_count=ds.count('i32')))

f = lambda x: xr.DataArray(x, coords=coords, dims=dims)
assert_eq(agg.f64_std, f(np.nanstd(df.f64.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.f64_mean, f(np.nanmean(df.f64.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.i32_sum, f(df.i32.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T))
assert_eq(agg.f64_std, f(np.nanstd(df.f64.values.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.f64_mean, f(np.nanmean(df.f64.values.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.i32_sum, f(df.i32.values.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T))
assert_eq(agg.i32_count, f(np.array([[5, 5], [5, 5]], dtype='i4')))


Expand Down
24 changes: 12 additions & 12 deletions datashader/tests/test_pandas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,11 +61,11 @@ def test_any():


def test_sum():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.sum('i32')), out)
assert_eq(c.points(df, 'x', 'y', ds.sum('i64')), out)
out = xr.DataArray(np.nansum(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nansum(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.sum('f32')), out)
assert_eq(c.points(df, 'x', 'y', ds.sum('f64')), out)
Expand All @@ -81,7 +81,7 @@ def test_min():


def test_max():
out = xr.DataArray(df.i64.reshape((2, 2, 5)).max(axis=2).astype('f8').T,
out = xr.DataArray(df.i64.values.reshape((2, 2, 5)).max(axis=2).astype('f8').T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.max('i32')), out)
assert_eq(c.points(df, 'x', 'y', ds.max('i64')), out)
Expand All @@ -90,33 +90,33 @@ def test_max():


def test_mean():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).mean(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).mean(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.mean('i32')), out)
assert_eq(c.points(df, 'x', 'y', ds.mean('i64')), out)
out = xr.DataArray(np.nanmean(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanmean(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.mean('f32')), out)
assert_eq(c.points(df, 'x', 'y', ds.mean('f64')), out)


def test_var():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).var(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).var(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.var('i32')), out)
assert_eq(c.points(df, 'x', 'y', ds.var('i64')), out)
out = xr.DataArray(np.nanvar(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanvar(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.var('f32')), out)
assert_eq(c.points(df, 'x', 'y', ds.var('f64')), out)


def test_std():
out = xr.DataArray(df.i32.reshape((2, 2, 5)).std(axis=2, dtype='f8').T,
out = xr.DataArray(df.i32.values.reshape((2, 2, 5)).std(axis=2, dtype='f8').T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.std('i32')), out)
assert_eq(c.points(df, 'x', 'y', ds.std('i64')), out)
out = xr.DataArray(np.nanstd(df.f64.reshape((2, 2, 5)), axis=2).T,
out = xr.DataArray(np.nanstd(df.f64.values.reshape((2, 2, 5)), axis=2).T,
coords=coords, dims=dims)
assert_eq(c.points(df, 'x', 'y', ds.std('f32')), out)
assert_eq(c.points(df, 'x', 'y', ds.std('f64')), out)
Expand All @@ -141,9 +141,9 @@ def test_multiple_aggregates():
i32_count=ds.count('i32')))

f = lambda x: xr.DataArray(x, coords=coords, dims=dims)
assert_eq(agg.f64_std, f(np.nanstd(df.f64.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.f64_mean, f(np.nanmean(df.f64.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.i32_sum, f(df.i32.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T))
assert_eq(agg.f64_std, f(np.nanstd(df.f64.values.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.f64_mean, f(np.nanmean(df.f64.values.reshape((2, 2, 5)), axis=2).T))
assert_eq(agg.i32_sum, f(df.i32.values.reshape((2, 2, 5)).sum(axis=2, dtype='f8').T))
assert_eq(agg.i32_count, f(np.array([[5, 5], [5, 5]], dtype='i4')))


Expand Down