Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if not override_skimage and not dams:
return skimage.morphology.watershed(b, seeds, sel, None, mask)
elif seeded:
b = impose_minima(a, seeds.astype(bool), connectivity)
levels = unique(b)
a = pad(a, a.max()+1)
b = pad(b, b.max()+1)
ar = a.ravel()
br = b.ravel()
ws = pad(seeds, 0)
wsr = ws.ravel()
neighbors = build_neighbors_array(a, connectivity)
level_pixels = build_levels_dict(b)
if show_progress: wspbar = ip.StandardProgressBar('Watershed...')
else: wspbar = ip.NoProgressBar()
for i, level in ip.with_progress(enumerate(levels),
pbar=wspbar, length=len(levels)):
idxs_adjacent_to_labels = queue([idx for idx in level_pixels[level] if
any(wsr[neighbors[idx]])])
while len(idxs_adjacent_to_labels) > 0:
idx = idxs_adjacent_to_labels.popleft()
if wsr[idx] > 0: continue # in case we already processed it
nidxs = neighbors[idx] # neighbors
lnidxs = nidxs[(wsr[nidxs] != 0).astype(bool)] # labeled neighbors
adj_labels = unique(wsr[lnidxs])
if len(adj_labels) == 1 or len(adj_labels) > 1 and not dams:
# assign a label
wsr[idx] = wsr[lnidxs][ar[lnidxs].argmin()]
idxs_adjacent_to_labels.extend(nidxs[((wsr[nidxs] == 0) *
(br[nidxs] == level)).astype(bool) ])
return juicy_center(ws)
def build_edges_slow(self, idxs):
if self.show_progress:
idxs = ip.with_progress(idxs, title='Graph ', pbar=self.pbar)
for idx in idxs:
nodeid = self.watershed_r[idx]
ns = idx + self.steps
ns = ns[self.mask[ns]]
adj = self.watershed_r[ns]
adj = set(adj)
for v in adj:
if v == nodeid:
continue
if self.has_edge(nodeid, v):
self[nodeid][v]['boundary'].append(idx)
else:
self.add_edge(nodeid, v, boundary=[idx])
def compute_feature_caches(self):
"""Use the feature manager to compute node and edge feature caches.
Parameters
----------
None
Returns
-------
None
"""
for n in ip.with_progress(
self.nodes(), title='Node caches ', pbar=self.pbar):
self.node[n]['feature-cache'] = \
self.feature_manager.create_node_cache(self, n)
for n1, n2 in ip.with_progress(
self.edges(), title='Edge caches ', pbar=self.pbar):
self[n1][n2]['feature-cache'] = \
self.feature_manager.create_edge_cache(self, n1, n2)
def compute_feature_caches(self):
"""Use the feature manager to compute node and edge feature caches.
Parameters
----------
None
Returns
-------
None
"""
for n in ip.with_progress(
self.nodes(), title='Node caches ', pbar=self.pbar):
self.node[n]['feature-cache'] = \
self.feature_manager.create_node_cache(self, n)
for n1, n2 in ip.with_progress(
self.edges(), title='Edge caches ', pbar=self.pbar):
self[n1][n2]['feature-cache'] = \
self.feature_manager.create_edge_cache(self, n1, n2)