29class HPLCController:
 30    # tables
 31    METHOD_TIMETABLE = Table(register="RCPMP1Method[1]", name="Timetable")
 32
 33    SEQUENCE_TABLE = Table(register="_sequence[1]", name="SeqTable1")
 34
 35    INJECTOR_TABLE = Table(register="RCWLS1Pretreatment[1]", name="InstructionTable")
 36
 37    PUMP_DEVICE = Device(register="RCPMP1Status")
 38
 39    INJECTOR_DEVICE = Device(register="RCWLS1Method")
 40
 41    SAMPLE_INFO = Table(register="_CONFIG", name="SampleInfo")
 42
 43    COLUMN_TEMP_DEVICE = Device(register="RCTHM1Method")
 44
 45    DAD_DEVICE = Device(register="RCDAD1Method")
 46
 47    def __init__(
 48        self,
 49        comm_dir: str,
 50        method_dir: Optional[str] = None,
 51        sequence_dir: Optional[str] = None,
 52        extra_data_dirs: Optional[List[str]] = None,
 53        offline: bool = False,
 54        debug: bool = False,
 55    ):
 56        """Initialize HPLC controller. The `hplc_talk.mac` macro file must be loaded in the Chemstation software.
 57        `comm_dir` must match the file path in the macro file. All file paths are normal strings, with the left slash
 58        double escaped: "C:\\my_folder\\"
 59
 60        :param comm_dir: Name of directory for communication, where ChemStation will read and write from. Can be any existing directory.
 61        the first one in the list. In other words, the first dir in the list is highest prio. Must be "normal" strings and not r-strings.
 62        :raises FileNotFoundError: If either `data_dir`, `method_dir`, `sequence_dir`, `sequence_data_dir`or `comm_dir` is not a valid directory.
 63        """
 64        self.comm: CommunicationController = CommunicationController(
 65            comm_dir=comm_dir, debug=debug, offline=offline
 66        )
 67        data_dirs: List[str] = []
 68        if not offline:
 69            if not method_dir or not sequence_dir or not extra_data_dirs:
 70                method_dir, sequence_dir, data_dirs = self.comm.get_chemstation_dirs()
 71            if extra_data_dirs:
 72                data_dirs.extend(extra_data_dirs)
 73            data_dirs = list(set([os.path.normpath(p) for p in data_dirs]))
 74        if (method_dir and sequence_dir and data_dirs and not offline) or offline:
 75            self.method_controller: MethodController = MethodController(
 76                controller=self.comm,
 77                src=method_dir,
 78                data_dirs=data_dirs,
 79                table=self.METHOD_TIMETABLE,
 80                offline=offline,
 81                injector=InjectorController(
 82                    controller=self.comm, table=self.INJECTOR_TABLE, offline=offline
 83                ),
 84                pump=PumpController(
 85                    controller=self.comm, table=self.PUMP_DEVICE, offline=offline
 86                ),
 87                dad=DADController(
 88                    controller=self.comm, table=self.DAD_DEVICE, offline=offline
 89                ),
 90                column=ColumnController(
 91                    controller=self.comm, table=self.COLUMN_TEMP_DEVICE, offline=offline
 92                ),
 93                sample_info=SampleInfo(
 94                    controller=self.comm, table=self.SAMPLE_INFO, offline=offline
 95                ),
 96            )
 97            self.sequence_controller: SequenceController = SequenceController(
 98                controller=self.comm,
 99                src=sequence_dir,
100                data_dirs=data_dirs,
101                table=self.SEQUENCE_TABLE,
102                method_controller=self.method_controller,
103                offline=offline,
104            )
105        elif not offline and (not method_dir or not sequence_dir or not data_dirs):
106            raise ValueError(
107                f"Expected a method dir: {method_dir}, sequence dir: {sequence_dir} and data dirs:{data_dirs} but one was None."
108            )
109        else:
110            raise ValueError("Expected error occured, please try again.")
111
112    def send(self, cmd: Union[Command, str]):
113        """Sends any Command or string to Chemstation.
114
115        :param cmd: the macro to send to Chemstation
116        """
117        if not self.comm:
118            raise RuntimeError(
119                "Communication controller must be initialized before sending command. It is currently in offline mode."
120            )
121        self.comm.send(cmd)
122
123    def receive(self) -> None | Response | str:
124        """Get the most recent response from Chemstation.
125
126        :return: most recent response from the most recently sent MACRO that returned a response.
127        """
128        if not self.comm:
129            raise RuntimeError(
130                "Communication controller must be initialized before sending command. It is currently in offline mode."
131            )
132        res = self.comm.receive()
133        if res.is_ok():
134            return res.ok_value
135        else:
136            return res.err_value
137
138    def status(self) -> Status:
139        """Get the current status of the HPLC machine.
140
141        :return: current status of the HPLC machine; Status types can be found in `pychemstation.utils.macro`
142        """
143        if not self.comm:
144            raise RuntimeError(
145                "Communication controller must be initialized before sending command. It is currently in offline mode."
146            )
147        return self.comm.get_status()
148
149    def switch_method(self, method_name: str):
150        """Allows the user to switch between pre-programmed methods. No need to append '.M'
151        to the end of the method name. For example. for the method named 'General-Poroshell.M',
152        only 'General-Poroshell' is needed.
153
154        :param method_name: any available method in Chemstation method directory
155        :raises  IndexError: Response did not have expected format. Try again.
156        :raises  AssertionError: The desired method is not selected. Try again.
157        """
158        self.method_controller.switch(method_name)
159
160    def switch_sequence(self, sequence_name: str):
161        """Allows the user to switch between pre-programmed sequences. The sequence name does not need the '.S' extension.
162         For example: for the method named 'mySeq.S', only 'mySeq' is needed.
163
164        :param sequence_name: The name of the sequence file
165        """
166        self.sequence_controller.switch(sequence_name)
167
168    def run_method(
169        self,
170        experiment_name: str,
171        add_timestamp: bool = True,
172        stall_while_running: bool = True,
173    ):
174        """This is the preferred method to trigger a run.
175        Starts the currently selected method, storing data
176        under the <data_dir>/<experiment_name>.D folder.
177        Device must be ready.
178
179        :param experiment_name: Name of the experiment
180        :param stall_while_running: whether to return or stall while HPLC runs.
181        :param add_timestamp: whether to append a timestamp in '%Y-%m-%d-%H-%M' format to end of experiment name.
182        """
183        self.method_controller.run(
184            experiment_name=experiment_name,
185            stall_while_running=stall_while_running,
186            add_timestamp=add_timestamp,
187        )
188
189    def stop_method(self):
190        """Stops the current running method, manual intervention may be needed."""
191        self.method_controller.stop()
192
193    def run_sequence(self, stall_while_running: bool = True):
194        """Starts the currently loaded sequence, storing data
195        under one of the data_dirs/<sequence table name> folder.
196        Device must be ready.
197
198        :param stall_while_running: whether to return or stall while HPLC runs.
199        """
200        self.sequence_controller.run(stall_while_running=stall_while_running)
201
202    def check_method_complete(self) -> Tuple[float, int]:
203        """Check if the currently running method (if any) is done.
204
205        :returns: the percent of the method run completed, and whether the run is complete.
206        """
207        return self.method_controller.check_hplc_run_finished()
208
209    def check_sequence_complete(self) -> Tuple[float, int]:
210        """Check if the currently running sequence (if any) is done.
211
212        :return: the percent of the sequence run completed, and whether the run is complete.
213        """
214        return self.sequence_controller.check_hplc_run_finished()
215
216    def edit_method(self, updated_method: MethodDetails, save: bool = False):
217        """Updated the currently loaded method in ChemStation with provided values.
218
219        :param updated_method: the method with updated values, to be sent to Chemstation to modify the currently loaded method.
220        :param save: whether this method should be saved to disk, or just modified.
221        """
222        self.method_controller.edit(updated_method, save)
223
224    def edit_sequence(self, updated_sequence: SequenceTable):
225        """Updates the currently loaded sequence table with the provided table, and saves the sequence.
226
227        :param updated_sequence: The sequence table to be written to the currently loaded sequence table.
228        """
229        self.sequence_controller.edit(updated_sequence)
230
231    def get_last_run_method_file_path(self) -> str:
232        """Get the folder (ending with .D) for last run method.
233
234        :return: Complete path for method run.
235        """
236        if len(self.method_controller.data_files) > 0:
237            return self.method_controller.data_files[-1]
238        else:
239            raise UserWarning("No data yet!")
240
241    def get_last_run_method_report(
242        self,
243        custom_path: Optional[str] = None,
244        report_type: ReportType = ReportType.CSV,
245    ) -> AgilentReport:
246        """Return data contained in the REPORT files. Use `aghplctools` if you want more report processing utility.
247
248        :param custom_path: path to sequence folder
249        :param report_type: read either the TXT or CSV version
250        :return: report data for method
251        """
252        return self.method_controller.get_report(
253            custom_path=custom_path, report_type=report_type
254        )[0]
255
256    def get_last_run_method_data(
257        self, read_uv: bool = False, custom_path: Optional[str] = None
258    ) -> Dict[int, AgilentHPLCChromatogram] | AgilentChannelChromatogramData:
259        """Returns the last run method data.
260
261        :param custom_path: If you want to just load method data but from a file path. This file path must be the complete file path.
262        :param read_uv: whether to also read the UV file
263        """
264        if read_uv:
265            return self.method_controller.get_data_uv(custom_path=custom_path)
266        else:
267            return self.method_controller.get_data(custom_path=custom_path)
268
269    def get_last_run_sequence_file_paths(self) -> SequenceDataFiles:
270        """Get the sequence folder and all run folders (ending with .D).
271
272        :return: `SequenceDataFiles` containing complete path locations for sequence folder and all runs.
273        """
274        if len(self.sequence_controller.data_files):
275            self.sequence_controller._fuzzy_match_most_recent_folder(
276                most_recent_folder=self.sequence_controller.data_files[-1],
277            )
278            return self.sequence_controller.data_files[-1]
279        else:
280            raise UserWarning("No data files yet!")
281
282    def get_last_run_sequence_reports(
283        self,
284        custom_path: Optional[str] = None,
285        report_type: ReportType = ReportType.CSV,
286    ) -> List[AgilentReport]:
287        """Return data contained in the REPORT files. Use `aghplctools` if you want more report processing utility.
288
289        :param custom_path: path to sequence folder
290        :param report_type: read either the TXT or CSV version
291        :return: list of reports for each row
292        """
293        return self.sequence_controller.get_report(
294            custom_path=custom_path, report_type=report_type
295        )
296
297    def get_last_run_sequence_data(
298        self, read_uv: bool = False, custom_path: Optional[str] = None
299    ) -> (
300        List[Dict[int, AgilentHPLCChromatogram]] | List[AgilentChannelChromatogramData]
301    ):
302        """Returns data for all rows in the last run sequence data.
303
304        :param custom_path: If you want to just load sequence data but from a file path. This file path must be the complete file path.
305        :param read_uv: whether to also read the UV file
306        """
307        if read_uv:
308            return self.sequence_controller.get_data_mult_uv(custom_path=custom_path)
309        else:
310            return self.sequence_controller.get_data(custom_path=custom_path)
311
312    def check_loaded_sequence(self) -> str:
313        """Returns the name of the currently loaded sequence."""
314        return self.sequence_controller.get_current_sequence_name()
315
316    def check_loaded_method(self) -> str:
317        """Returns the name of the currently loaded method."""
318        return self.method_controller.get_current_method_name()
319
320    def load_method(self) -> MethodDetails:
321        """Returns details of the currently loaded method, such as its starting modifier conditions and timetable."""
322        return self.method_controller.load()
323
324    def load_sequence(self) -> SequenceTable:
325        """Returns the currently loaded sequence."""
326        return self.sequence_controller.load()
327
328    def load_injector_program(self) -> InjectorTable | None:
329        return self.method_controller.injector.load()
330
331    def load_sample_location(self):
332        self.method_controller.get_location()
333
334    def standby(self):
335        """Switches all modules in standby mode. All lamps and pumps are switched off."""
336        self.send(Command.STANDBY_CMD)
337
338    def preprun(self):
339        """Prepares all modules for run. All lamps and pumps are switched on."""
340        self.send(Command.PREPRUN_CMD)
341
342    def lamp_on(self):
343        """Turns the UV lamp on."""
344        self.method_controller.dad.turn_on()
345
346    def lamp_off(self):
347        """Turns the UV lamp off."""
348        self.method_controller.dad.turn_off()
349
350    def pump_on(self):
351        """Turns on the pump on."""
352        self.method_controller.pump.turn_on()
353
354    def pump_off(self):
355        """Turns the pump off."""
356        self.method_controller.pump.turn_off()
357
358    def instrument_off(self):
359        """Shuts the entire instrument off, including pumps, lamps, thermostat."""
360        self.send(Command.INSTRUMENT_OFF)
361
362    def instrument_on(self):
363        """Turns the entire instrument on, including pumps, lamps, thermostat."""
364        self.send(Command.INSTRUMENT_ON)