Loading autocnet/graph/edge.py +16 −0 Original line number Diff line number Diff line Loading @@ -556,3 +556,19 @@ class Edge(dict, MutableMapping): pixel space """ self.overlap_latlon_coords, self["source_mbr"], self["destin_mbr"] = self.source.geodata.compute_overlap(self.destination.geodata, **kwargs) def get_matches(self): # pragma: no cover if self.matches.empty: return pd.DataFrame() match, _ = self.clean(clean_keys=list(self.masks.columns)) match = match[['source_image', 'source_idx', 'destination_image', 'destination_idx']] skps = self.get_keypoints('source', index=match.source_idx) skps.columns = ['source_x', 'source_y'] dkps = self.get_keypoints('destination', index=match.destination_idx) dkps.columns = ['destination_x', 'destination_y'] match = match.join(skps, on='source_idx') match = match.join(dkps, on='destination_idx') matches.append(match) return matches autocnet/graph/network.py +88 −2 Original line number Diff line number Diff line Loading @@ -24,6 +24,7 @@ from autocnet.graph.node import Node from autocnet.io import network as io_network from autocnet.vis.graph_view import plot_graph, cluster_plot # The total number of pixels squared that can fit into the keys number of GB of RAM for SIFT. MAXSIZE = {0:None, 2:6250, Loading Loading @@ -55,6 +56,7 @@ class CandidateGraph(nx.Graph): edge_attr_dict_factory = Edge def __init__(self, *args, basepath=None, **kwargs): # self.edge_attr_dict_factory = decorate_class(Edge, create_cg_updater(self), exclude=['clean', 'get_keypoints']) super(CandidateGraph, self).__init__(*args, **kwargs) self.graph['node_counter'] = 0 node_labels = {} Loading Loading @@ -84,6 +86,8 @@ class CandidateGraph(nx.Graph): self.graph['creationdate'] = strftime("%Y-%m-%d %H:%M:%S", gmtime()) self.graph['modifieddate'] = strftime("%Y-%m-%d %H:%M:%S", gmtime()) def get_matches(self, clean_keys=[], edges=[]): return self.apply_func_to_edges('get_matches') def __eq__(self, other): eq = True Loading Loading @@ -420,7 +424,8 @@ class CandidateGraph(nx.Graph): graph_mask_keys : list of keys in graph_masks """ if not isinstance(function, str): return_lis = [] if callable(function): function = function.__name__ for s, d, edge in self.edges_iter(data=True): Loading @@ -429,7 +434,58 @@ class CandidateGraph(nx.Graph): except: raise AttributeError(function, ' is not an attribute of Edge') else: func(*args, **kwargs) ret = func(*args, **kwargs) return_lis.append(ret) if any(return_lis): return return_lis def apply(self, function, on='edge',out=None, args=(), **kwargs): """ Applys a function to every node or edge, returns collected return values. TODO: Merge with apply_func_to_edges? Parameters ---------- function : callable Function to apply to graph. Should accept (id, data). on : string Whether to use nodes or edges. default is 'edge'. out : var Optionally put the output in a variable rather than returning it args : iterable Some iterable of positional arguments for function. kwargs : dict keyword args to pass into function. """ options = { 'edge' : self.edges_iter, 'edges' : self.edges_iter, 'e' : self.edges_iter, 0 : self.edges_iter, 'node' : self.nodes_iter, 'nodes' : self.nodes_iter, 'n' : self.nodes_iter, 1 : self.nodes_iter } if not callable(function): raise TypeError('{} is not callable.'.format(function)) res = [] for elem in options[on](data=True): res.append(function(elem, *args, **kwargs)) if out: out=res else: return res def symmetry_checks(self): ''' Loading Loading @@ -712,6 +768,36 @@ class CandidateGraph(nx.Graph): H.graph = self.graph return H # def nodes_iter(self, data=False): # s = super(CandidateGraph, self) # nodes = s.nodes_iter(data) # ret = [] # for n in nodes: # if data: # if n[0] in self.nodemask: # ret.append(n) # else: # if n in self.nodemask: # ret.append(n) # return iter(ret) # def edges_iter(self, nbunch=[], data=False, key=False): # s = super(CandidateGraph, self) # if not isinstance(nbunch, list): # nbunch = [nbunch] # # if nbunch: # nbunch = [node for node in nbunch if nbunch not in list(self.nodemask)] # else: # nbunch = list(self.nodemask) # # try: # return s.edges_iter(nbunch=nbunch, data=data) # except: # return s.edges_iter([self.node[node]['image_path'] for node in nbunch], data=data) def subgraph_from_matches(self): """ Returns a sub-graph where all edges have matches. Loading autocnet/graph/tests/test_network.py +15 −0 Original line number Diff line number Diff line Loading @@ -258,6 +258,21 @@ def test_is_complete(graph): assert False == incomplete_graph.is_complete() assert True == graph.is_complete() def test_apply(graph): def set_matches(x): s,d,e = x e.matches = ['fake', 'fake', 'fake'] def get_matches(x): s,d,e = x return e.matches graph.apply(set_matches) results = graph.apply(get_matches) for matches in results: assert len(matches) == 3 def test_footprints(geo_graph): # This is just testing the interface - should get a geodataframe back assert isinstance(geo_graph.footprints(), gpd.GeoDataFrame) autocnet/utils/tests/test_utils.py +35 −0 Original line number Diff line number Diff line Loading @@ -159,3 +159,38 @@ class TestUtils(unittest.TestCase): patchwork = Patchwork(a=1, b=2, c=3) self.assertEqual(patchwork.get(['a', 'b']), [1, 2]) self.assertEqual(patchwork.get('c'), 3) def test_decorate_class(self): class Test(object): def __init__(self): self.test = 'original' def get_test(self): return self.test def dec(func): return lambda x:'decorated' Dec_Test = utils.decorate_class(Test, dec) undecorated = Test() decorated = Dec_Test() self.assertEqual(undecorated.get_test(), 'original') self.assertEqual(decorated.get_test(), 'decorated') with self.assertRaises(Exception): utils.decorate_class(Test, 'Totally not a callable') def test_generate_decorator(self): def func_to_wrap(x): return x+1 def wrapper(): # Should be able to access run-time namespace return ret + 1, test decorator = utils.create_decorator(wrapper, test=0) wrapped_func = decorator(func_to_wrap) self.assertTrue(wrapped_func(1),2) autocnet/utils/utils.py +53 −0 Original line number Diff line number Diff line Loading @@ -366,3 +366,56 @@ def methodispatch(func): wrapper.register = dispatcher.register update_wrapper(wrapper, dispatcher) return wrapper def decorate_class(cls, decorator, exclude=[], *args, **kwargs): # pragma: no cover """ Decorates a class with a give docorator. Returns a subclass with dectorations applied Parameters ---------- cls : Class A class to be decorated decorator : callable callable to wrap cls's methods with exclude : list list of method names to exclude from being decorated args, kwargs : list, dict Parameters to pass into decorator """ if not callable(decorator): raise Exception('Decorator must be callable.') def decorate(cls): attributes = cls.__dict__.keys() for attr in attributes: # there's propably a better way to do this if callable(getattr(cls, attr)): name = getattr(cls, attr).__name__ if name[0] == '_' or name in exclude: continue setattr(cls, attr, decorator(getattr(cls, attr))) return cls # return decorated copy (i.e. a subclass with decorations) return decorate(type('cls_copy', cls.__bases__, dict(cls.__dict__))) def create_decorator(dec, **namespace): """ Create a decorator function using arbirary params. The objects passed in can be used in the body. Originally designed with the idea of automatically updating one object after the decorated object was modified. """ def decorator(func, *args, **kwargs): def wrapper(*args, **kwarg): for key in namespace.keys(): locals()[key] = namespace[key] ret = func(*args, **kwargs) exec(dec.__code__, locals(), globals()) if ret: return ret return wrapper return decorator Loading
autocnet/graph/edge.py +16 −0 Original line number Diff line number Diff line Loading @@ -556,3 +556,19 @@ class Edge(dict, MutableMapping): pixel space """ self.overlap_latlon_coords, self["source_mbr"], self["destin_mbr"] = self.source.geodata.compute_overlap(self.destination.geodata, **kwargs) def get_matches(self): # pragma: no cover if self.matches.empty: return pd.DataFrame() match, _ = self.clean(clean_keys=list(self.masks.columns)) match = match[['source_image', 'source_idx', 'destination_image', 'destination_idx']] skps = self.get_keypoints('source', index=match.source_idx) skps.columns = ['source_x', 'source_y'] dkps = self.get_keypoints('destination', index=match.destination_idx) dkps.columns = ['destination_x', 'destination_y'] match = match.join(skps, on='source_idx') match = match.join(dkps, on='destination_idx') matches.append(match) return matches
autocnet/graph/network.py +88 −2 Original line number Diff line number Diff line Loading @@ -24,6 +24,7 @@ from autocnet.graph.node import Node from autocnet.io import network as io_network from autocnet.vis.graph_view import plot_graph, cluster_plot # The total number of pixels squared that can fit into the keys number of GB of RAM for SIFT. MAXSIZE = {0:None, 2:6250, Loading Loading @@ -55,6 +56,7 @@ class CandidateGraph(nx.Graph): edge_attr_dict_factory = Edge def __init__(self, *args, basepath=None, **kwargs): # self.edge_attr_dict_factory = decorate_class(Edge, create_cg_updater(self), exclude=['clean', 'get_keypoints']) super(CandidateGraph, self).__init__(*args, **kwargs) self.graph['node_counter'] = 0 node_labels = {} Loading Loading @@ -84,6 +86,8 @@ class CandidateGraph(nx.Graph): self.graph['creationdate'] = strftime("%Y-%m-%d %H:%M:%S", gmtime()) self.graph['modifieddate'] = strftime("%Y-%m-%d %H:%M:%S", gmtime()) def get_matches(self, clean_keys=[], edges=[]): return self.apply_func_to_edges('get_matches') def __eq__(self, other): eq = True Loading Loading @@ -420,7 +424,8 @@ class CandidateGraph(nx.Graph): graph_mask_keys : list of keys in graph_masks """ if not isinstance(function, str): return_lis = [] if callable(function): function = function.__name__ for s, d, edge in self.edges_iter(data=True): Loading @@ -429,7 +434,58 @@ class CandidateGraph(nx.Graph): except: raise AttributeError(function, ' is not an attribute of Edge') else: func(*args, **kwargs) ret = func(*args, **kwargs) return_lis.append(ret) if any(return_lis): return return_lis def apply(self, function, on='edge',out=None, args=(), **kwargs): """ Applys a function to every node or edge, returns collected return values. TODO: Merge with apply_func_to_edges? Parameters ---------- function : callable Function to apply to graph. Should accept (id, data). on : string Whether to use nodes or edges. default is 'edge'. out : var Optionally put the output in a variable rather than returning it args : iterable Some iterable of positional arguments for function. kwargs : dict keyword args to pass into function. """ options = { 'edge' : self.edges_iter, 'edges' : self.edges_iter, 'e' : self.edges_iter, 0 : self.edges_iter, 'node' : self.nodes_iter, 'nodes' : self.nodes_iter, 'n' : self.nodes_iter, 1 : self.nodes_iter } if not callable(function): raise TypeError('{} is not callable.'.format(function)) res = [] for elem in options[on](data=True): res.append(function(elem, *args, **kwargs)) if out: out=res else: return res def symmetry_checks(self): ''' Loading Loading @@ -712,6 +768,36 @@ class CandidateGraph(nx.Graph): H.graph = self.graph return H # def nodes_iter(self, data=False): # s = super(CandidateGraph, self) # nodes = s.nodes_iter(data) # ret = [] # for n in nodes: # if data: # if n[0] in self.nodemask: # ret.append(n) # else: # if n in self.nodemask: # ret.append(n) # return iter(ret) # def edges_iter(self, nbunch=[], data=False, key=False): # s = super(CandidateGraph, self) # if not isinstance(nbunch, list): # nbunch = [nbunch] # # if nbunch: # nbunch = [node for node in nbunch if nbunch not in list(self.nodemask)] # else: # nbunch = list(self.nodemask) # # try: # return s.edges_iter(nbunch=nbunch, data=data) # except: # return s.edges_iter([self.node[node]['image_path'] for node in nbunch], data=data) def subgraph_from_matches(self): """ Returns a sub-graph where all edges have matches. Loading
autocnet/graph/tests/test_network.py +15 −0 Original line number Diff line number Diff line Loading @@ -258,6 +258,21 @@ def test_is_complete(graph): assert False == incomplete_graph.is_complete() assert True == graph.is_complete() def test_apply(graph): def set_matches(x): s,d,e = x e.matches = ['fake', 'fake', 'fake'] def get_matches(x): s,d,e = x return e.matches graph.apply(set_matches) results = graph.apply(get_matches) for matches in results: assert len(matches) == 3 def test_footprints(geo_graph): # This is just testing the interface - should get a geodataframe back assert isinstance(geo_graph.footprints(), gpd.GeoDataFrame)
autocnet/utils/tests/test_utils.py +35 −0 Original line number Diff line number Diff line Loading @@ -159,3 +159,38 @@ class TestUtils(unittest.TestCase): patchwork = Patchwork(a=1, b=2, c=3) self.assertEqual(patchwork.get(['a', 'b']), [1, 2]) self.assertEqual(patchwork.get('c'), 3) def test_decorate_class(self): class Test(object): def __init__(self): self.test = 'original' def get_test(self): return self.test def dec(func): return lambda x:'decorated' Dec_Test = utils.decorate_class(Test, dec) undecorated = Test() decorated = Dec_Test() self.assertEqual(undecorated.get_test(), 'original') self.assertEqual(decorated.get_test(), 'decorated') with self.assertRaises(Exception): utils.decorate_class(Test, 'Totally not a callable') def test_generate_decorator(self): def func_to_wrap(x): return x+1 def wrapper(): # Should be able to access run-time namespace return ret + 1, test decorator = utils.create_decorator(wrapper, test=0) wrapped_func = decorator(func_to_wrap) self.assertTrue(wrapped_func(1),2)
autocnet/utils/utils.py +53 −0 Original line number Diff line number Diff line Loading @@ -366,3 +366,56 @@ def methodispatch(func): wrapper.register = dispatcher.register update_wrapper(wrapper, dispatcher) return wrapper def decorate_class(cls, decorator, exclude=[], *args, **kwargs): # pragma: no cover """ Decorates a class with a give docorator. Returns a subclass with dectorations applied Parameters ---------- cls : Class A class to be decorated decorator : callable callable to wrap cls's methods with exclude : list list of method names to exclude from being decorated args, kwargs : list, dict Parameters to pass into decorator """ if not callable(decorator): raise Exception('Decorator must be callable.') def decorate(cls): attributes = cls.__dict__.keys() for attr in attributes: # there's propably a better way to do this if callable(getattr(cls, attr)): name = getattr(cls, attr).__name__ if name[0] == '_' or name in exclude: continue setattr(cls, attr, decorator(getattr(cls, attr))) return cls # return decorated copy (i.e. a subclass with decorations) return decorate(type('cls_copy', cls.__bases__, dict(cls.__dict__))) def create_decorator(dec, **namespace): """ Create a decorator function using arbirary params. The objects passed in can be used in the body. Originally designed with the idea of automatically updating one object after the decorated object was modified. """ def decorator(func, *args, **kwargs): def wrapper(*args, **kwarg): for key in namespace.keys(): locals()[key] = namespace[key] ret = func(*args, **kwargs) exec(dec.__code__, locals(), globals()) if ret: return ret return wrapper return decorator