|
| 1 | +import contextlib |
| 2 | +import os |
| 3 | +import tempfile |
1 | 4 | import warnings |
2 | 5 |
|
3 | 6 | import prettytable |
@@ -70,6 +73,8 @@ class Controls(BaseModel, validate_assignment=True, extra="forbid"): |
70 | 73 | pUnitGamma: float = Field(0.2, gt=0.0, lt=1.0) |
71 | 74 | boundHandling: BoundHandling = BoundHandling.Reflect |
72 | 75 | adaptPCR: bool = True |
| 76 | + # Private field for IPC file |
| 77 | + _IPCFilePath: str = "" |
73 | 78 |
|
74 | 79 | @model_validator(mode="wrap") |
75 | 80 | def warn_setting_incorrect_properties(self, handler: ValidatorFunctionWrapHandler) -> "Controls": |
@@ -131,3 +136,23 @@ def __str__(self) -> str: |
131 | 136 | table.field_names = ["Property", "Value"] |
132 | 137 | table.add_rows([[k, v] for k, v in self.model_dump().items()]) |
133 | 138 | return table.get_string() |
| 139 | + |
| 140 | + def initialise_IPC(self): |
| 141 | + """Setup the inter-process communication file.""" |
| 142 | + IPCFile = tempfile.mkstemp() |
| 143 | + self._IPCFilePath = IPCFile[1] |
| 144 | + with open(self._IPCFilePath, "wb") as f: |
| 145 | + f.write(b"0") |
| 146 | + return None |
| 147 | + |
| 148 | + def sendStopEvent(self): |
| 149 | + """Sends the stop event via the inter-process communication file.""" |
| 150 | + with open(self._IPCFilePath, "wb") as f: |
| 151 | + f.write(b"1") |
| 152 | + return None |
| 153 | + |
| 154 | + def delete_IPC(self): |
| 155 | + """Delete the inter-process communication file.""" |
| 156 | + with contextlib.suppress(FileNotFoundError): |
| 157 | + os.remove(self._IPCFilePath) |
| 158 | + return None |
0 commit comments