def__init__(self,working_dir="",distribution_file="generator.in",run_file="photo_track.in",log_dir="log",log_file_name="log.log",run_number=None)->None:super().__init__(working_dir,distribution_file,run_file,log_dir,log_file_name)# Set run_number as a property, either take from the argument given or load from run_filec=adjust_settings.ChangeSettings(self.run_file)ifrun_numberisnotNone:self.run_number=run_numberself.set_run_number(run_number)else:self.run_number=int(c.read_setting("RUN"))# ASTRA programmesself.astra_program=self.working_dir.joinpath("Astra.exe")self.generator_program=self.working_dir.joinpath("generator.exe")self.fieldplot_program=self.working_dir.joinpath("fieldplot.exe")self.lineplot_program=self.working_dir.joinpath("lineplot.exe")self.perform_checks()
def_find_parameter_definition_file(self,parameter:str)->pathlib.Path:"""Method to find whether the parameter is defined in the self.run_file or in the self.distribution file Args: parameter (str): Parameter, whose definition file is to be found. Returns: pathlib.Path: pathlib.Path to the definition file """# try to find it in the run_file, if not succesful (c.find_setting returns None), try distribution_filec=adjust_settings.ChangeSettings(self.run_file)ifc.find_setting(parameter)isNone:c=adjust_settings.ChangeSettings(self.distribution_file)returnself.distribution_file# this is to check if neither the parameter is neither in the run nor in the distribution fileifc.find_setting(parameter)isNone:self.logger.warning()(f"Parameter {parameter} was not found in the run nor the distribution file.")returnNonereturnself.run_file
defchange_parameter(self,parameter,new_value):"""Method to change a parameter in the input files. Args: file (str, path): Input file with the parameter. parameter (str): Parameter in the input file to be changed. new_value (str, int, float): New value of the parameter to be set. """# skip if parameter is Noneifstr(parameter).lower().strip()=="none":return# get the file where the parameter is definedparameter_file=self._find_parameter_definition_file(parameter)ifparameter_fileisNone:self.logger.warning(f"Parameter {parameter} was not changes because it was not found.")return# changing the parameterc=adjust_settings.ChangeSettings(parameter_file)c.update(parameter,str(new_value))self.logger.info(f"Updated parameter {parameter} to: {new_value} in {c.file}.")
defget_parameter_value(self,parameter:str)->str:"""Method to get the value of the parameter. Args: parameter (str): Parameter, whose value is to be found. Returns: str: Value of the parameter (if found). """# get the file where the parameter is defined parameter_file=self._find_parameter_definition_file(parameter)# if not found, return Noneifparameter_fileisNone:returnNoneelse:# read the setting and return itc=adjust_settings.ChangeSettings(parameter_file)returnc.read_setting(parameter)
defincrease_run_number(self):"""Method to increase the run number. """c=adjust_settings.ChangeSettings(self.run_file)c.update("RUN",self.run_number+1)self.run_number+=1self.logger.info(f"Run number increased to {self.run_number}.")
deflog_input_file(self,file_to_log):"""Method to log file to log directory. Args: file (str, path): File to be logged. """backed_input_file=self.log_dir.joinpath(file_to_log.name)# create filelock for save copyfile_to_log_lock=self.acquire_fileLock(file_to_log)backed_input_file_lock=self.acquire_fileLock(backed_input_file)withfile_to_log_lock:withbacked_input_file_lock:self.logger.info(f"Logging input file {file_to_log.name} to {backed_input_file}.")# TODO: check if the file does not already exist -> prevent rewriting the fileshutil.copy(file_to_log,backed_input_file)
defperform_checks(self)->None:"""Method to check if all ASTRA process can be run. Raises: Exception: ASTRA programs are missing in the directory. Exception: ASTRA and GENERATOR input files missing in the directory. """self.logger.debug("Running checks.")# check if ASTRA files existforfin[self.astra_program,self.generator_program,self.fieldplot_program,self.lineplot_program]:ifnotf.is_file():raiseException(f"Astra files are missing in the directory {self.working_dir}")# check if input files existforfin[self.distribution_file,self.run_file]:ifnotf.is_file():raiseException(f"Input files are missing in the directory {self.working_dir}")self.logger.debug("All checks completed - SUCCESS.")
defrun_astra(self,shell=True)->None:"""Runs the ASTRA programme on the run file. Args: shell (bool, optional): If True, ASTRA console output will be logged. Defaults to True. """self.logger.info(f"(RUN {self.run_number}) Running astra with input file: {self.run_file}")self.log_input_file(self.run_file)p=subprocess.Popen([self.astra_program,self.run_file.stem],shell=shell,stdin=subprocess.PIPE,stdout=subprocess.PIPE,cwd=self.working_dir)# b to make it a bitlike objectprocess_output,_=p.communicate(input=b'\n')# log subprocess outputforlineinprocess_output.split(b"\n"):self.logger.info(line)self.logger.info(f"(RUN {self.run_number}) Astra run completed.")
defrun_generator(self,shell=True)->None:"""Runs the GENERATOR programme on the generator file. Args: shell (bool, optional): If True, GENERATOR console output will be logged. Defaults to True. """self.logger.info(f"(RUN {self.run_number}) Running generator with input file: {self.distribution_file}")self.log_input_file(self.distribution_file)p=subprocess.Popen([self.generator_program,self.distribution_file.stem],shell=shell,stdin=subprocess.PIPE,stdout=subprocess.PIPE,cwd=self.working_dir)# b to make it a bitlike objectprocess_output,_=p.communicate(input=b'\n')# log subprocess outputforlineinprocess_output.split(b"\n"):self.logger.info(line)self.logger.info(f"(RUN {self.run_number}) Generator run completed.")
defset_run_number(self,new_run_number):"""Method to set the run number to the new_run_number. Args: new_run_number (str, int): New run number to be set. """c=adjust_settings.ChangeSettings(self.run_file)c.update("RUN",new_run_number)self.run_number=new_run_numberself.logger.info(f"Run number set to {new_run_number}.")
Last update:
October 31, 2023
Created:
October 31, 2023