28class CommunicationController(ABCCommunicationController):
 29    """Class that communicates with Agilent using Macros
 30
 31    :param comm_dir: the complete directory path that was used in the MACRO file, common file that pychemstation and Chemstation use to communicate.
 32    :param cmd_file: name of the write file that pychemstation writes MACROs to, in `comm_dir`
 33    :param reply_file: name of the read file that Chemstation replies to, in `comm_dir
 34    :param offline: whether or not communication with Chemstation is to be established
 35    :param debug: if True, prints all send MACROs to an out.txt file
 36    """
 37
 38    def __init__(
 39        self,
 40        comm_dir: str,
 41        cmd_file: str = "cmd",
 42        reply_file: str = "reply",
 43        offline: bool = False,
 44        debug: bool = False,
 45    ):
 46        super().__init__(comm_dir, cmd_file, reply_file, offline, debug)
 47
 48    def get_num_val(self, cmd: str) -> Union[int, float]:
 49        tries = 10
 50        for _ in range(tries):
 51            self.send(Command.GET_NUM_VAL_CMD.value.format(cmd=cmd))
 52            res = self.receive()
 53            if res.is_ok():
 54                return res.ok_value.num_response
 55        raise RuntimeError("Failed to get number.")
 56
 57    def get_text_val(self, cmd: str) -> str:
 58        tries = 5
 59        for _ in range(tries):
 60            self.send(Command.GET_TEXT_VAL_CMD.value.format(cmd=cmd))
 61            res = self.receive()
 62            if res.is_ok():
 63                return res.ok_value.string_response
 64        raise RuntimeError("Failed to get string")
 65
 66    def get_status(self) -> Status:
 67        """Get device status(es).
 68
 69        :return: list of ChemStation's current status
 70        """
 71        self.send(Command.GET_STATUS_CMD)
 72        time.sleep(1)
 73
 74        try:
 75            res = self.receive()
 76            if res.is_err():
 77                return HPLCErrorStatus.NORESPONSE
 78            if res.is_ok():
 79                parsed_response = self.receive().ok_value.string_response
 80                self._most_recent_hplc_status = str_to_status(parsed_response)
 81                return self._most_recent_hplc_status
 82            else:
 83                raise RuntimeError("Failed to get status")
 84        except IOError:
 85            return HPLCErrorStatus.NORESPONSE
 86        except IndexError:
 87            return HPLCErrorStatus.MALFORMED
 88
 89    def _send(self, cmd: str, cmd_no: int, num_attempts=5) -> None:
 90        """Low-level execution primitive. Sends a command string to HPLC.
 91
 92        :param cmd: string to be sent to HPLC
 93        :param cmd_no: Command number
 94        :param num_attempts: Number of attempts to send the command before raising exception.
 95        :raises IOError: Could not write to command file.
 96        """
 97        err = None
 98        for _ in range(num_attempts):
 99            time.sleep(1)
100            try:
101                with open(self.cmd_file, "w", encoding="utf8") as cmd_file:
102                    cmd_file.write(f"{cmd_no} {cmd}")
103            except IOError as e:
104                err = e
105                continue
106            else:
107                return
108        else:
109            raise IOError(f"Failed to send command #{cmd_no}: {cmd}.") from err
110
111    def _receive(self, cmd_no: int, num_attempts=100) -> Result[str, str]:
112        """Low-level execution primitive. Recives a response from HPLC.
113
114        :param cmd_no: Command number
115        :param num_attempts: Number of retries to open reply file
116        :raises IOError: Could not read reply file.
117        :return: Potential ChemStation response
118        """
119        err: Optional[Union[OSError, IndexError, ValueError]] = None
120        err_msg = ""
121        for _ in range(num_attempts):
122            time.sleep(1)
123
124            try:
125                with open(self.reply_file, "r", encoding="utf_16") as reply_file:
126                    response = reply_file.read()
127            except OSError as e:
128                err = e
129                continue
130
131            try:
132                first_line = response.splitlines()[0]
133                try:
134                    response_no = int(first_line.split()[0])
135                except ValueError as e:
136                    err = e
137                    err_msg = f"Caused by {first_line}"
138            except IndexError as e:
139                err = e
140                continue
141
142            # check that response corresponds to sent command
143            if response_no == cmd_no:
144                return Ok(response)
145            else:
146                continue
147        else:
148            return Err(
149                f"Failed to receive reply to command #{cmd_no} due to {err} caused by {err_msg}."
150            )
151
152    def get_chemstation_dirs(self) -> Tuple[str, str, List[str]]:
153        method_dir, sequence_dir, data_dirs = None, None, None
154        for _ in range(10):
155            self.send(Command.GET_METHOD_DIR)
156            res = self.receive()
157            if res.is_ok():
158                method_dir = res.ok_value.string_response
159            self.send(Command.GET_SEQUENCE_DIR)
160            res = self.receive()
161            if res.is_ok():
162                sequence_dir = res.ok_value.string_response
163            self.send(Command.GET_DATA_DIRS)
164            res = self.receive()
165            if res.is_ok():
166                data_dirs = res.ok().string_response.split("|")
167            if method_dir and sequence_dir and data_dirs:
168                if not sequence_dir[0].isalpha():
169                    sequence_dir = "C:" + sequence_dir
170                if not method_dir[0].isalpha():
171                    method_dir = "C:" + method_dir
172                for i, data_dir in enumerate(data_dirs):
173                    if not data_dir[0].isalpha():
174                        data_dirs[i] = "C:" + data_dir
175                return method_dir, sequence_dir, data_dirs
176        raise ValueError(
177            f"One of the method: {method_dir}, sequence: {sequence_dir} or data directories: {data_dirs} could not be found, please provide your own."
178        )