REF #41: Add commandline support for topsim experiment.

pyproject.toml

@@ -29,11 +29,12 @@ dependencies = [
     "tables >= 3.7",
     "tqdm",
     "h5py",
-    "click"
+    "click",
+    "tabulate"
 ]
 
 [project.scripts]
-topsim = "topsim.cli:cli"
+topsim = "topsim.runtime.commands:cli"
 
 [project.urls]
 Homepage = "https://github.com/top-sim/topsim"

test/simulation_data/heft_single_observation_simulation.json

@@ -1,57 +1,59 @@
 {
-	"instrument": {
-		"telescope": {
-			"total_arrays": 36,
-			"max_ingest_resources": 1,
-			"pipelines": {
-				"emu": {
-					"workflow": "test/data/config/workflow_config_minutes_longtask.json",
-					"ingest_demand": 1
-				}
-			},
-			"observations": [
-				{
-					"name": "emu",
-					"start": 1,
-					"duration": 10,
-					"instrument_demand": 36,
-					"data_product_rate": 0.03333333333333333
-				}
-			]
-		}
-	},
-	"cluster": {
-		"header": {
-		},
-		"system": {
-			"resources": {
-				"cat0_m0": {
-					"flops": 7000.0,
-					"compute_bandwidth": 1.0
-				},
-				"cat1_m1": {
-					"flops": 6000.0,
-					"compute_bandwidth": 1.0
-				},
-				"cat2_m2": {
-					"flops": 11000.0,
-					"compute_bandwidth": 1.0
-				}
-			},
-			"system_bandwidth": 1.0
-		}
-	},
-	"buffer": {
-		"hot": {
-			"capacity": 500,
-			"max_ingest_rate": 0.08333333333333333
-		},
-		"cold": {
-			"capacity": 250,
-			"max_data_rate": 0.03333333333333333
-		}
-	},
-	"planning": "heft",
-	"scheduling": "fifo",
-	"timestep": "minutes"
-}
+  "instrument": {
+    "telescope": {
+      "total_arrays": 36,
+      "max_ingest_resources": 1,
+      "pipelines": {
+        "emu": {
+          "workflow": "test/data/config/workflow_config_minutes_longtask.json",
+          "ingest_demand": 1
+        }
+      },
+      "observations": [
+        {
+          "name": "emu",
+          "start": 1,
+          "duration": 10,
+          "instrument_demand": 36,
+          "data_product_rate": 0.03333333333333333
+        }
+      ]
+    }
+  },
+  "cluster": {
+    "header": {},
+    "system": {
+      "resources": {
+        "cat0": {
+          "compute_bandwidth": 1.0,
+          "flops": 7000.0,
+          "count": 1
+        },
+        "cat1": {
+          "compute_bandwidth": 1.0,
+          "flops": 6000.0,
+          "count": 1
+        },
+        "cat2": {
+          "compute_bandwidth": 1.0,
+          "flops": 11000.0,
+          "count": 1
+        }
+      },
+      "system_bandwidth": 1.0
+    }
+  },
+  "buffer": {
+    "hot": {
+      "capacity": 500,
+      "max_ingest_rate": 0.08333333333333333
+    },
+    "cold": {
+      "capacity": 250,
+      "max_data_rate": 0.03333333333333333
+    }
+  },
+  "planning": "heft",
+  "scheduling": "fifo",
+  "timestep": "minutes"
+}

test/test_runtime.py

