pychemstation.utils.abc_tables.table

  1"""
  2Abstract module containing shared logic for Method and Sequence tables.
  3
  4Authors: Lucy Hao
  5"""
  6
  7from __future__ import annotations
  8
  9import abc
 10from abc import abstractmethod
 11from collections.abc import Callable
 12from typing import Optional, Union
 13
 14from result import Err, Result
 15
 16from ..macro import Command, Response
 17from ..method_types import MethodDetails, PType, Param
 18from ..sequence_types import SequenceTable
 19from ..table_types import Table, RegisterFlag, TableOperation, Device
 20from ...control.controllers import CommunicationController
 21
 22TableType = Union[MethodDetails, SequenceTable]
 23
 24
 25class ABCTableController(abc.ABC):
 26    """Abstract controller for all table-like objects in Chemstation.
 27    :param controller: controller for sending MACROs to Chemstation
 28    :param table: contains register keys needed for accessing table in Chemstation.
 29    """
 30
 31    def __init__(
 32        self,
 33        controller: Optional[CommunicationController],
 34        table: Table | Device,
 35    ):
 36        self.controller = controller
 37        self.table_locator = table
 38        self.table_state: Optional[TableType] = None
 39
 40    def __new__(cls, *args, **kwargs):
 41        if cls is ABCTableController:
 42            raise TypeError(f"only children of '{cls.__name__}' may be instantiated")
 43        return object.__new__(cls, *args, **kwargs)
 44
 45    @abstractmethod
 46    def download(self):
 47        pass
 48
 49    @abc.abstractmethod
 50    def get_row(self, row: int):
 51        pass
 52
 53    def receive(self) -> Result[Response, str]:
 54        if self.controller:
 55            for _ in range(10):
 56                try:
 57                    return self.controller.receive()
 58                except IndexError:
 59                    continue
 60            return Err("Could not parse response")
 61        else:
 62            raise ValueError("Controller is offline!")
 63
 64    def send(self, cmd: Union[Command, str]):
 65        if not self.controller:
 66            raise RuntimeError(
 67                "Communication controller must be initialized before sending command. It is currently in offline mode."
 68            )
 69        self.controller.send(cmd)
 70
 71    def sleepy_send(self, cmd: Union[Command, str]):
 72        if self.controller:
 73            self.controller.sleepy_send(cmd)
 74        else:
 75            raise ValueError("Controller is offline")
 76
 77    def sleep(self, seconds: int):
 78        """Tells the HPLC to wait for a specified number of seconds.
 79
 80        :param seconds: number of seconds to wait
 81        """
 82        self.send(Command.SLEEP_CMD.value.format(seconds=seconds))
 83
 84    def get_num(self, row: int, col_name: RegisterFlag) -> Union[int, float]:
 85        if isinstance(self.table_locator, Table) and self.controller:
 86            return self.controller.get_num_val(
 87                TableOperation.GET_ROW_VAL.value.format(
 88                    register=self.table_locator.register,
 89                    table_name=self.table_locator.name,
 90                    row=row,
 91                    col_name=col_name.value,
 92                )
 93            )
 94        else:
 95            raise ValueError("Controller is offline")
 96
 97    def get_text(self, row: int, col_name: RegisterFlag) -> str:
 98        if isinstance(self.table_locator, Table) and self.controller:
 99            return self.controller.get_text_val(
100                TableOperation.GET_ROW_TEXT.value.format(
101                    register=self.table_locator.register,
102                    table_name=self.table_locator.name,
103                    row=row,
104                    col_name=col_name.value,
105                )
106            )
107        else:
108            raise ValueError("Controller is offline")
109
110    def add_new_col_num(self, col_name: RegisterFlag, val: Union[int, float]):
111        if not (isinstance(val, int) or isinstance(val, float)):
112            raise ValueError(f"{val} must be an int or float.")
113        if isinstance(self.table_locator, Table) and self.controller:
114            self.sleepy_send(
115                TableOperation.NEW_COL_VAL.value.format(
116                    register=self.table_locator.register,
117                    table_name=self.table_locator.name,
118                    col_name=col_name,
119                    val=val,
120                )
121            )
122        else:
123            raise ValueError("require table, not device")
124
125    def add_new_col_text(self, col_name: RegisterFlag, val: str):
126        if not isinstance(val, str):
127            raise ValueError(f"{val} must be a str.")
128        if isinstance(self.table_locator, Table) and self.controller:
129            self.sleepy_send(
130                TableOperation.NEW_COL_TEXT.value.format(
131                    register=self.table_locator.register,
132                    table_name=self.table_locator.name,
133                    col_name=col_name,
134                    val=val,
135                )
136            )
137        else:
138            raise ValueError("require table not device")
139
140    def _edit_row_num(
141        self, col_name: RegisterFlag, val: Union[int, float], row: Optional[int] = None
142    ):
143        if not (isinstance(val, int) or isinstance(val, float)):
144            raise ValueError(f"{val} must be an int or float.")
145        num_rows = self.get_row_count_safely()
146        if row and num_rows < row:
147            raise ValueError("Not enough rows to edit!")
148
149        if isinstance(self.table_locator, Table) and self.controller:
150            self.sleepy_send(
151                TableOperation.EDIT_ROW_VAL.value.format(
152                    register=self.table_locator.register,
153                    table_name=self.table_locator.name,
154                    row=row if row is not None else "response_num",
155                    col_name=col_name,
156                    val=val,
157                )
158            )
159
160    def _edit_row_text(
161        self, col_name: RegisterFlag, val: str, row: Optional[int] = None
162    ):
163        if not isinstance(val, str):
164            raise ValueError(f"{val} must be a str.")
165        num_rows = self.get_row_count_safely()
166        if row and num_rows < row:
167            raise ValueError("Not enough rows to edit!")
168
169        if isinstance(self.table_locator, Table) and self.controller:
170            self.sleepy_send(
171                TableOperation.EDIT_ROW_TEXT.value.format(
172                    register=self.table_locator.register,
173                    table_name=self.table_locator.name,
174                    row=row if row is not None else "response_num",
175                    col_name=col_name,
176                    val=val,
177                )
178            )
179
180    def delete_row(self, row: int):
181        if isinstance(self.table_locator, Table) and self.controller:
182            self.sleepy_send(
183                TableOperation.DELETE_ROW.value.format(
184                    register=self.table_locator.register,
185                    table_name=self.table_locator.name,
186                    row=row,
187                )
188            )
189        else:
190            raise ValueError("controller is offline or given device, need table")
191
192    def get_row_count_safely(self) -> int:
193        row_count = self.get_num_rows()
194        tries = 10
195        i = 0
196        while row_count.is_err() and i < tries:
197            row_count = self.get_num_rows()
198            i += 1
199        if row_count.is_ok():
200            return int(row_count.ok_value.num_response)
201        else:
202            raise ValueError("couldn't read row count, table might not exist")
203
204    def add_row(self):
205        """Adds a row to the provided table for currently loaded method or sequence."""
206        previous_row_count = self.get_row_count_safely()
207        self.sleepy_send(
208            TableOperation.NEW_ROW.value.format(
209                register=self.table_locator.register, table_name=self.table_locator.name
210            )
211        )
212        new_row_count = self.get_row_count_safely()
213        if previous_row_count + 1 != new_row_count:
214            raise ValueError("Row could not be added.")
215
216    def delete_table(self):
217        """Deletes the table."""
218        self.sleepy_send(
219            TableOperation.DELETE_TABLE.value.format(
220                register=self.table_locator.register, table_name=self.table_locator.name
221            )
222        )
223
224    def new_table(self):
225        """Creates the table."""
226        self.send(
227            TableOperation.CREATE_TABLE.value.format(
228                register=self.table_locator.register, table_name=self.table_locator.name
229            )
230        )
231
232    def get_num_rows(self) -> Result[Response, str]:
233        if isinstance(self.table_locator, Table) and self.controller:
234            self.send(
235                Command.GET_ROWS_CMD.value.format(
236                    register=self.table_locator.register,
237                    table_name=self.table_locator.name,
238                    col_name=RegisterFlag.NUM_ROWS,
239                )
240            )
241            res = self.controller.receive()
242            if res.is_ok():
243                return res
244            else:
245                return Err("No rows could be read.")
246        else:
247            raise ValueError(
248                "controller was offline or was given a device and not a table"
249            )
250
251    def move_row(self, from_row: int, to_row: int):
252        if isinstance(self.table_locator, Table) and self.controller:
253            self.send(
254                TableOperation.MOVE_ROW.value.format(
255                    register=self.table_locator.register,
256                    table_name=self.table_locator.name,
257                    from_row=from_row,
258                    to_row=to_row,
259                )
260            )
261        else:
262            raise ValueError("controller is offline or given device, need table")
263
264    def _read_str_param(self, register_flag: RegisterFlag):
265        if self.controller:
266            try:
267                return self.controller.get_text_val(
268                    cmd=TableOperation.GET_OBJ_HDR_TEXT.value.format(
269                        register=self.table_locator.register,
270                        register_flag=register_flag,
271                    )
272                )
273            except RuntimeError:
274                return self.controller.get_text_val(
275                    cmd=TableOperation.GET_OBJ_HDR_TEXT.value.format(
276                        register=self.table_locator.register + "[2]",
277                        register_flag=register_flag,
278                    )
279                )
280        raise ValueError("Communication controller is not online!")
281
282    def _read_num_param(self, register_flag: RegisterFlag):
283        if self.controller:
284            return self.controller.get_num_val(
285                cmd=TableOperation.GET_OBJ_HDR_VAL.value.format(
286                    register=self.table_locator.register,
287                    register_flag=register_flag,
288                )
289            )
290        raise ValueError("Communication controller is not online!")
291
292    def _update_param(
293        self, param: Param, register_num: Optional[int] = None, sleep: bool = True
294    ):
295        register = self.table_locator.register
296        setting_command = (
297            TableOperation.UPDATE_OBJ_HDR_VAL
298            if param.ptype == PType.NUM
299            else TableOperation.UPDATE_OBJ_HDR_TEXT
300        )
301        send_method: Callable = self.sleepy_send if sleep else self.send
302        if isinstance(param.chemstation_key, list):
303            for register_flag in param.chemstation_key:
304                send_method(
305                    setting_command.value.format(
306                        register=f"{register}[{str(register_num)}]"
307                        if register_num
308                        else register,
309                        register_flag=register_flag,
310                        val=param.val,
311                    )
312                )
313        else:
314            register_flag = param.chemstation_key
315            send_method(
316                setting_command.value.format(
317                    register=f"{register}[{str(register_num)}]"
318                    if register_num
319                    else register,
320                    register_flag=register_flag,
321                    val=param.val,
322                )
323            )
324        self.download()