core

Classes

Core

Core(
    working_dir="",
    distribution_file="generator.in",
    run_file="photo_track.in",
    log_dir="log",
    log_file_name="log.log",
    console_log=False,
    logger_name=None,
)

Bases: object

Parameters:

Name	Type	Description	Default
working_dir	str	Name of the working directory (can be relative path). Defaults to "".	''
distribution_file	str	Name of the ASTRA distribution file. Defaults to "generator.in".	'generator.in'
run_file	str	Name of the ASTRA run file. Defaults to "photo_track.in".	'photo_track.in'
log_dir	str	Name of the log directory, that will be created inside the working directory (can be relative path). Defaults to "log".	'log'
log_file_name	str	Name of the log file. Defaults to "log.log".	'log.log'
console_log	bool	If False, the log will not be written to the console. Defaults to False.	False
logger_name	str | None	Name of the logger for possible logger sharing. If set to None, new logger will be created. Defaults to None.	None

Source code in common/core.py

def __init__(self, working_dir = "", distribution_file="generator.in", run_file="photo_track.in", log_dir="log", log_file_name="log.log", console_log=False, logger_name=None) -> None:
    """Core class for ASTRA processes - correctly sets directories and log file

    Args:
        working_dir (str, optional): Name of the working directory (can be relative path). Defaults to "".
        distribution_file (str, optional): Name of the ASTRA distribution file. Defaults to "generator.in".
        run_file (str, optional): Name of the ASTRA run file. Defaults to "photo_track.in".
        log_dir (str, optional): Name of the log directory, that will be created inside the working directory (can be relative path). Defaults to "log".
        log_file_name (str, optional): Name of the log file. Defaults to "log.log".
        console_log (bool, optional): If False, the log will not be written to the console. Defaults to False.
        logger_name (str | None, optional): Name of the logger for possible logger sharing. If set to None, new logger will be created. Defaults to None.
    """
    # Load main config file and set data directory
    self.config = configuration.load_config()
    self.data_dir = pathlib.Path(self.config["Paths"]["data_dir"])

    # Set working directory
    self.working_dir = self.data_dir.joinpath(working_dir)
    self.working_dir.mkdir(parents=True, exist_ok=True)

    # Set input files
    self.distribution_file = self.working_dir.joinpath(distribution_file)
    self.run_file = self.working_dir.joinpath(run_file)

    # Setup logger, creates bunch of new variable, please check corresponding function
    self.setup_logger(log_dir, log_file_name, console_log, logger_name=logger_name)

    # this is the main astraLock used to handle parallel astraProcesses
    self.astraLock = self.acquire_fileLock(self.working_dir.joinpath("astraLock"))

Functions

_parameter_values_initialization

_parameter_values_initialization(
    parameter,
    values=None,
    start=None,
    end=None,
    step=None,
    n_steps=None,
)

Method to initialise the scan settings. To be used internally within methods of child classes. Based on what parameters are specified, the method calculates the rest of them.

For to method not to raise Exception, the parameter must be fully defined. This can be achieved by the following: 1. Full values list (values) 2. Start and end values (start, end) + step size (step) 3. Start and end values (start, end) + number of steps (n_steps)

This list also corresponds to the order at which the definition of the parameter is checked.

Once the parameter is fully defined (one of the points above is well inputed), the rest of inputed values may be internally overwritten to match the calculated values from the full definition.

Parameters:

Name	Type	Description	Default
parameter	str	Name of the scanning parameter.	required
values	list	Full list of values to be scanned. If specified, defines fully the parameter. Defaults to None.	None
start	float	Starting value of the scan. Will be taken into account only if values argument was not specified. For full definition of the parameter, the final argument along with step or n_steps must be specified. Defaults to None.	None
end	float	Final value of the scan. Will be taken into account only if values argument was not specified. For full definition of the parameter, the start argument along with step or n_steps must be specified. Defaults to None.	None
step	float | list	Step size to be performed. For constant step size, use float type, for varying step sizes, use list of floats. Will be taken into account only if values argument was not specified. For full definition of the parameter, the start and end arguments must be specified. Defaults to None.	None
n_steps	int	Number of constant steps to be performed. Will be taken into account only if values and steps arguments were not specified. For full definition of the parameter, the start and end arguments must be specified.. Defaults to None.	None