@@ -0,0 +1,69 @@
+# Copyright (C) 2025 RW Bunney
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""
+Test runtime.commands and runtime.parser
+"""
+
+import unittest
+
+from topsim.runtime.parser import import_class, parse_experiment
+from topsim.user.schedule.batch_allocation import BatchProcessing
+from topsim.user.schedule.dynamic_plan import DynamicSchedulingFromPlan
+from topsim.user.plan.batch_planning import BatchPlanning
+from topsim.user.plan.static_planning import SHADOWPlanning
+
+class TestImportClass(unittest.TestCase):
+
+    def test_import_class(self):
+        """
+        Confirm we import classes correctly
+        """
+        class_exists = "topsim.user.schedule.batch_allocation.BatchProcessing"
+        cls = import_class(class_exists)
+        self.assertEqual(BatchProcessing, cls)
+
+    def test_import_class_doesnt_exist(self):
+
+        class_fake = "topsim.user.schedule.fake_allocation.TestFake"
+        self.assertRaises(ModuleNotFoundError, import_class, class_fake)
+
+    def test_import_module_node_class(self):
+
+        module = "topsim.user.schedule.batch_allocation"
+        self.assertRaises(RuntimeError, import_class, module)
+
+class TestExperimentParser(unittest.TestCase):
+
+    planning = ["topsim.user.plan.batch_planning.BatchPlanning",
+                "topsim.user.plan.static_planning.SHADOWPlanning"]
+
+    scheduling = ["topsim.user.schedule.batch_allocation.BatchProcessing",
+                  "topsim.user.schedule.dynamic_plan.DynamicSchedulingFromPlan"]
+
+    data = ["NoData", "TaskEdgeData", "EdgeData"]
+
+    def test_parse_experiment(self):
+
+        allocator_combinations, data_combinations = parse_experiment(self.planning,
+                                                                     self.scheduling,
+                                                                     self.data)
+        exp_alloc_comb = [(BatchPlanning, BatchProcessing),
+                          (SHADOWPlanning, DynamicSchedulingFromPlan)]
+        exp_data_comb = [ {'use_task_data': False, 'use_edge_data': False},
+                              {'use_task_data': True, 'use_edge_data': True},
+                              {'use_task_data': False, 'use_edge_data': True}]
+        self.assertListEqual(exp_alloc_comb, allocator_combinations)
+        self.assertListEqual(exp_data_comb, data_combinations)

test/test_simulation.py

@@ -50,12 +50,6 @@ def setUp(self) -> None:
         self.env = simpy.Environment()
         self.output = f'test/data/output/hdf5.h5'
 
-    def tearDown(self):
-        output = f'test/data/output/hdf5.h5'
-        os.remove(output)
-        # os.remove(f'{output}')
-        # os.remove(f'{output}-tasks.pkl')
-
     def test_simulation_produces_file(self):
         simulation = Simulation(
             self.env,
@@ -74,10 +68,19 @@ def test_simulation_produces_file(self):
 
         store = pd.HDFStore(self.output)
         store.close()
+        os.remove(self.output)
 
-        # store[f'{s}/standard_simulation/sim']
-
-    # def
+    def test_simulation_nofile_exception(self):
+        self.assertRaises(ValueError, Simulation,
+            self.env,
+            CONFIG,
+            Telescope,
+            planning_model=SHADOWPlanning('heft'),
+            scheduling=DynamicSchedulingFromPlan(),
+            delay=None,
+            timestamp=0,
+            to_file=True,
+        )
 
 
 class TestSimulationBatchProcessing(unittest.TestCase):

topsim/core/scheduler.py

@@ -272,7 +272,7 @@ def allocate_tasks(self, observation):
         task_pool = set()
         _total_tasks = 0 # len(current_plan.tasks)
         _curr_tasks = 0 # len(current_plan.tasks)
-        _tqdm = True
+        _tqdm = False
         pbar_setup = False
         pbar = None
         _event_added = False

topsim/core/simulation.py

@@ -16,7 +16,7 @@
 from topsim.core.delay import DelayModel
 
 LOGGER = logging.getLogger(__name__)
-
+HEARTBEAT_INT_SECONDS = 10*60 # Seconds
 
 class Simulation:
     """
@@ -193,6 +193,8 @@ def __init__(
                 self._hdf5_store.close()
             except Exception as e:
                 LOGGER.error('%s', e)
+                self._hdf5_store = None
+
         elif self.to_file and hdf5_path is None:
             raise ValueError(
                 'Attempted to initialise Simulation object that outputs'
@@ -257,12 +259,21 @@ def start(self, runtime=-1):
         self.env.process(self.scheduler.run())
         self.env.process(self.buffer.run())
 
+        heartbeat = time.monotonic() # use monotonic to avoid going backward
+
         if runtime > 0:
             self.env.run(until=runtime)
         else:
+
             while not self.is_finished():
                 self.env.run(self.env.now + 1)
-            # self.env.run(self.env.now + 1)
+
+                now = time.monotonic()
+                if now - heartbeat > HEARTBEAT_INT_SECONDS:
+                    wall_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                    LOGGER.info("Wall time: %s / sim_time: %d", wall_time, self.env.now)
+
+                    heartbeat = now            # self.env.run(self.env.now + 1)
 
         LOGGER.info("Simulation Finished @ %s", self.env.now)
         self.monitor.collate_events()