pychemstation.utils.abc_tables.abc_comm

  1"""
  2Module to provide API for the communication with Agilent HPLC systems.
  3
  4HPLCController sends commands to Chemstation software via a command file.
  5Answers are received via reply file. On the Chemstation side, a custom
  6Macro monitors the command file, executes commands and writes to the reply file.
  7Each command is given a number (cmd_no) to keep track of which commands have
  8been processed.
  9
 10Authors: Alexander Hammer, Hessam Mehr, Lucy Hao
 11"""
 12
 13import abc
 14import os
 15import time
 16from abc import abstractmethod
 17from typing import Union
 18
 19from result import Err, Ok, Result
 20
 21from ..macro import HPLCAvailStatus, Command, Response, Status
 22
 23
 24class ABCCommunicationController(abc.ABC):
 25    """Abstract class representing the communication controller.
 26
 27    :param comm_dir: the complete directory path that was used in the MACRO file, common file that pychemstation and Chemstation use to communicate.
 28    :param cmd_file: name of the write file that pychemstation writes MACROs to, in `comm_dir`
 29    :param reply_file: name of the read file that Chemstation replies to, in `comm_dir
 30    :param offline: whether or not communication with Chemstation is to be established
 31    :param debug: if True, prints all send MACROs to an out.txt file
 32    """
 33
 34    # maximum command number
 35    MAX_CMD_NO = 255
 36
 37    def __init__(
 38        self,
 39        comm_dir: str,
 40        cmd_file: str = "cmd",
 41        reply_file: str = "reply",
 42        offline: bool = False,
 43        debug: bool = False,
 44    ):
 45        if not offline:
 46            self.debug = debug
 47            if os.path.isdir(comm_dir):
 48                self.cmd_file = os.path.join(comm_dir, cmd_file)
 49                self.reply_file = os.path.join(comm_dir, reply_file)
 50                self.cmd_no = 0
 51            else:
 52                raise FileNotFoundError(f"comm_dir: {comm_dir} not found.")
 53
 54            # Create files for Chemstation to communicate with Python
 55            open(self.cmd_file, "a").close()
 56            open(self.reply_file, "a").close()
 57
 58            self.reset_cmd_counter()
 59            self._most_recent_hplc_status: Status = self.get_status()
 60
 61    @abstractmethod
 62    def get_num_val(self, cmd: str) -> Union[int, float]:
 63        pass
 64
 65    @abstractmethod
 66    def get_text_val(self, cmd: str) -> str:
 67        pass
 68
 69    @abstractmethod
 70    def get_status(self) -> Status:
 71        pass
 72
 73    @abstractmethod
 74    def _send(self, cmd: str, cmd_no: int, num_attempts=5) -> None:
 75        pass
 76
 77    @abstractmethod
 78    def _receive(self, cmd_no: int, num_attempts=100) -> Result[str, str]:
 79        pass
 80
 81    def set_status(self):
 82        """Updates current status of HPLC machine"""
 83        self._most_recent_hplc_status = self.get_status()
 84
 85    def check_if_not_running(self) -> bool:
 86        """Checks if HPLC machine is in an available state, meaning a state that data is not being written.
 87
 88        :return: whether the HPLC machine is in a safe state to retrieve data back."""
 89        self.set_status()
 90        hplc_avail = isinstance(self._most_recent_hplc_status, HPLCAvailStatus)
 91        time.sleep(10)
 92        self.set_status()
 93        hplc_actually_avail = isinstance(self._most_recent_hplc_status, HPLCAvailStatus)
 94        time.sleep(10)
 95        self.set_status()
 96        hplc_final_check_avail = isinstance(
 97            self._most_recent_hplc_status, HPLCAvailStatus
 98        )
 99        return hplc_avail and hplc_actually_avail and hplc_final_check_avail
100
101    def sleepy_send(self, cmd: Union[Command, str]):
102        self.send("Sleep 0.1")
103        self.send(cmd)
104        self.send("Sleep 0.1")
105
106    def send(self, cmd: Union[Command, str]):
107        """Sends a command to Chemstation.
108
109        :param cmd: Command to be sent to HPLC
110        """
111        if self.cmd_no == self.MAX_CMD_NO:
112            self.reset_cmd_counter()
113
114        cmd_to_send: str = cmd.value if isinstance(cmd, Command) else cmd
115        self.cmd_no += 1
116        self._send(cmd_to_send, self.cmd_no)
117        if self.debug:
118            f = open("out.txt", "a")
119            f.write(cmd_to_send + "\n")
120            f.close()
121
122    def receive(self) -> Result[Response, str]:
123        """Returns messages received in reply file.
124
125        :return: ChemStation response
126        """
127        num_response_prefix = "Numerical Responses:"
128        str_response_prefix = "String Responses:"
129        possible_response = self._receive(self.cmd_no)
130        if possible_response.is_ok():
131            lines = possible_response.ok_value.splitlines()
132            for line in lines:
133                if str_response_prefix in line and num_response_prefix in line:
134                    string_responses_dirty, _, numerical_responses = line.partition(
135                        num_response_prefix
136                    )
137                    _, _, string_responses = string_responses_dirty.partition(
138                        str_response_prefix
139                    )
140                    return Ok(
141                        Response(
142                            string_response=string_responses.strip(),
143                            num_response=float(numerical_responses.strip()),
144                        )
145                    )
146            return Err("Could not retrieve HPLC response")
147        else:
148            return Err(f"Could not establish response to HPLC: {possible_response}")
149
150    def reset_cmd_counter(self):
151        """Resets the command counter."""
152        self._send(Command.RESET_COUNTER_CMD.value, cmd_no=self.MAX_CMD_NO + 1)
153        self._receive(cmd_no=self.MAX_CMD_NO + 1)
154        self.cmd_no = 0
155
156    def stop_macro(self):
157        """Stops Macro execution. Connection will be lost."""
158        self.send(Command.STOP_MACRO_CMD)