pychemstation.utils.abc_tables.run

  1"""
  2Abstract module containing shared logic for Method and Sequence tables.
  3
  4Authors: Lucy Hao
  5"""
  6
  7from __future__ import annotations
  8
  9import abc
 10import math
 11import os
 12import time
 13import warnings
 14from typing import Dict, List, Optional, Tuple, Union, Set
 15
 16import polling
 17import rainbow as rb
 18from result import Err, Ok, Result
 19
 20from ..macro import HPLCRunningStatus, Command
 21from ..method_types import MethodDetails
 22from ..sequence_types import SequenceTable
 23from ..table_types import Table, T
 24from ...analysis.chromatogram import (
 25    AgilentChannelChromatogramData,
 26    AgilentHPLCChromatogram,
 27)
 28
 29from .table import ABCTableController
 30from ...analysis.process_report import (
 31    ReportType,
 32    AgilentReport,
 33    CSVProcessor,
 34    TXTProcessor,
 35)
 36from ...control.controllers import CommunicationController
 37
 38TableType = Union[MethodDetails, SequenceTable]
 39
 40
 41class RunController(ABCTableController, abc.ABC):
 42    """Abstract controller for all tables that can trigger runs on Chemstation.
 43
 44    :param controller: the controller for sending MACROs, must be initialized for a run to be triggered.
 45    :param src: complete directory path where files containing run parameters are stored.
 46    :param data_dirs: list of complete directories that Chemstation will write data to.
 47    :param table: contains register keys for accessing table in Chemstation.
 48    :param offline: whether the communication controller is online.
 49    """
 50
 51    def __init__(
 52        self,
 53        controller: Optional[CommunicationController],
 54        src: Optional[str],
 55        data_dirs: Optional[List[str]],
 56        table: Table,
 57        offline: bool = False,
 58    ):
 59        super().__init__(controller=controller, table=table)
 60        self.table_state: Optional[TableType] = None
 61        self.curr_run_starting_time: Optional[float] = None
 62        self.timeout: Optional[float] = None
 63        self.current_run_child_files: Set[str] = set()
 64
 65        if not offline:
 66            if src and not os.path.isdir(src):
 67                raise FileNotFoundError(f"dir: {src} not found.")
 68            if data_dirs:
 69                for d in data_dirs:
 70                    if not os.path.isdir(d):
 71                        raise FileNotFoundError(f"dir: {d} not found.")
 72                    if r"\\" in d:
 73                        raise ValueError("Data directories should not be raw strings!")
 74            if src and data_dirs:
 75                self.src: str = src
 76                self.data_dirs: List[str] = data_dirs
 77
 78        self.spectra: dict[str, AgilentHPLCChromatogram] = {
 79            "A": AgilentHPLCChromatogram(),
 80            "B": AgilentHPLCChromatogram(),
 81            "C": AgilentHPLCChromatogram(),
 82            "D": AgilentHPLCChromatogram(),
 83            "E": AgilentHPLCChromatogram(),
 84            "F": AgilentHPLCChromatogram(),
 85            "G": AgilentHPLCChromatogram(),
 86            "H": AgilentHPLCChromatogram(),
 87        }
 88        self.uv: Dict[int, AgilentHPLCChromatogram] = {}
 89        self.data_files: List = []
 90
 91    def __new__(cls, *args, **kwargs):
 92        if cls is RunController:
 93            raise TypeError(f"only children of '{cls.__name__}' may be instantiated")
 94        return object.__new__(cls)
 95
 96    @abc.abstractmethod
 97    def _fuzzy_match_most_recent_folder(self, most_recent_folder: T) -> Result[T, str]:
 98        pass
 99
