Skip to content
Snippets Groups Projects
Commit 50730c6b authored by Olivier Bertrand's avatar Olivier Bertrand
Browse files

remove backup files

parent 5034afbc
No related branches found
No related tags found
No related merge requests found
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 0 additions and 1415 deletions
doc/source/example/processing/testblubb.py~ deleted 100644 → 0
+ 0
− 22
View file @ 5034afbc
import numpy as np
import matplotlib.pyplot as plt
import os
import sys
sys.path.insert(0, '/media/luiza/Daten/Repos/toolbox-navigation/src/database')
from database import DataBaseLoad
sys.path.insert(0, '/media/luiza/Daten/Repos/toolbox-navigation/src/processing')
import processing
# 1) Connect to the database
mydb_filename = os.path.abspath('../database.db')
mydb = DataBaseLoad(mydb_filename)
# 2) Define the position-orinetation at which
# we want the image and get the scene
rowid = 5000
my_scene = processing.scene(mydb, rowid=rowid)
print(my_scene.shape)
print(my_scene.type)
print('elevation' in my_scene)
navipy/__init__.py~ deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
navipy/__pycache__/__init__.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/comparing/__init__.py~ deleted 100644 → 0
+ 0
− 143
View file @ 5034afbc
"""
The Place comparator list different methods to
compare a current place to a memorised place or
memorised places.
"""
import numpy as np
from navipy.processing.tools import is_ibpc, is_obpc
from navipy.processing.tools import check_scene
def simple_imagediff(current, memory):
"""Compute the difference between
the current and memorised place code
:param current: current place code
:param memory: memorised place code
:returns: the image difference
:rtype: float
..ref: Zeil, J., 2012. Visual homing: an insect perspective.
Current opinion in neurobiology
"""
if not isinstance(current, np.ndarray):
raise TypeError('current place code should be a numpy array')
if not isinstance(memory, np.ndarray):
raise TypeError('memory place code should be a numpy array')
if not np.all(current.shape == memory.shape):
raise Exception('memory and current place code should\
have the same shape')
check_scene(current)
check_scene(memory)
diff = current - memory
if is_ibpc(current):
return diff
elif is_obpc(current):
return diff
else:
raise TypeError('place code is neither an ibpc nor obpc')
def imagediff(current, memory):
"""Compute the root mean square difference between
the current and memorised place code
:param current: current place code
:param memory: memorised place code
:returns: the image difference
:rtype: float #array(1,4) for ibpc and float for obpc
..ref: Zeil, J., 2012. Visual homing: an insect perspective.
Current opinion in neurobiology
"""
simple_diff = simple_imagediff(current, memory)
diff = np.power(simple_diff, 2)
if is_ibpc(current):
return np.sqrt(diff.mean(axis=0).mean(axis=0)) # 1
elif is_obpc(current):
return np.sqrt(diff.mean(axis=0).mean(axis=0))
else:
raise TypeError('place code is neither an ibpc nor obpc')
def rot_imagediff(current, memory):
"""Compute the rotational image difference between
the current and memorised place code.
:param current: current place code
:param memory: memorised place code
:returns: the rotational image difference
:rtype: (np.ndarray)
..ref: Zeil, J., 2012. Visual homing: an insect perspective.
Current opinion in neurobiology
..note: assume that the image is periodic along the x axis
(the left-right axis)
"""
if not is_ibpc(current): # and not is_obpc(current):
raise TypeError('The current and memory place code\
should be image based')
if not is_ibpc(memory): # and not is_obpc(memory):
raise TypeError('The current and memory place code\
should be image based')
check_scene(current)
check_scene(memory)
ridf = np.zeros((current.shape[1], current.shape[2]))
for azimuth_i in range(0, current.shape[1]):
rot_im = np.roll(current, azimuth_i, axis=1)
ridf[azimuth_i, :] = np.squeeze(imagediff(rot_im, memory)) # rot_im
return ridf
def diff_optic_flow(current, memory):
"""Computes the direction of motion from current
to memory by using the optic flow under the
constrain that the brightness is constant, (small movement),
using a taylor expansion and solving the equation:
0=I_t+\delta I*<u,v> or I_x+I_y+I_t=0
afterwards the aperture problem is solved by a
Matrix equation Ax=b, where x=(u,v) and
A=(I_x,I_y) and b = I_t
:param current: current place code
:param memory: memorised place code
:returns: a directional vectors
:rtype: (np.ndarray)
..ref: aperture problem:
Shimojo, Shinsuke, Gerald H. Silverman, and Ken Nakayama:
"Occlusion and the solution to the aperture problem for motion."
Vision research 29.5 (1989): 619-626.
optic flow:
Horn, Berthold KP, and Brian G. Schunck.:
"Determining optical flow."
Artificial intelligence 17.1-3 (1981): 185-203.
"""
if not is_ibpc(current): # and not is_obpc(current):
raise TypeError('The current and memory place code\
should be image based')
if not is_ibpc(memory): # and not is_obpc(memory):
raise TypeError('The current and memory place code\
should be image based')
check_scene(current)
check_scene(memory)
currroll = np.roll(current, 1, axis=1)
dx = current - currroll
memroll = np.roll(memory, 1, axis=1)
dy = memory - memroll
dy = np.reshape(dy, (np.prod(dy.shape), 1))
dx = np.reshape(dx, (np.prod(dx.shape), 1))
di = current - memory
di = np.reshape(di, (np.prod(di.shape), 1))
a_matrix = np.column_stack([dy, dx])
a_matrix_sqr = np.dot(np.transpose(a_matrix), a_matrix)
b_vector = np.dot(np.transpose(a_matrix), di)
res = np.linalg.solve(a_matrix_sqr, b_vector)
return res
def gradient(current, memory):
return 0
navipy/comparing/__pycache__/__init__.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/comparing/__pycache__/test.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/comparing/test.py~ deleted 100644 → 0
+ 0
− 236
View file @ 5034afbc
import unittest
import numpy as np
import navipy.database as database
import navipy.comparing as comparing
from navipy.processing.tools import is_numeric_array
import pkg_resources
class TestCase(unittest.TestCase):
def setUp(self):
self.mydb_filename = pkg_resources.resource_filename(
'navipy', 'resources/database.db')
self.mydb = database.DataBaseLoad(self.mydb_filename)
def test_imagediff_curr(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
curr2 = curr.copy()
curr2[3, 5, 2, 0] = np.nan
curr3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
curr3 = [curr3, curr3, curr3]
curr3 = np.array(curr3)
curr4 = np.zeros((3, 4, 5, 0))
# put useless stuff here
with self.assertRaises(ValueError):
comparing.imagediff(curr2, mem)
with self.assertRaises(Exception):
comparing.imagediff(curr3, mem)
with self.assertRaises(Exception):
comparing.imagediff(curr4, mem)
# should be working -> check if result is correct
for s in [curr]:
diff = comparing.imagediff(s, mem)
# self.assertTrue(diff.shape[1] == 1)
self.assertTrue(diff.shape[0] > 0)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
def test_imagediff_memory(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
mem2 = curr.copy()
mem2[3, 5, 2] = np.nan
mem3 = [[1, 2], [1, 2], [1, 2]]
mem3 = [mem3, mem3, mem3]
mem3 = np.array(mem3)
mem4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.imagediff(curr, mem2)
with self.assertRaises(Exception):
comparing.imagediff(curr, mem3)
with self.assertRaises(Exception):
comparing.imagediff(curr, mem4)
# should be working -> check if result is correct
for s in [mem]:
diff = comparing.imagediff(curr, s)
self.assertFalse(diff.shape[0] <= 0)
# self.assertTrue(diff.shape[1] == 1)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
def test_rot_imagediff_curr(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
curr2 = curr.copy()
curr2[3, 5, 2] = np.nan
curr3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
curr3 = [curr3, curr3, curr3]
curr3 = np.array(curr3)
curr4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.rot_imagediff(curr2, mem)
with self.assertRaises(Exception):
comparing.rot_imagediff(curr3, mem)
with self.assertRaises(Exception):
comparing.rot_imagediff(curr4, mem)
# should be working -> check if result is correct
for s in [curr]:
diff = comparing.rot_imagediff(s, mem)
self.assertFalse(diff.shape[0] <= 0)
self.assertTrue(diff.shape[1] == 4)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
def test_rotimagediff_memory(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
mem2 = curr.copy()
mem2[3, 5, 2] = np.nan
mem3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
mem3 = [mem3, mem3, mem3]
mem3 = np.array(mem3)
mem4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.rot_imagediff(curr, mem2)
with self.assertRaises(Exception):
comparing.rot_imagediff(curr, mem3)
with self.assertRaises(Exception):
comparing.rot_imagediff(curr, mem4)
# should be working -> check if result is correct
for s in [mem]:
diff = comparing.rot_imagediff(curr, s)
self.assertFalse(diff.shape[0] <= 0)
self.assertTrue(diff.shape[1] == 4)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
def test_simple_imagediff_curr(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
curr2 = curr.copy()
curr2[3, 5, 2] = np.nan
curr3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
curr3 = [curr3, curr3, curr3]
curr3 = np.array(curr3)
curr4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.simple_imagediff(curr2, mem)
with self.assertRaises(Exception):
comparing.simple_imagediff(curr3, mem)
with self.assertRaises(Exception):
comparing.simple_imagediff(curr4, mem)
# should be working -> check if result is correct
for s in [curr]:
diff = comparing.simple_imagediff(s, mem)
self.assertFalse(diff.shape[0] <= 0)
self.assertTrue(diff.shape[1] > 0)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
self.assertTrue(diff.shape[2] == 4)
# self.assertTrue(diff.shape[3] == 1)
def test_simple_imagediff_mem(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
mem2 = curr.copy()
mem2[3, 5, 2] = np.nan
mem3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
mem3 = [mem3, mem3, mem3]
mem3 = np.array(mem3)
mem4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.simple_imagediff(curr, mem2)
with self.assertRaises(Exception):
comparing.simple_imagediff(curr, mem3)
with self.assertRaises(Exception):
comparing.simple_imagediff(curr, mem4)
# should be working -> check if result is correct
for s in [mem]:
diff = comparing.simple_imagediff(curr, s)
self.assertFalse(diff.shape[0] <= 0)
self.assertTrue(diff.shape[1] > 0)
self.assertFalse(np.any(np.isnan(diff)))
self.assertTrue(is_numeric_array(diff))
self.assertTrue(diff.shape[2] == 4)
# self.assertTrue(diff.shape[3] == 1)
def test_diff_optic_flow_memory(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
mem2 = curr.copy()
mem2[3, 5, 2] = np.nan
mem3 = [[1, 2, 3], [1, 2, 3], [1, 2, 3]]
mem3 = [mem3, mem3, mem3]
mem3 = np.array(mem3)
mem4 = np.zeros((3, 4, 5))
with self.assertRaises(ValueError):
comparing.diff_optic_flow(curr, mem2)
with self.assertRaises(Exception):
comparing.diff_optic_flow(curr, mem3)
with self.assertRaises(Exception):
comparing.diff_optic_flow(curr, mem4)
# should be working -> check if result is correct
for s in [mem]:
vec = comparing.diff_optic_flow(curr, s)
self.assertFalse(vec.shape[1] == (1, 2))
self.assertFalse(np.any(np.isnan(vec)))
self.assertTrue(is_numeric_array(vec))
def test_diff_optic_flow_curr(self):
curr = self.mydb.read_image(rowid=1)
mem = self.mydb.read_image(rowid=2)
curr = np.expand_dims(curr, axis=3)
mem = np.expand_dims(mem, axis=3)
curr2 = curr.copy()
curr2[3, 5, 2] = np.nan
curr3 = [[1, 2], [1, 2], [1, 2]]
curr3 = [curr3, curr3, curr3]
curr3 = np.array(curr3)
curr4 = np.zeros((3, 4, 5, 1))
with self.assertRaises(ValueError):
comparing.diff_optic_flow(curr2, mem)
with self.assertRaises(Exception):
comparing.diff_optic_flow(curr3, mem)
with self.assertRaises(Exception):
comparing.diff_optic_flow(curr4, mem)
# should be working -> check if result is correct
for s in [mem]:
vec = comparing.diff_optic_flow(s, curr)
self.assertFalse(vec.shape[1] == (1, 2))
self.assertFalse(np.any(np.isnan(vec)))
self.assertTrue(is_numeric_array(vec))
if __name__ == '__main__':
unittest.main()
navipy/database/__init__.py~ deleted 100644 → 0
+ 0
− 623
View file @ 5034afbc
This diff is collapsed.
navipy/database/__pycache__/__init__.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/database/__pycache__/database.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/__pycache__/__init__.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/__pycache__/agent.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/__pycache__/maths.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/__pycache__/test_agent.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/__pycache__/test_maths.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/moving/agent.py~ deleted 100644 → 0
+ 0
− 361
View file @ 5034afbc
"""
"""
import numpy as np
import pandas as pd
import networkx as nx
import multiprocessing
from multiprocessing import Queue, JoinableQueue, Process
import inspect
try:
from pandas.api.types import is_numeric_dtype
except ImportError:
from pandas.core.common import is_numeric_dtype
from navipy.database import DataBaseLoad
import navipy.moving.maths as navimomath
def defaultcallback(database, posorients):
raise NameError('No Callback')
class AbstractAgent():
"""
An abtract class for agent
"""
def __init__(self,
database_filename,
memory_friendly=False):
self.db = DataBaseLoad(database_filename)
self.dbname = database_filename
if memory_friendly:
self.__posorients = None
else:
self.__posorients = self.db.posorients
# set mode of motion
mode_move = {'mode': 'on_cubic_grid',
'param': {'grid_spacing':
pd.Series(data=1,
index=['dx', 'dy', 'dz'])}}
self.mode_of_motion = mode_move
@property
def posorients(self):
toreturn = self.__posorients
if toreturn is not None:
toreturn = toreturn.copy()
return toreturn
@property
def mode_of_motion(self):
"""
"""
toreturn = self.__mode_move
toreturn['describe'] = \
navimomath.mode_moves_supported()[
self.__mode_move['mode']]['describe']
return toreturn
@mode_of_motion.setter
def mode_of_motion(self, mode):
"""
"""
if not isinstance(mode, dict):
raise TypeError('Mode is not a dictionary')
if 'mode' not in mode:
raise KeyError("'mode' is not a key of mode")
if 'param' not in mode:
raise KeyError("'param' is not a key of mode")
if mode['mode'] in navimomath.mode_moves_supported().keys():
for param in navimomath.mode_moves_supported()[
mode['mode']]['param']:
if param not in mode['param']:
raise KeyError(
"'{}' is not in mode['param']".format(param))
self.__mode_move = mode
else:
raise ValueError('mode is not supported')
def abstractmove(self, posorients_vel):
if isinstance(posorients_vel, pd.Series) is False:
raise TypeError('posorients_vel should be a pandas Series')
for col in ['x', 'y', 'z', 'alpha_0', 'alpha_1', 'alpha_2',
'dx', 'dy', 'dz', 'dalpha_0', 'dalpha_1', 'dalpha_2']:
if col not in posorients_vel.index:
raise KeyError(
'posorients_vel should have {} as index'.format(col))
# Compute the next position
posorients_vel = navimomath.next_pos(
posorients_vel,
move_mode=self.__mode_move['mode'],
move_param=self.__mode_move['param'])
# Compute the closest possible position
if posorients_vel is None:
tmp = navimomath.closest_pos_memory_friendly(
posorients_vel,
self.db)
posorients_vel[['x', 'y', 'z',
'alpha_0', 'alpha_1', 'alpha_2']] = tmp
posorients_vel.name = tmp.name
else:
tmp = navimomath.closest_pos(
posorients_vel,
self.__posorients)
posorients_vel[['x', 'y', 'z',
'alpha_0', 'alpha_1', 'alpha_2']] = tmp
posorients_vel.name = tmp.name
return posorients_vel
class Single(AbstractAgent, Process):
def __init__(self,
database_filename,
initial_condition,
memory_friendly=False,
posorients_queue=None,
results_queue=None):
if (posorients_queue is not None) and (results_queue is not None):
multiprocessing.Process.__init__(self)
AbstractAgent.__init__(self, database_filename,
memory_friendly)
self.__posorientvel = pd.Series(
data=0,
index=['x', 'y', 'z',
'alpha_0', 'alpha_1', 'alpha_2',
'dx', 'dy', 'dz',
'dalpha_0', 'dalpha_1', 'dalpha_2'],
dtype=np.float)
if isinstance(initial_condition, pd.Series):
if is_numeric_dtype(initial_condition):
common_id = list(set(initial_condition.index).intersection(
self.__posorientvel.index))
self.__posorientvel.loc[common_id] = \
initial_condition.loc[common_id]
else:
raise TypeError('vel should be numeric')
else:
raise TypeError('vel should be a pandas Series')
self.__posorients_queue = posorients_queue
self.__results_queue = results_queue
self.__callback_function = defaultcallback
def move(self):
# Compute the next position
tmp = self.__callback_function(database=self.db,
posorient=self.__posorientvel)
common_id = list(set(tmp.index).intersection(
self.__posorientvel.index))
self.__posorientvel.loc[common_id] = tmp.loc[common_id]
self.__posorientvel = self.abstractmove(self.__posorientvel)
def fly(self, nsteps):
"""move until either speed is null, or nsteps has been reached"""
prev_move = self.__posorientvel
for stepi in range(nsteps):
self.move()
if prev_move.equals(self.__posorientvel):
break
prev_move = self.__posorientvel
def run(self):
""" Only supported when multiprocess"""
if self.__posorients_queue is None or self.__results_queue is None:
raise NameError('Single agent class has not be inititialised '
+ 'with multiprocessing suppport')
proc_name = self.name
print('Process {} started'.format(proc_name))
while True:
start_posorient = self.__posorients_queue.get(timeout=1)
if start_posorient is None:
# Poison pill means shutdown)
break
common_id = list(set(start_posorient.index).intersection(
self.__posorientvel.index))
self.__posorientvel.loc[common_id] = start_posorient.loc[common_id]
self.move()
next_posorient = self.__posorientvel
self.__posorients_queue.task_done()
self.__results_queue.put((start_posorient, next_posorient))
self.__posorients_queue.task_done()
print('Process {} done'.format(proc_name))
@property
def callback_function(self):
return inspect.getsourcelines(self.__callback_function)
@callback_function.setter
def callback_function(self, callback_function):
self.__callback_function = callback_function
@property
def position(self):
return self.__posorientvel.loc[['x', 'y', 'z']]
@property
def velocity(self):
return self.__posorientvel.loc[['dx', 'dy', 'dz']]
@property
def orientation(self):
return self.__posorientvel.loc[['alpha_0', 'alpha_1', 'alpha_2']]
@property
def angular_velocity(self):
return self.__posorientvel.loc[['dalpha_0', 'dalpha_1', 'dalpha_2']]
class Multi(AbstractAgent):
def __init__(self, database_filename):
super().__init__(database_filename, False)
# Init the graph
self.__graph = nx.DiGraph()
for row_id, posor in self.db.posorients.iterrows():
posor.name = row_id
self.__graph.add_node(row_id,
posorient=posor)
@property
def graph(self):
return self.__graph
@graph.setter
def graph(self, graph):
if isinstance(graph, nx.DiGraph) is False:
raise TypeError('graph is not a nx.DiGraph')
self.__graph = graph.copy()
self.check_graph()
def build_graph(self, callback_function,
ncpu=5,
timeout=1):
# Build a list of nodes
results_edges = []
posorients_queue = JoinableQueue()
results_queue = Queue()
for node in self.__graph.nodes:
posorients_queue.put(self.__graph.nodes[node]['posorient'])
initpos = 0 * self.__graph.nodes[node]['posorient']
# Start ndatabase loader
num_agents = ncpu
agents = [Single(self.dbname,
initial_condition=initpos,
memory_friendly=False,
posorients_queue=posorients_queue,
results_queue=results_queue)
for _ in range(num_agents)]
for w in agents:
w.callback_function = callback_function
w.start()
# Add a poison pill for each agent
for _ in range(num_agents):
posorients_queue.put(None)
# Wait for all of the tasks to finish
# posorients_queue.join()
for _ in range(nx.number_of_nodes(self.__graph)):
result = results_queue.get(timeout=timeout)
results_edges.append((result[0].name,
result[1].name))
# print(results_edges[-1])
self.__graph.add_edges_from(results_edges)
self.check_graph()
def check_graph(self):
self.check_single_target()
def check_single_target(self):
for node in self.__graph.nodes:
# not connected -> loop not ran
for count, _ in enumerate(self.__graph.neighbors(node)):
# count == 0 -> connected to one node
# count == 1 -> connected to two nodes
if count > 0:
raise ValueError(
'Node {} leads to several locations'.format(node))
def find_attractors(self):
"""Return a list of node going to each attractor in a graph
"""
attractors = list()
for attractor in nx.attracting_components(self.__graph):
att = dict()
att['attractor'] = attractor
attractors.append(att)
return attractors
def find_attractors_sources(self, attractors=None):
"""Find all sources going to each attractors
"""
if attractors is None:
attractors = self.find_attractors()
if isinstance(attractors, list) is False:
raise TypeError('Attractors should be a list of dict')
elif len(attractors) == 0:
raise ValueError('No attractors found')
# Check attractor
for att in attractors:
keyatt = att.keys()
if 'attractor' not in keyatt:
raise ValueError(
'Each attractors should contain the key attractor')
# Calculate connection
for att_i, att in enumerate(attractors):
# [0] because one node of the attractor is enough
# all other node of the attractor are connected to this one
target = list(att['attractor'])[0]
attractors[att_i]['paths'] = \
nx.shortest_path(self.graph, target=target)
attractors[att_i]['sources'] = \
list(attractors[att_i]['paths'].keys())
return attractors
def catchment_area(self, attractors=None):
"""Return the catchment area for attractors
"""
if attractors is None:
attractors = self.find_attractors_sources()
if isinstance(attractors, list) is False:
raise TypeError('Attractors should be a list of dict')
elif len(attractors) == 0:
raise ValueError('No attractors found')
# Check attractor
for att in attractors:
keyatt = att.keys()
if 'sources' not in keyatt:
raise ValueError(
'Each attractors should contains a list of sources')
return [len(att['sources']) for att in attractors]
def reach_goals(self, goals):
""" Return all paths to the goals """
return nx.shortest_path(self.__graph, target=goals)
def neighboring_nodes(self, target):
""" Return the nodes going to the target """
# Reverse graph because nx.neighbors give the end node
# and we want to find the start node going to target
# not where target goes.
tmpgraph = self.__graph.reverse(copy=True)
neighbors = tmpgraph.neighbors(target)
return neighbors
navipy/moving/maths.py~ deleted 100644 → 0
+ 0
− 30
View file @ 5034afbc
"""
Mathematical computation are done in this module. Those involve mostly
geometry, and predefined grids shapes
"""
import numpy as np
import pandas as pd
def next_pos(motion_vec, grid_spacing=1, grid_mode='cubic'):
"""
"""
supported_grid_mode = ['cubic',
None]
assert isinstance(motion_vec, pd.Series),\
'motion vector must be a pandas Series'
assert grid_mode in supported_grid_mode,
'grid mode must is not supported {}'.format(grid_mode)
speed = motion_vec.loc[['dx', 'dy', 'dz']]
position = motion_vec.loc[['x', 'y', 'z']]
if grid_mode == 'cubic':
speed /= np.linalg.norm(speed)
scaling = grid_spacing / (2 * np.sin(np.pi / 8))
elif grid_mode == None:
scaling = 1
else:
raise ValueError('grid_mode is not supported')
return position + speed.rename({'dx': 'x', 'dy': 'y', 'dz': 'z'}) * scaling
navipy/processing/__pycache__/Scene_processing.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/processing/__pycache__/__init__.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
navipy/processing/__pycache__/constants.cpython-35.pyc deleted 100644 → 0
+ 0
− 0
View file @ 5034afbc
File deleted
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment