More test coverage

Author: MartinBraquet

gcaa/core/allocation.py

@@ -0,0 +1,62 @@
+import numpy as np
+
+
+def remove_completed_tasks(pos_t: np.ndarray, ind):
+    """
+    pos_t: (nt, 2) array of task positions
+    ind: list/array of completed task indices (1-based in MATLAB)
+    """
+
+    # Convert MATLAB's 1-based indices → Python 0-based
+    ind_set = {ind}
+
+    # Keep only rows whose index is NOT in ind_set
+    mask = [i not in ind_set for i in range(pos_t.shape[0])]
+    pos_t_new = pos_t[mask, :]
+
+    return pos_t_new
+
+
+def update_path(p, pos_a, pos_t, time_step, agents, nt):
+    """
+    p: list of lists of task indices (0-based)
+    pos_a: (na, 2) agent positions
+    pos_t: (nt, 2) task positions
+    agents: object with Speed and Lt fields
+    nt: number of remaining tasks
+    """
+
+    ind_completed_tasks = []
+
+    for i in range(pos_a.shape[0]):
+
+        if len(p[i]) == 0:
+            continue  # no remaining tasks for agent i
+
+        # Next task index (Python 0-based)
+        task_idx = p[i][0]
+
+        d_a_t = pos_t[task_idx] - pos_a[i]
+        dist = np.linalg.norm(d_a_t)
+
+        # Can agent reach this task within this timestep?
+        if dist < time_step * agents.Speed[i]:
+
+            # Snap agent to task
+            pos_a[i] = pos_t[task_idx]
+
+            nt -= 1
+            agents.Lt[i] -= 1
+
+            # Record completed task
+            ind_completed_tasks.append(p[i][0])
+
+            # Remove completed task from path
+            p[i] = p[i][1:]
+
+        else:
+            # Move agent toward task
+            direction = d_a_t / dist
+            pos_a[i] = pos_a[i] + direction * time_step * agents.Speed[i]
+
+    return p, pos_a, ind_completed_tasks, nt, agents

gcaa/core/dta.py

@@ -1,4 +1,3 @@
-import json
 import time
 from dataclasses import dataclass
 from textwrap import wrap
@@ -7,12 +6,12 @@
 import numpy as np
 
 from gcaa.algorithms.greedy import GCAASolution
-from gcaa.tools.constants import SIMU_DIR
 from gcaa.core.control import ComputeCommandParamsWithVelocity, \
     OptimalControlSolution
 from gcaa.core.utility import CalcTaskUtility
 from gcaa.tools.basic import PrettyDict
-from gcaa.tools.disk import mkdir, dump_json
+from gcaa.tools.constants import SIMU_DIR
+from gcaa.tools.disk import dump_json
 from gcaa.tools.plotting import plotMapAllocation, PlotAgentRange
 from gcaa.tools.serialize import make_json_serializable
 
@@ -36,67 +35,6 @@ class Agents:
     kdrag: float
 
 
-def remove_completed_tasks(pos_t: np.ndarray, ind):
-    """
-    pos_t: (nt, 2) array of task positions
-    ind: list/array of completed task indices (1-based in MATLAB)
-    """
-
-    # Convert MATLAB's 1-based indices → Python 0-based
-    ind_set = {ind}
-
-    # Keep only rows whose index is NOT in ind_set
-    mask = [i not in ind_set for i in range(pos_t.shape[0])]
-    pos_t_new = pos_t[mask, :]
-
-    return pos_t_new
-
-
-def update_path(p, pos_a, pos_t, time_step, agents, nt):
-    """
-    p: list of lists of task indices (0-based)
-    pos_a: (na, 2) agent positions
-    pos_t: (nt, 2) task positions
-    agents: object with Speed and Lt fields
-    nt: number of remaining tasks
-    """
-
-    ind_completed_tasks = []
-
-    for i in range(pos_a.shape[0]):
-
-        if len(p[i]) == 0:
-            continue  # no remaining tasks for agent i
-
-        # Next task index (Python 0-based)
-        task_idx = p[i][0]
-
-        d_a_t = pos_t[task_idx] - pos_a[i]
-        dist = np.linalg.norm(d_a_t)
-
-        # Can agent reach this task within this timestep?
-        if dist < time_step * agents.Speed[i]:
-
-            # Snap agent to task
-            pos_a[i] = pos_t[task_idx]
-
-            nt -= 1
-            agents.Lt[i] -= 1
-
-            # Record completed task
-            ind_completed_tasks.append(p[i][0])
-
-            # Remove completed task from path
-            p[i] = p[i][1:]
-
-        else:
-            # Move agent toward task
-            direction = d_a_t / dist
-            pos_a[i] = pos_a[i] + direction * time_step * agents.Speed[i]
-
-    return p, pos_a, ind_completed_tasks, nt, agents
-
-
 def optimal_control_dta(
     na=5,
     nt=4,
@@ -118,7 +56,8 @@ def optimal_control_dta(
     """
 
     Lt = 1
-    nt_loiter = int(np.ceil(0.0 * nt)) if use_GCAA else 0
+    loiter_ratio=.0
+    nt_loiter = int(np.ceil(loiter_ratio * nt)) if use_GCAA else 0
     task_type = np.zeros(nt, dtype=int)
     task_type[:nt_loiter] = 1
     lambda_ = 1

gcaa/tests/test_allocation.py

@@ -0,0 +1,65 @@
+import numpy as np
+
+from gcaa.core.allocation import remove_completed_tasks, update_path
+
+
+def test_remove_completed_tasks_single_index_removal():
+    pos_t = np.array([
+        [0.0, 0.0],  # 0
+        [1.0, 1.0],  # 1
+        [2.0, 2.0],  # 2
+    ])
+
+    # Current implementation treats 'ind' as a single 0-based index
+    out = remove_completed_tasks(pos_t, 1)
+    # Expect row index 1 removed
+    expected = np.array([[0.0, 0.0], [2.0, 2.0]])
+    assert np.array_equal(out, expected)
+
+    # Index that doesn't match any (e.g., outside range) should keep mask true for all
+    out2 = remove_completed_tasks(pos_t, 5)
+    assert np.array_equal(out2, pos_t)
+
+
+def test_update_path_completion_and_motion():
+    # Two agents, three tasks
+    pos_a = np.array([
+        [0.0, 0.0],   # agent 0 near task 0
+        [0.0, 0.0],   # agent 1 far from task 2
+    ], dtype=float)
+
+    pos_t = np.array([
+        [0.05, 0.0],  # task 0 close to agent 0
+        [1.0, 0.0],   # task 1 (unused)
+        [1.0, 0.0],   # task 2 targeted by agent 1
+    ], dtype=float)
+
+    # Paths: agent 0 -> task 0, agent 1 -> task 2
+    p = [[0], [2]]
+
+    class AgentsStub:
+        def __init__(self):
+            self.Speed = np.array([1.0, 0.1])  # agent 0 fast, agent 1 slow
+            self.Lt = np.array([1, 1])
+
+    agents = AgentsStub()
+
+    time_step = 0.1
+    nt = 3
+
+    p2, pos_a2, completed, nt2, agents2 = update_path(p, pos_a.copy(), pos_t, time_step, agents, nt)
+
+    # Agent 0 should complete task 0 since dist=0.05 < time_step*Speed=0.1
+    assert 0 in completed
+    assert np.allclose(pos_a2[0], pos_t[0])
+    assert p2[0] == []
+    assert agents2.Lt[0] == 0
+
+    # Agent 1 should move towards task 2 but not reach it
+    # Initial dist is 1.0; step size is 0.1*0.1 = 0.01 along x-axis
+    assert 2 not in completed
+    assert np.allclose(pos_a2[1], [0.01, 0.0])
+    assert p2[1] == [2]
+
+    # nt reduced by number of completed tasks (1)
+    assert nt2 == nt - 1

gcaa/tests/test_dta.py

@@ -9,6 +9,13 @@ class TestDTA(TestCase):
     def setUp(self):
         np.random.seed(0)
 
+    def test_fixed_alloc(self):
+        optimal_control_dta(
+            n_rounds=4,
+            na=3,
+            nt=2,
+        )
+
     def test(self):
         results = optimal_control_dta(
             n_rounds=20,

gcaa/tests/test_utility.py

@@ -0,0 +1,81 @@
+import numpy as np
+
+from gcaa.core.utility import MinimumCostAlongLoitering
+
+
+def test_minimum_cost_along_loitering_basic():
+    agent_pos = np.array([0.0, 0.0])
+    agent_va = np.array([0.0, 0.0])
+
+    task_pos = np.array([
+        [1.0, 1.0],
+        [2.0, -1.0]
+    ])
+
+    task_radius = np.array([0.5, 1.0])
+    task_tloiter = np.array([1.0, 2.0])
+    task_tf = np.array([5.0, 10.0])
+    kdrag = 0.1
+    j = 0
+
+    rin, vt, rho = MinimumCostAlongLoitering(
+        agent_pos,
+        agent_va,
+        task_pos,
+        task_radius,
+        task_tloiter,
+        task_tf,
+        j,
+        kdrag
+    )
+
+    assert rin.shape == (2,)
+    assert vt.shape == (2,)
+    assert np.isfinite(rho)
+    assert rho > 0
+
+
+def test_minimum_cost_monotonic_radius():
+    agent_pos = np.array([0.0, 0.0])
+    agent_va = np.array([0.0, 0.0])
+    task_pos = np.array([[1.0, 0.0]])
+    task_tloiter = np.array([1.0])
+    task_tf = np.array([5.0])
+    kdrag = 0.1
+    j = 0
+
+    r_small = np.array([0.2])
+    r_large = np.array([2.0])
+
+    _, _, rho_small = MinimumCostAlongLoitering(
+        agent_pos, agent_va, task_pos,
+        r_small, task_tloiter, task_tf, j, kdrag
+    )
+    _, _, rho_large = MinimumCostAlongLoitering(
+        agent_pos, agent_va, task_pos,
+        r_large, task_tloiter, task_tf, j, kdrag
+    )
+
+    # Larger radius → longer loiter → higher cost
+    assert rho_large > rho_small
+
+
+def test_minimum_cost_repeatability():
+    # Ensure deterministic behavior (no randomness)
+    agent_pos = np.array([0.0, 0.0])
+    agent_va = np.array([0.0, 0.0])
+    task_pos = np.array([[0.5, 0.8]])
+    radius = np.array([1.0])
+    tloiter = np.array([1.2])
+    tf = np.array([4.0])
+    j = 0
+    kdrag = 0.2
+
+    r1 = MinimumCostAlongLoitering(agent_pos, agent_va, task_pos,
+                                   radius, tloiter, tf, j, kdrag)
+    r2 = MinimumCostAlongLoitering(agent_pos, agent_va, task_pos,
+                                   radius, tloiter, tf, j, kdrag)
+
+    assert np.allclose(r1[0], r2[0])
+    assert np.allclose(r1[1], r2[1])
+    assert np.isclose(r1[2], r2[2])