update_info_json method is added to update and add the additional inf…

micromagneticmodel/driver.py

@@ -1,6 +1,8 @@
 import abc
 import datetime
+import importlib.metadata
 import json
+import math
 import pathlib
 import subprocess as sp
 import sys
@@ -124,23 +126,35 @@ def drive(
             If system directory already exists and append=False.
 
         """
+        drive_kwargs = kwargs.copy()
         # This method is implemented in the derived driver class. It raises
         # exception if any of the arguments are not valid.
         self.drive_kwargs_setup(kwargs)
         self._check_system(system)
         workingdir = self._setup_working_directory(
             system=system, dirname=dirname, mode="drive", append=append
         )
+        start_time = datetime.datetime.now()
 
         with uu.changedir(workingdir):
             self._write_input_files(
                 system=system,
                 ovf_format=ovf_format,
                 **kwargs,
             )
-            self._call(system=system, runner=runner, verbose=verbose, **kwargs)
-            self._read_data(system)
 
+            self._write_info_json(system, start_time, **drive_kwargs)
+            try:
+                self._call(system=system, runner=runner, verbose=verbose, **kwargs)
+            except Exception:
+                success = False
+                raise
+            else:
+                success = True
+            finally:
+                end_time = datetime.datetime.now()
+                self._update_info_json(start_time, end_time, success)
+            self._read_data(system)
         system.drive_number += 1
 
     def schedule(
@@ -234,6 +248,7 @@ def schedule(
             If system directory already exists and append=False.
 
         """
+        schedule_kwargs = kwargs.copy()
         # This method is implemented in the derived driver class. It raises
         # exception if any of the arguments are not valid.
         self.schedule_kwargs_setup(kwargs)
@@ -247,6 +262,8 @@ def schedule(
         if pathlib.Path(header).exists():
             header = pathlib.Path(header).absolute()
 
+        start_time = datetime.datetime.now()
+
         with uu.changedir(workingdir):
             self._write_input_files(
                 system=system,
@@ -256,6 +273,7 @@ def schedule(
             self._write_schedule_script(
                 system=system, header=header, script_name=script_name, runner=runner
             )
+            self._write_info_json(system, start_time, **schedule_kwargs)
 
             stdout = stderr = sp.PIPE
             if sys.platform == "win32":
@@ -292,15 +310,19 @@ def _write_schedule_script(self, system, header, script_name, runner):
             f.write("\n")
             f.write("\n".join(run_commands))
 
-    def _write_info_json(self, system, **kwargs):
+    def _write_info_json(self, system, start_time, **kwargs):
         info = {}
         info["drive_number"] = system.drive_number
-        info["date"] = datetime.datetime.now().strftime("%Y-%m-%d")
-        info["time"] = datetime.datetime.now().strftime("%H:%M:%S")
-        info["driver"] = self.__class__.__name__
+        info["date"] = start_time.strftime("%Y-%m-%d")
+        info["time"] = start_time.strftime("%H:%M:%S")
+        info["start_time"] = start_time.isoformat(timespec="seconds")
         # "adapter" is the ubermag package (e.g. oommfc) that communicates with the
         # calculator (e.g. OOMMF)
         info["adapter"] = self.__module__.split(".")[0]
+        info["adapter_version"] = importlib.metadata.version(
+            self.__module__.split(".")[0]
+        )
+        info["driver"] = self.__class__.__name__
         for k, v in kwargs.items():
             info[k] = v
         with open("info.json", "w", encoding="utf-8") as jsonfile:
@@ -325,3 +347,19 @@ def _setup_working_directory(system, dirname, mode, append=True):
         workingdir = system_dir / f"{mode}-{next_number}"
         workingdir.mkdir(parents=True)
         return workingdir
+
+    @staticmethod
+    def _conversion_to_hms(time_difference):
+        total_seconds = math.ceil(time_difference.total_seconds())
+        hours, remainder = divmod(total_seconds, 3600)
+        minutes, seconds = divmod(remainder, 60)
+        return f"{hours:02}:{minutes:02}:{seconds:02}"
+
+    def _update_info_json(self, start_time, end_time, success):
+        with open("info.json", encoding="utf-8") as jsonfile:
+            info = json.load(jsonfile)
+        info["end_time"] = end_time.isoformat(timespec="seconds")
+        info["elapsed_time"] = self._conversion_to_hms(end_time - start_time)
+        info["success"] = success
+        with open("info.json", "w", encoding="utf-8") as jsonfile:
+            json.dump(info, jsonfile)

micromagneticmodel/tests/test_driver.py

@@ -1,4 +1,3 @@
-import datetime
 import json
 
 import discretisedfield as df
@@ -37,7 +36,6 @@ def _check_system(self, system):
     def _write_input_files(self, system, **kwargs):
         with open(f"{system.name}.input", "w", encoding="utf-8") as f:
             f.write(str(-1))  # factor -1 used to invert magnetisation direction in call
-        self._write_info_json(system, **kwargs)
 
     def _call(self, system, runner, **kwargs):
         with open(f"{system.name}.input", encoding="utf-8") as f:
@@ -79,10 +77,24 @@ def test_external_driver(tmp_path):
     assert info["driver"] == "MyExternalDriver"
     assert info["drive_number"] == 0
 
-    info_time = datetime.datetime.fromisoformat(f"{info['date']}T{info['time']}")
-    now = datetime.datetime.now()
-    # assumption: this test runs in under one minute
-    assert (now - info_time).total_seconds() < 60
+    assert "start_time" in info
+    assert "end_time" in info
+    assert "elapsed_time" in info
+    assert "success" in info
+    assert info["success"] is True
+
+    def _parse_time_str_to_seconds(time_str):
+        h, m, s = map(int, time_str.split(":"))
+        return h * 3600 + m * 60 + s
+
+    elapsed_time = info["elapsed_time"]
+
+    if isinstance(elapsed_time, str):  # HH:MM:SS format
+        elapsed_seconds = _parse_time_str_to_seconds(elapsed_time)
+        # assumption: this test runs in under one minute
+        assert elapsed_seconds < 60
+    else:
+        raise TypeError("Unexpected format for elapsed_time")
 
     with pytest.raises(FileExistsError):
         driver.drive(system, dirname=str(tmp_path), append=False)