Returns:

Name	Type	Description
dict		Dictionary fully defining the scanning parameter. "parameter": Parameter name, "start": Start of the scan, "end": End of the scan, "values": Values of the scan, "step": Size of constant step, "steps": List of all steps, "n_steps": number of steps

Source code in common/core.py

def _parameter_values_initialization(self, parameter, values=None, start=None, end=None, step=None, n_steps=None):
    """Method to initialise the scan settings. To be used internally within methods of child classes.
    Based on what parameters are specified, the method calculates the rest of them. 

    For to method not to raise Exception, the parameter must be fully defined. This can be achieved by the following:
    1. Full values list (`values`)
    2. Start and end values (`start`, `end`) + step size (`step`)
    3. Start and end values (`start`, `end`) + number of steps (`n_steps`)

    This list also corresponds to the order at which the definition of the parameter is checked.

    Once the parameter is fully defined (one of the points above is well inputed), the rest of inputed values may be internally overwritten to match the calculated values from the full definition.

    Args:
        parameter (str): Name of the scanning parameter.
        values (list, optional): Full list of values to be scanned. If specified, defines fully the parameter. Defaults to None.
        start (float, optional): Starting value of the scan. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `final` argument along with `step` or `n_steps` must be specified. Defaults to None.
        end (float, optional): Final value of the scan. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `start` argument along with `step` or `n_steps` must be specified. Defaults to None.
        step (float | list, optional): Step size to be performed. For constant step size, use `float` type, for varying step sizes, use list of floats. Will be taken into account only if `values` argument was not specified. For full definition of the parameter, the `start` and `end` arguments must be specified. Defaults to None.
        n_steps (int, optional): Number of constant steps to be performed. Will be taken into account only if `values` and `steps` arguments were not specified. For full definition of the parameter, the `start` and `end` arguments must be specified.. Defaults to None.

    Returns:
        dict: Dictionary fully defining the scanning parameter.

            "parameter": Parameter name,
            "start": Start of the scan,
            "end": End of the scan,
            "values": Values of the scan,
            "step": Size of constant step,
            "steps": List of all steps,
            "n_steps": number of steps  
    """        
    if values is not None:
        if not isinstance(values, list):
            if not isinstance(values, np.ndarray):
                raise Exception(f"Values are expected to be a list, not {type(values)}")
        start = min(values)
        end = max(values)
        n_steps = len(values)
        steps = [values[i+1] - values[i] for i in range(n_steps)[:-1]]
        # initiate the step value
        if len(list(dict.fromkeys(steps))) != 1:
            # if it does not make sense (e.g. there are different steps), set this value to None
            step = None
        else:
            # if all step sizes are the same, set it
            step = steps[0]
    else:   
        if end < start:
            self.logger.error(f"Starting value of parameter {parameter} is larger than the end value (start: {start} > end: {end}.")
        # initialize step size and number of steps
        if step is None and n_steps is None:
            raise Exception(f"Neither step nor number of steps of {parameter} parameter set. Unable to perform the scan.")
        if step is not None:
            n_steps = int((end - start)//step) + 1
        else:
            step = float((Decimal(str(end)) - Decimal(str(start)))/Decimal(str(n_steps -1)))

        values = [float(Decimal(str(start)) + Decimal(str(i))*Decimal(str(step))) for i in range(n_steps)]
        steps = [step for i in range(n_steps)]

    return  {"parameter": parameter,
            "start": start,
            "end": end,
            "values": values,
            "step": step,
            "steps": steps,
            "n_steps": n_steps            
            }

acquire_fileLock

acquire_fileLock(fileName)

Method to obtain the lock of a file for multithreading.

Parameters:

Name	Type	Description	Default
fileName	str	Name of the file to be locked.	required

Returns:

Type	Description
FileLock	filelock.FileLock: Filelock instance for the file.

Source code in common/core.py

def acquire_fileLock(self, fileName:str) -> filelock.FileLock:
    """Method to obtain the lock of a file for multithreading.

    Args:
        fileName (str): Name of the file to be locked.

    Returns:
        filelock.FileLock: Filelock instance for the file.
    """
    # make sure the fileName is a Path instance
    fileName = pathlib.Path(fileName)
    # make sure the parent directory exists for correct lock file creation
    fileName.parent.mkdir(parents=True, exist_ok=True)
    # return the ready file lock
    return filelock.FileLock(fileName.parent / (fileName.name + ".lock"))

get_first_free_run_number

get_first_free_run_number()

Method to obtain the first free run number in the working directory.

Returns:

Name	Type	Description
int	int	Number of first free run number.

Source code in common/core.py

def get_first_free_run_number(self) -> int:
    """Method to obtain the first free run number in the working directory.

    Returns:
        int: Number of first free run number.
    """
    # try to get last run number from the run output file
    output_handler = self.get_output_file_handler(self.working_dir)
    if output_handler.output_content.shape[0] != 0:
        return output_handler.output_content["Run number"].max()+1

    # get max run number from ASTRA output files
    run_numbers = [int(f.name.split(".")[-1]) for f in self.working_dir.iterdir() if f.is_file() 
                 and re.match(fr".*{self.run_file.stem}.*\.[0-9][0-9][0-9]", str(f))]
    run_numbers += [0] # if the folder is empty -> start from 1 (which is added on the next line)
    return max(run_numbers) + 1

get_output_file_handler

get_output_file_handler(
    working_dir, output_file_name="output_file.xlsx"
)

Method to obtain the output file handler for writing to the run output file.

Parameters:

Name	Type	Description	Default
working_dir	str | Path	Path to the working dir.	required
output_file_name	str	Name of the output file to be written to. Defaults to "output_file.xlsx".	'output_file.xlsx'

Returns:

Type	Description
OutputFileHandler	output_file_handler.OutputFileHandler: Instance of the output file handler.

Source code in common/core.py

def get_output_file_handler(self, working_dir:str, output_file_name="output_file.xlsx") -> output_file_handler.OutputFileHandler:
    """Method to obtain the output file handler for writing to the run output file.

    Args:
        working_dir (str | pathlib.Path): Path to the working dir.
        output_file_name (str, optional): Name of the output file to be written to. Defaults to "output_file.xlsx".

    Returns:
        output_file_handler.OutputFileHandler: Instance of the output file handler.
    """
    # look if handler for this output file already exists
    for handler in Core._output_file_handlers:
        if handler.output_file == pathlib.Path(working_dir).joinpath(output_file_name):
            return handler
    # if not, create new
    new_handler = output_file_handler.OutputFileHandler(working_dir, output_file_name)
    Core._output_file_handlers += [new_handler]
    return new_handler

setup_logger

setup_logger(
    log_dir, log_file_name, console_log, logger_name=None
)

Function to setup internal logger

Parameters:

Name	Type	Description	Default
log_dir	str | None	Name of the log directory, that will be created inside the working directory. If set to None, log dir and logging file will not be used.	required
log_file_name	str	Name of log file, if None, there will be no log file.	required
console_log	bool	If False, the log is not written to the console	required

Source code in common/core.py

def setup_logger(self, log_dir:str, log_file_name:str, console_log:bool, logger_name=None) -> None:
    """Function to setup internal logger

    Args:
        log_dir (str | None): Name of the log directory, that will be created inside the working directory. If set to None, log dir and logging file will not be used.
        log_file_name (str): Name of log file, if None, there will be no log file.
        console_log (bool): If False, the log is not written to the console
    """
    # Get logger and basic setup
    self.logger = logging.getLogger(''.join(random.choice(string.ascii_lowercase) for i in range(7)) if logger_name is None else logger_name) # generate random name for the logger
    self.logger.setLevel(logging.INFO)
    formatter = logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')

    # Make sure the logging directory exists
    if log_dir is not None:
        self.log_dir = self.working_dir.joinpath(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)

        # Set up log file handler
        if log_file_name is not None:
            self.log_file = self.log_dir.joinpath(log_file_name)
            self.log_file.parent.mkdir(parents=True, exist_ok=True) # in case log file was to be in other subdirectory
            fh = logging.FileHandler(self.log_file)
            fh.setLevel(logging.DEBUG)
            fh.setFormatter(formatter)
            self.logger.addHandler(fh)

    # Console log is setup only if console_log is True
    if console_log:
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        ch.setFormatter(formatter)
        self.logger.addHandler(ch)

---

Last update: October 31, 2023
Created: October 31, 2023