100    @abc.abstractmethod
101    def get_data(
102        self, custom_path: Optional[str] = None
103    ) -> Union[List[AgilentChannelChromatogramData], AgilentChannelChromatogramData]:
104        pass
105
106    @abc.abstractmethod
107    def get_data_uv(
108        self, custom_path: str | None = None
109    ) -> Dict[int, AgilentHPLCChromatogram]:
110        pass
111
112    @abc.abstractmethod
113    def get_report(
114        self, custom_path: str, report_type: ReportType = ReportType.TXT
115    ) -> List[AgilentReport]:
116        pass
117
118    def check_hplc_is_running(self) -> bool:
119        if self.controller:
120            try:
121                started_running = polling.poll(
122                    lambda: isinstance(self.controller.get_status(), HPLCRunningStatus),
123                    step=1,
124                    max_tries=20,
125                )
126            except Exception as e:
127                print(e)
128                return False
129            if started_running:
130                self.curr_run_starting_time = time.time()
131            return started_running
132        else:
133            raise ValueError("Controller is offline")
134
135    def check_hplc_run_finished(self) -> Tuple[float, bool]:
136        if self.controller:
137            try:
138                _, current_run_file = self.get_current_run_data_dir_file()
139                sample_file, extension, _ = current_run_file.partition(".D")
140                self.current_run_child_files.add(sample_file)
141            except Exception:
142                pass
143            done_running = self.controller.check_if_not_running()
144            if self.curr_run_starting_time and self.timeout:
145                time_passed = time.time() - self.curr_run_starting_time
146                if time_passed > self.timeout:
147                    enough_time_passed = time_passed >= self.timeout
148                    run_finished = enough_time_passed and done_running
149                    if run_finished:
150                        self._reset_time()
151                        return 0, run_finished
152                else:
153                    time_left = self.timeout - time_passed
154                    return time_left, self.controller.check_if_not_running()
155            return 0, self.controller.check_if_not_running()
156        raise ValueError("Controller is offline!")
157
158    def check_hplc_done_running(self) -> Ok[T] | Err[str]:
159        """Checks if ChemStation has finished running and can read data back
160
161        :return: Data file object containing most recent run file information.
162        """
163        self.current_run_child_files = set()
164        if self.timeout is not None:
165            finished_run = False
166            minutes = math.ceil(self.timeout / 60)
167            try:
168                finished_run = not polling.poll(
169                    lambda: self.check_hplc_run_finished()[1],
170                    max_tries=minutes - 1,
171                    step=50,
172                )
173            except (
174                polling.TimeoutException,
175                polling.PollingException,
176                polling.MaxCallException,
177            ):
178                try:
179                    finished_run = polling.poll(
180                        lambda: self.check_hplc_run_finished()[1],
181                        timeout=self.timeout / 2,
182                        step=1,
183                    )
184                except (
185                    polling.TimeoutException,
186                    polling.PollingException,
187                    polling.MaxCallException,
188                ):
189                    pass
190        else:
191            raise ValueError("Timeout value is None, no comparison can be made.")
192
193        check_folder = self._fuzzy_match_most_recent_folder(self.data_files[-1])
194        if check_folder.is_ok() and finished_run:
195            return check_folder
196        elif check_folder.is_ok():
197            try:
198                finished_run = polling.poll(
199                    lambda: self.check_hplc_run_finished()[1], max_tries=10, step=50
200                )
201                if finished_run:
202                    return check_folder
203            except Exception:
204                self._reset_time()
205                return self.data_files[-1]
206        return Err("Run not may not have completed.")
207
208    def get_uv_spectrum(self, path: str):
209        data_uv = rb.agilent.chemstation.parse_file(os.path.join(path, "DAD1.UV"))
210        times = data_uv.xlabels
211        wavelengths = data_uv.ylabels
212        absorbances = data_uv.data.transpose()
213        for i, w in enumerate(wavelengths):
214            self.uv[w] = AgilentHPLCChromatogram()
215            self.uv[w].attach_spectrum(times, absorbances[i])
216
217    def get_report_details(
218        self, path: str, report_type: ReportType = ReportType.TXT
219    ) -> AgilentReport:
220        if report_type is ReportType.TXT:
221            txt_report = TXTProcessor(path).process_report()
222            if txt_report.is_ok():
223                return txt_report.ok_value
224            elif txt_report.is_err():
225                raise ValueError(txt_report.err_value)
226        if report_type is ReportType.CSV:
227            csv_report = CSVProcessor(path).process_report()
228            if csv_report.is_ok():
229                return csv_report.ok_value
230            elif csv_report.is_err():
231                raise ValueError(csv_report.err_value)
232        raise ValueError("Expected one of ReportType.TXT or ReportType.CSV")
233
234    def get_spectrum_at_channels(self, data_path: str):
235        """Load chromatogram for any channel in spectra dictionary."""
236        for channel, spec in self.spectra.items():
237            try:
238                spec.load_spectrum(data_path=data_path, channel=channel)
239            except FileNotFoundError:
240                self.spectra[channel] = AgilentHPLCChromatogram()
241                warning = f"No data at channel: {channel}"
242                warnings.warn(warning)
243
244    def _reset_time(self):
245        self.curr_run_starting_time = None
246        self.timeout = None
247
248    def get_current_run_data_dir_file(self) -> Tuple[str, str]:
249        self.send(Command.GET_CURRENT_RUN_DATA_DIR)
250        full_path_name = self.receive()
251        self.send(Command.GET_CURRENT_RUN_DATA_FILE)
252        current_sample_file = self.receive()
253        if full_path_name.is_ok() and current_sample_file.is_ok():
254            if os.path.isdir(full_path_name.ok_value.string_response) and os.path.isdir(
255                os.path.join(
256                    full_path_name.ok_value.string_response,
257                    current_sample_file.ok_value.string_response,
258                )
259            ):
260                return (
261                    full_path_name.ok_value.string_response,
262                    current_sample_file.ok_value.string_response,
263                )
264        raise ValueError("Couldn't read data dir and file or doesn't exist yet.")