Coverage for / opt / hostedtoolcache / Python / 3.11.14 / x64 / lib / python3.11 / site-packages / rattlesnake / components / abstract_environment.py: 85%
159 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-27 18:22 +0000
1# -*- coding: utf-8 -*-
2"""
3Abstract environment that can be used to create new environment control strategies
4in the controller.
6Rattlesnake Vibration Control Software
7Copyright (C) 2021 National Technology & Engineering Solutions of Sandia, LLC
8(NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S.
9Government retains certain rights in this software.
11This program is free software: you can redistribute it and/or modify
12it under the terms of the GNU General Public License as published by
13the Free Software Foundation, either version 3 of the License, or
14(at your option) any later version.
16This program is distributed in the hope that it will be useful,
17but WITHOUT ANY WARRANTY; without even the implied warranty of
18MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19GNU General Public License for more details.
21You should have received a copy of the GNU General Public License
22along with this program. If not, see <https://www.gnu.org/licenses/>.
23"""
25import multiprocessing as mp
26import multiprocessing.sharedctypes # pylint: disable=unused-import
27import os
28import traceback
29from abc import ABC, abstractmethod
30from datetime import datetime
31from multiprocessing.queues import Queue
33import netCDF4 as nc4
34import openpyxl
36from .utilities import DataAcquisitionParameters, GlobalCommands, VerboseMessageQueue
38PICKLE_ON_ERROR = False
40if PICKLE_ON_ERROR:
41 import pickle
44class AbstractMetadata(ABC):
45 """Abstract class for storing metadata for an environment.
47 This class is used as a storage container for parameters used by an
48 environment. It is returned by the environment UI's
49 ``collect_environment_definition_parameters`` function as well as its
50 ``initialize_environment`` function. Various parts of the controller and
51 environment will query the class's data members for parameter values.
53 Classes inheriting from AbstractMetadata must define:
54 1. store_to_netcdf - A function defining the way the parameters are
55 stored to a netCDF file saved during streaming operations.
56 """
58 @abstractmethod
59 def store_to_netcdf(
60 self, netcdf_group_handle: nc4._netCDF4.Group
61 ): # pylint: disable=c-extension-no-member
62 """Store parameters to a group in a netCDF streaming file.
64 This function stores parameters from the environment into the netCDF
65 file in a group with the environment's name as its name. The function
66 will receive a reference to the group within the dataset and should
67 store the environment's parameters into that group in the form of
68 attributes, dimensions, or variables.
70 This function is the "write" counterpart to the retrieve_metadata
71 function in the AbstractUI class, which will read parameters from
72 the netCDF file to populate the parameters in the user interface.
74 Parameters
75 ----------
76 netcdf_group_handle : nc4._netCDF4.Group
77 A reference to the Group within the netCDF dataset where the
78 environment's metadata is stored.
80 """
83class AbstractUI(ABC):
84 """Abstract User Interface class defining the interface with the controller
86 This class is used to define the interface between the User Interface of a
87 environment in the controller and the main controller."""
89 @abstractmethod
90 def __init__(
91 self,
92 environment_name: str,
93 environment_command_queue: VerboseMessageQueue,
94 controller_communication_queue: VerboseMessageQueue,
95 log_file_queue: Queue,
96 ):
97 """
98 Stores data required by the controller to interact with the UI
100 This class stores data required by the controller to interact with the
101 user interface for a given environment. This includes the environment
102 name and queues to pass information between the controller and
103 environment. It additionally initializes the ``command_map`` which is
104 used by the Test Profile functionality to map profile instructions to
105 operations on the user interface.
108 Parameters
109 ----------
110 environment_name : str
111 The name of the environment
112 environment_command_queue : VerboseMessageQueue
113 A queue that will provide instructions to the corresponding
114 environment
115 controller_communication_queue : VerboseMessageQueue
116 The queue that relays global communication messages to the controller
117 log_file_queue : Queue
118 The queue that will be used to put messages to the log file.
121 """
122 self._environment_name = environment_name
123 self._log_name = environment_name + " UI"
124 self._log_file_queue = log_file_queue
125 self._environment_command_queue = environment_command_queue
126 self._controller_communication_queue = controller_communication_queue
127 self._command_map = {
128 "Start Control": self.start_control,
129 "Stop Control": self.stop_control,
130 }
132 @property
133 def command_map(self) -> dict:
134 """Dictionary mapping profile instructions to functions of the UI that
135 are called when the instruction is executed."""
136 return self._command_map
138 @abstractmethod
139 def start_control(self):
140 """Runs the corresponding environment in the controller"""
142 @abstractmethod
143 def stop_control(self):
144 """Stops the corresponding environment in the controller"""
146 @abstractmethod
147 def collect_environment_definition_parameters(self) -> AbstractMetadata:
148 """
149 Collect the parameters from the user interface defining the environment
151 Returns
152 -------
153 AbstractMetadata
154 A metadata or parameters object containing the parameters defining
155 the corresponding environment.
157 """
159 @abstractmethod
160 def initialize_data_acquisition(self, data_acquisition_parameters: DataAcquisitionParameters):
161 """Update the user interface with data acquisition parameters
163 This function is called when the Data Acquisition parameters are
164 initialized. This function should set up the environment user interface
165 accordingly.
167 Parameters
168 ----------
169 data_acquisition_parameters : DataAcquisitionParameters :
170 Container containing the data acquisition parameters, including
171 channel table and sampling information.
173 """
175 @abstractmethod
176 def initialize_environment(self) -> AbstractMetadata:
177 """
178 Update the user interface with environment parameters
180 This function is called when the Environment parameters are initialized.
181 This function should set up the user interface accordingly. It must
182 return the parameters class of the environment that inherits from
183 AbstractMetadata.
185 Returns
186 -------
187 AbstractMetadata
188 An AbstractMetadata-inheriting object that contains the parameters
189 defining the environment.
191 """
193 @abstractmethod
194 def retrieve_metadata(
195 self, netcdf_handle: nc4._netCDF4.Dataset
196 ): # pylint: disable=c-extension-no-member
197 """Collects environment parameters from a netCDF dataset.
199 This function retrieves parameters from a netCDF dataset that was written
200 by the controller during streaming. It must populate the widgets
201 in the user interface with the proper information.
203 This function is the "read" counterpart to the store_to_netcdf
204 function in the AbstractMetadata class, which will write parameters to
205 the netCDF file to document the metadata.
207 Note that the entire dataset is passed to this function, so the function
208 should collect parameters pertaining to the environment from a Group
209 in the dataset sharing the environment's name, e.g.
211 ``group = netcdf_handle.groups[self.environment_name]``
212 ``self.definition_widget.parameter_selector.setValue(group.parameter)``
214 Parameters
215 ----------
216 netcdf_handle : nc4._netCDF4.Dataset :
217 The netCDF dataset from which the data will be read. It should have
218 a group name with the enviroment's name.
220 """
222 @abstractmethod
223 def update_gui(self, queue_data: tuple):
224 """Update the environment's graphical user interface
226 This function will receive data from the gui_update_queue that
227 specifies how the user interface should be updated. Data will usually
228 be received as ``(instruction,data)`` pairs, where the ``instruction`` notes
229 what operation should be taken or which widget should be modified, and
230 the ``data`` notes what data should be used in the update.
232 Parameters
233 ----------
234 queue_data : tuple
235 A tuple containing ``(instruction,data)`` pairs where ``instruction``
236 defines and operation or widget to be modified and ``data`` contains
237 the data used to perform the operation.
238 """
240 @property
241 def log_file_queue(self) -> Queue:
242 """A property containing a reference to the queue accepting messages
243 that will be written to the log file"""
244 return self._log_file_queue
246 @property
247 def environment_command_queue(self) -> VerboseMessageQueue:
248 """A property containing a reference to the queue accepting commands
249 that will be delivered to the environment"""
250 return self._environment_command_queue
252 @property
253 def controller_communication_queue(self) -> VerboseMessageQueue:
254 """A property containing a reference to the queue accepting global
255 commands that will be delivered to the controller"""
256 return self._controller_communication_queue
258 @property
259 def environment_name(self):
260 """A property containing the environment's name"""
261 return self._environment_name
263 @property
264 def log_name(self):
265 """A property containing the name that the UI will be referenced by in
266 the log file, which will typically be ``self.environment_name + ' UI'``"""
267 return self._log_name
269 def log(self, message: str):
270 """Write a message to the log file
272 This function puts a message onto the ``log_file_queue`` so it will
273 eventually be written to the log file.
275 When written to the log file, the message will include the date and
276 time that the message was queued, the name that the UI uses in the log
277 file (``self.log_file``), and then the message itself.
279 Parameters
280 ----------
281 message : str :
282 A message that will be written to the log file.
284 """
285 self.log_file_queue.put(f"{datetime.now()}: {self.log_name} -- {message}\n")
287 @staticmethod
288 @abstractmethod
289 def create_environment_template(
290 environment_name: str, workbook: openpyxl.workbook.workbook.Workbook
291 ):
292 """Creates a template worksheet in an Excel workbook defining the
293 environment.
295 This function creates a template worksheet in an Excel workbook that
296 when filled out could be read by the controller to re-create the
297 environment.
299 This function is the "write" counterpart to the
300 ``set_parameters_from_template`` function in the ``AbstractUI`` class,
301 which reads the values from the template file to populate the user
302 interface.
304 Parameters
305 ----------
306 environment_name : str :
307 The name of the environment that will specify the worksheet's name
308 workbook : openpyxl.workbook.workbook.Workbook :
309 A reference to an ``openpyxl`` workbook.
311 """
313 @abstractmethod
314 def set_parameters_from_template(self, worksheet: openpyxl.worksheet.worksheet.Worksheet):
315 """
316 Collects parameters for the user interface from the Excel template file
318 This function reads a filled out template worksheet to create an
319 environment. Cells on this worksheet contain parameters needed to
320 specify the environment, so this function should read those cells and
321 update the UI widgets with those parameters.
323 This function is the "read" counterpart to the
324 ``create_environment_template`` function in the ``AbstractUI`` class,
325 which writes a template file that can be filled out by a user.
328 Parameters
329 ----------
330 worksheet : openpyxl.worksheet.worksheet.Worksheet
331 An openpyxl worksheet that contains the environment template.
332 Cells on this worksheet should contain the parameters needed for the
333 user interface.
335 """
338class AbstractEnvironment(ABC):
339 """Abstract Environment class defining the interface with the controller
341 This class is used to define the operation of an environment within the
342 controller, which must be completed by subclasses inheriting from this
343 class. Children of this class will sit in a While loop in the
344 ``AbstractEnvironment.run()`` function. While in this loop, the
345 Environment will pull instructions and data from the
346 ``command_queue`` and then use the ``command_map`` to map those instructions
347 to functions in the class.
349 All child classes inheriting from AbstractEnvironment will require functions
350 to be defined for global operations of the controller, which are already
351 mapped in the ``command_map``. Any additional operations must be defined
352 by functions and then added to the command_map when initilizing the child
353 class.
355 All functions called via the ``command_map`` must accept one input argument
356 which is the data passed along with the command. For functions that do not
357 require additional data, this argument can be ignored, but it must still be
358 present in the function's calling signature.
360 The run function will continue until one of the functions called by
361 ``command_map`` returns a truthy value, which signifies the controller to
362 quit. Therefore, any functions mapped to ``command_map`` that should not
363 instruct the program to quit should not return any value that could be
364 interpreted as true."""
366 def dump_to_dict(self):
367 """Dumps the environment to a dictionary to be pickled if an error occurs"""
368 if PICKLE_ON_ERROR:
369 state = self.__dict__.copy()
370 for key, value in state.items():
371 try:
372 pickle.dumps(value)
373 print(f"{key} is pickleable")
374 except Exception: # pylint: disable=broad-exception-caught
375 print(f"{key} is not pickleable")
376 state[key] = None
377 return state
378 else:
379 return self.__dict__.copy()
381 def __init__(
382 self,
383 environment_name: str,
384 command_queue: VerboseMessageQueue,
385 gui_update_queue: Queue,
386 controller_communication_queue: VerboseMessageQueue,
387 log_file_queue: Queue,
388 data_in_queue: Queue,
389 data_out_queue: Queue,
390 acquisition_active: mp.sharedctypes.Synchronized,
391 output_active: mp.sharedctypes.Synchronized,
392 ):
393 self._environment_name = environment_name
394 self._command_queue = command_queue
395 self._gui_update_queue = gui_update_queue
396 self._controller_communication_queue = controller_communication_queue
397 self._log_file_queue = log_file_queue
398 self._data_in_queue = data_in_queue
399 self._data_out_queue = data_out_queue
400 self._command_map = {
401 GlobalCommands.QUIT: self.quit,
402 GlobalCommands.INITIALIZE_DATA_ACQUISITION: self.initialize_data_acquisition_parameters,
403 GlobalCommands.INITIALIZE_ENVIRONMENT_PARAMETERS: self.initialize_environment_test_parameters,
404 GlobalCommands.STOP_ENVIRONMENT: self.stop_environment,
405 }
406 self._acquisition_active = acquisition_active
407 self._output_active = output_active
409 @property
410 def acquisition_active(self):
411 """Flag to check if acquisition is active"""
412 # print('Checking if Acquisition Active: {:}'.format(bool(self._acquisition_active.value)))
413 return bool(self._acquisition_active.value)
415 @property
416 def output_active(self):
417 """Flag to check if output is active"""
418 # print('Checking if Output Active: {:}'.format(bool(self._output_active.value)))
419 return bool(self._output_active.value)
421 @abstractmethod
422 def initialize_data_acquisition_parameters(
423 self, data_acquisition_parameters: DataAcquisitionParameters
424 ):
425 """Initialize the data acquisition parameters in the environment.
427 The environment will receive the global data acquisition parameters from
428 the controller, and must set itself up accordingly.
430 Parameters
431 ----------
432 data_acquisition_parameters : DataAcquisitionParameters :
433 A container containing data acquisition parameters, including
434 channels active in the environment as well as sampling parameters.
435 """
437 @abstractmethod
438 def initialize_environment_test_parameters(self, environment_parameters: AbstractMetadata):
439 """
440 Initialize the environment parameters specific to this environment
442 The environment will recieve parameters defining itself from the
443 user interface and must set itself up accordingly.
445 Parameters
446 ----------
447 environment_parameters : AbstractMetadata
448 A container containing the parameters defining the environment
450 """
452 @abstractmethod
453 def stop_environment(self, data):
454 """Stop the environment gracefully
456 This function defines the operations to shut down the environment
457 gracefully so there is no hard stop that might damage test equipment
458 or parts.
460 Parameters
461 ----------
462 data : Ignored
463 This parameter is not used by the function but must be present
464 due to the calling signature of functions called through the
465 ``command_map``
467 """
469 @property
470 def environment_command_queue(self) -> VerboseMessageQueue:
471 """The queue that provides commands to the environment."""
472 return self._command_queue
474 @property
475 def data_in_queue(self) -> Queue:
476 """The queue from which data is delivered to the environment"""
477 return self._data_in_queue
479 @property
480 def data_out_queue(self) -> Queue:
481 """The queue to which data is written that will be output to exciters"""
482 return self._data_out_queue
484 @property
485 def gui_update_queue(self) -> Queue:
486 """The queue that GUI update instructions are written to"""
487 return self._gui_update_queue
489 @property
490 def controller_communication_queue(self) -> Queue:
491 """The queue that global controller updates are written to"""
492 return self._controller_communication_queue
494 @property
495 def log_file_queue(self) -> Queue:
496 """The queue that log file messages are written to"""
497 return self._log_file_queue
499 def log(self, message: str):
500 """Write a message to the log file
502 This function puts a message onto the ``log_file_queue`` so it will
503 eventually be written to the log file.
505 When written to the log file, the message will include the date and
506 time that the message was queued, the name of the environment, and
507 then the message itself.
509 Parameters
510 ----------
511 message : str :
512 A message that will be written to the log file.
513 """
514 self.log_file_queue.put(f"{datetime.now()}: {self.environment_name} -- {message}\n")
516 @property
517 def environment_name(self) -> str:
518 """A string defining the name of the environment"""
519 return self._environment_name
521 @property
522 def command_map(self) -> dict:
523 """A dictionary that maps commands received by the ``command_queue`` to functions in the class"""
524 return self._command_map
526 def map_command(self, key, function):
527 """A function that maps an instruction to a function in the ``command_map``
529 Parameters
530 ----------
531 key :
532 The instruction that will be pulled from the ``command_queue``
534 function :
535 A reference to the function that will be called when the ``key``
536 message is received.
538 """
539 self._command_map[key] = function
541 def run(self):
542 """The main function that is run by the environment's process
544 A function that is called by the environment's process function that
545 sits in a while loop waiting for instructions on the command queue.
547 When the instructions are recieved, they are separated into
548 ``(message,data)`` pairs. The ``message`` is used in conjuction with
549 the ``command_map`` to identify which function should be called, and
550 the ``data`` is passed to that function as the argument. If the
551 function returns a truthy value, it signals to the ``run`` function
552 that it is time to stop the loop and exit.
555 """
556 self.log(f"Starting Process with PID {os.getpid()}")
557 while True:
558 # Get the message from the queue
559 message, data = self.environment_command_queue.get(self.environment_name)
560 # Call the function corresponding to that message with the data as argument
561 try:
562 function = self.command_map[message]
563 except KeyError:
564 self.log(
565 f"Undefined Message {message}, acceptable messages are {[key for key in self.command_map]}"
566 )
567 continue
568 try:
569 halt_flag = function(data)
570 except Exception: # pylint: disable=broad-exception-caught
571 tb = traceback.format_exc()
572 self.log(f"ERROR\n\n {tb}")
573 self.gui_update_queue.put(
574 (
575 "error",
576 (
577 f"{self.environment_name} Error",
578 f"!!!UNKNOWN ERROR!!!\n\n{tb}",
579 ),
580 )
581 )
582 if PICKLE_ON_ERROR:
583 with open(
584 f"debug_data/{self.environment_name}_error_state.txt", "w", encoding="utf-8"
585 ) as f:
586 f.write(f"{tb}")
587 with open(f"debug_data/{self.environment_name}_error_state.pkl", "wb") as f:
588 dic = self.dump_to_dict()
589 pickle.dump(dic, f)
590 print("Done Writing Pickle File from Error...")
591 halt_flag = False
592 # If we get a true value, stop.
593 if halt_flag:
594 self.log("Stopping Process")
595 break
597 def quit(self, data): # pylint: disable=unused-argument
598 """Returns True to stop the ``run`` while loop and exit the process
600 Parameters
601 ----------
602 data : Ignored
603 This parameter is not used by the function but must be present
604 due to the calling signature of functions called through the
605 ``command_map``
607 Returns
608 -------
609 True :
610 This function returns True to signal to the ``run`` while loop
611 that it is time to close down the environment.
613 """
614 return True
617def run_process(
618 environment_name: str,
619 input_queue: VerboseMessageQueue,
620 gui_update_queue: Queue,
621 controller_communication_queue: VerboseMessageQueue,
622 log_file_queue: Queue,
623 data_in_queue: Queue,
624 data_out_queue: Queue,
625 acquisition_active: mp.sharedctypes.Synchronized,
626 output_active: mp.sharedctypes.Synchronized,
627):
628 """A function called by ``multiprocessing.Process`` to start the environment
630 This function should not be called directly, but used as a template for
631 other environments to start up.
633 Parameters
634 ----------
635 environment_name : str :
636 The name of the environment
638 input_queue : VerboseMessageQueue :
639 The command queue for the environment
641 gui_update_queue : Queue :
642 The queue that accepts GUI update ``(message,data)`` pairs.
644 controller_communication_queue : VerboseMessageQueue :
645 The queue where global instructions to the controller can be written
647 log_file_queue : Queue :
648 The queue where logging messages can be written
650 data_in_queue : Queue :
651 The queue from which the environment will receive data from the
652 acquisition hardware
654 data_out_queue : Queue :
655 The queue to which the environment should write data so it will be output
656 to the excitation devices in the output hardware
658 """
659 process_class = AbstractEnvironment( # pylint: disable=abstract-class-instantiated
660 environment_name,
661 input_queue,
662 gui_update_queue,
663 controller_communication_queue,
664 log_file_queue,
665 data_in_queue,
666 data_out_queue,
667 acquisition_active,
668 output_active,
669 )
670 process_class.run()