Adds code for IPC files (#77)

Author: DrPaulSharp

RATapi/controls.py

@@ -1,3 +1,6 @@
+import contextlib
+import os
+import tempfile
 import warnings
 
 import prettytable
@@ -70,6 +73,8 @@ class Controls(BaseModel, validate_assignment=True, extra="forbid"):
     pUnitGamma: float = Field(0.2, gt=0.0, lt=1.0)
     boundHandling: BoundHandling = BoundHandling.Reflect
     adaptPCR: bool = True
+    # Private field for IPC file
+    _IPCFilePath: str = ""
 
     @model_validator(mode="wrap")
     def warn_setting_incorrect_properties(self, handler: ValidatorFunctionWrapHandler) -> "Controls":
@@ -131,3 +136,31 @@ def __str__(self) -> str:
         table.field_names = ["Property", "Value"]
         table.add_rows([[k, v] for k, v in self.model_dump().items()])
         return table.get_string()
+
+    def initialise_IPC(self):
+        """Setup the inter-process communication file."""
+        IPC_obj, self._IPCFilePath = tempfile.mkstemp()
+        os.write(IPC_obj, b"0")
+        os.close(IPC_obj)
+        return None
+
+    def sendStopEvent(self):
+        """Sends the stop event via the inter-process communication file.
+
+        Warnings
+        --------
+        UserWarning
+            Raised if we try to delete an IPC file that was not initialised.
+        """
+        if os.path.isfile(self._IPCFilePath):
+            with open(self._IPCFilePath, "wb") as f:
+                f.write(b"1")
+        else:
+            warnings.warn("An IPC file was not initialised.", UserWarning, stacklevel=2)
+        return None
+
+    def delete_IPC(self):
+        """Delete the inter-process communication file."""
+        with contextlib.suppress(FileNotFoundError):
+            os.remove(self._IPCFilePath)
+        return None

RATapi/inputs.py

@@ -470,4 +470,6 @@ def make_controls(input_controls: RATapi.Controls, checks: Checks) -> Control:
     # IPC
     controls.IPCFilePath = ""
 
+    controls.IPCFilePath = input_controls._IPCFilePath
+
     return controls

RATapi/run.py

@@ -130,6 +130,8 @@ def run(project, controls):
         for index, value in enumerate(getattr(problem_definition, parameter_field[class_list])):
             getattr(project, class_list)[index].value = value
 
+    controls.delete_IPC()
+
     if display_on:
         print("Finished RAT " + horizontal_line)
 

tests/test_controls.py

@@ -1,5 +1,8 @@
 """Test the controls module."""
 
+import contextlib
+import os
+import tempfile
 from typing import Any, Union
 
 import pydantic
@@ -9,6 +12,17 @@
 from RATapi.utils.enums import BoundHandling, Display, Parallel, Procedures, Strategies
 
 
+@pytest.fixture
+def IPC_controls():
+    """A controls object with a temporary file set as the IPC file."""
+    IPC_controls = Controls()
+    IPC_obj, IPC_controls._IPCFilePath = tempfile.mkstemp()
+    os.close(IPC_obj)
+    yield IPC_controls
+    with contextlib.suppress(FileNotFoundError):
+        os.remove(IPC_controls._IPCFilePath)
+
+
 def test_initialise_procedure_error() -> None:
     """Tests for a ValidationError if the procedure property of the Controls class is initialised with an invalid
     value.
@@ -851,3 +865,35 @@ def test_dream_nChains_error(self, value: int) -> None:
     def test_control_class_dream_str(self, table_str) -> None:
         """Tests the Dream model __str__."""
         assert self.dream.__str__() == table_str
+
+
+def test_initialise_IPC() -> None:
+    """Tests that an Inter-Process Communication File can be set up."""
+    test_controls = Controls()
+    test_controls.initialise_IPC()
+    assert test_controls._IPCFilePath != ""
+    with open(test_controls._IPCFilePath, "rb") as f:
+        file_content = f.read()
+    assert file_content == b"0"
+    os.remove(test_controls._IPCFilePath)
+
+
+def test_sendStopEvent(IPC_controls) -> None:
+    """Tests that an Inter-Process Communication File can be modified."""
+    IPC_controls.sendStopEvent()
+    with open(IPC_controls._IPCFilePath, "rb") as f:
+        file_content = f.read()
+    assert file_content == b"1"
+
+
+def test_sendStopEvent_empty_file() -> None:
+    """Tests that we do not write to a non-existent Inter-Process Communication File."""
+    test_controls = Controls()
+    with pytest.warns(UserWarning, match="An IPC file was not initialised."):
+        test_controls.sendStopEvent()
+
+
+def test_delete_IPC(IPC_controls) -> None:
+    """Tests that an Inter-Process Communication File can be safely removed."""
+    IPC_controls.delete_IPC()
+    assert not os.path.isfile(IPC_controls._IPCFilePath)