simulation_handler.helpers

  1""" 
  2Helper functions for the simulation environment.
  3This module contains various utility functions for logging, data cleaning,
  4and random number generation used in the simulation.
  5"""
  6
  7import os
  8import shutil
  9import re
 10import random
 11import os
 12import glob
 13
 14import numpy as np
 15from scipy.stats import truncnorm
 16import pandas as pd
 17
 18import constants
 19
 20#######################
 21# Logs
 22#######################
 23
 24
 25def log_line(run_id, filename, line):
 26    """
 27    Logs a line to a specified file in the run's logs directory.
 28    Args:
 29        run_id (str): Unique identifier for the run to save the log.
 30        filename (str): Name of the file to log the line.
 31        line (str): The line to log.
 32    Returns:
 33        None: Appends the line to the specified log file.
 34    """
 35    try:
 36        with open(f'.{run_id}/logs/{filename}', 'a') as f:
 37            f.write(line + '\n')
 38    except FileNotFoundError:
 39        with open(f'.{run_id}/logs/{filename}', 'w') as f:
 40            f.write(line + '\n')
 41
 42
 43def save_warning(run_id, message, print_val=True):
 44    """
 45    Save a warning message to the logs directory for the current run.
 46    If the logs directory does not exist, it will be created.
 47    Args:
 48        run_id (str): The identifier for the current run.
 49        message (str): The warning message to be saved.
 50        print_val (bool): Whether to print the message to the console. Default is True.
 51    Returns:
 52        None
 53    """
 54    if not os.path.exists(f'.{run_id}/logs/warnings.txt'):
 55        with open(f'.{run_id}/logs/warnings.txt', 'w') as f:
 56            f.write('Warnings:\n')
 57        with open(f'.{run_id}/logs/warnings.txt', 'a') as f:
 58            f.write(f'{message}\n')
 59    else:
 60        with open(f'.{run_id}/logs/warnings.txt', 'a') as f:
 61            f.write(f'{message}\n')
 62    if print_val:
 63        print(f"{message}")
 64
 65
 66def clear_env(env, ship_proc, truck_proc, train_proc, data_taker_proc):
 67    """
 68    Clear the environment and interrupt all processes.
 69    Args:
 70        env (Simpy.Environment): The simulation environment to be cleared.
 71        ship_proc (Simpy.Process): The process for handling ships.
 72        truck_proc (Simpy.Process): The process for handling trucks.
 73        train_proc (Simpy.Process): The process for handling trains.
 74        data_taker_proc: The process for taking data.
 75    Returns:
 76        None
 77    """
 78    try:
 79        ship_proc.interrupt()
 80    except:
 81        pass
 82    try:
 83        truck_proc.interrupt()
 84    except:
 85        pass
 86    try:
 87        train_proc.interrupt()
 88    except:
 89        pass
 90
 91    try:
 92        data_taker_proc.interrupt()
 93    except:
 94        pass
 95
 96    env._queue.clear()
 97
 98
 99def clear_logs(run_id):
100    """
101    Clear the logs directory for the current run by removing all files and subdirectories.
102    Then create the necessary subdirectories again.
103    This function is useful for resetting the logs directory before starting a new simulation run.
104    Args:
105        run_id (str): The identifier for the current run.
106    Returns:
107        None
108    """
109
110    if constants.DELETE_RESULTS_FOLDER == False:
111        shutil.rmtree(f'.{run_id}/logs', ignore_errors=True)
112        shutil.rmtree(f'.{run_id}/plots', ignore_errors=True)
113        # shutil.rmtree(f'.{run_id}/animations', ignore_errors=True)
114        shutil.rmtree(f'.{run_id}/bottlenecks', ignore_errors=True)
115
116    # os.makedirs(f'.{run_id}/animations')
117
118    os.makedirs(f'.{run_id}/logs')
119    os.makedirs(f'.{run_id}/logs/availability')
120    os.makedirs(f'.{run_id}/plots')
121    os.makedirs(f'.{run_id}/bottlenecks')
122    os.makedirs(f'.{run_id}/bottlenecks/chassisBays/')
123    os.makedirs(f'.{run_id}/plots/TerminalProcessCharts/')
124    os.makedirs(f'.{run_id}/plots/TerminalProcessDist/')
125    os.makedirs(f'.{run_id}/plots/DwellTimes/')
126    os.makedirs(f'.{run_id}/plots/TurnTimes/')
127    os.makedirs(f'.{run_id}/plots/DwellTimesDist/')
128    os.makedirs(f'.{run_id}/plots/TurnTimesDist/')
129    os.makedirs(f'.{run_id}/plots/scenarios')
130    os.makedirs(f'.{run_id}/plots/shipDistribution')
131
132    os.makedirs(f'.{run_id}/plots/truckDistribution')
133    os.makedirs(f'.{run_id}/plots/TruckDwellByCargo')
134    os.makedirs(f'.{run_id}/plots/TruckDwellByTerminal')
135    os.makedirs(f'.{run_id}/plots/TruckArrivalByCargo')
136    os.makedirs(f'.{run_id}/plots/TruckArrivalByTerminal')
137
138    # Create a force_action.txt file under logs
139    with open(f'.{run_id}/logs/force_action.txt', 'w') as f:
140        f.write('Force Actions')
141        f.write('\n')
142
143
144def clean_results_directory():
145    """
146    Remove existing result files and directories matching the pattern `.Results*`.
147    This function is useful for cleaning up the results directory before starting a new simulation run.
148    Args:
149        None
150    Returns:
151        None
152"""
153    for file_path in glob.glob(".Results*"):
154        try:
155            if os.path.isfile(file_path):
156                os.remove(file_path)
157            elif os.path.isdir(file_path):
158                for root, dirs, files in os.walk(file_path, topdown=False):
159                    for name in files:
160                        os.remove(os.path.join(root, name))
161                    for name in dirs:
162                        os.rmdir(os.path.join(root, name))
163                os.rmdir(file_path)
164        except OSError as e:
165            print(f"Error removing {file_path}: {e}")
166
167#######################
168# Data cleaning
169#######################
170
171
172def initialize_rng(seed):
173    """
174    Initialize random state objects for repeatable randomness across runs.
175    Args:
176        seed (int): The seed value for the random number generator.
177    Returns:
178        None
179    """
180    global rng_random, rng_numpy
181    rng_random = random.Random(seed)
182    rng_numpy = np.random.default_rng(seed)
183
184
185def clean_data(directory):
186    """
187    Clean the terminal data by parsing ranges and converting them to tuples.
188    This function reads a CSV file containing terminal data, processes the range strings,
189    and converts them into tuples of integers or floats.
190    Args:
191        directory (str): The directory where the terminal data CSV file is located.
192    Returns:
193        pd.DataFrame: A DataFrame containing the cleaned terminal data with ranges parsed into tuples.
194    """
195
196    def parse_range(range_str):
197        """
198        Parse a range in the format '(a,b)' from a string into a tuple of integers or floats.
199        If the input is a single number, it will be returned as an integer.
200        Args:
201            range_str (str): The range string to be parsed.
202        Returns:
203            tuple or int: A tuple of integers if the input is a range, or an integer if it's a single number.
204        """
205        # convert range string to a string
206        range_str = str(range_str)
207        try:
208            match = re.match(r"\(([\d.]+)\s*,\s*([\d.]+)\)", range_str)
209            start, end = int(match.group(1)), int(match.group(2))
210            return (start, end)
211        except:
212            return int(range_str)
213
214    csv_data = pd.read_csv(f'{directory}/inputs/terminal_data.csv')
215    for column in csv_data.columns[3:]:
216        csv_data[column] = csv_data[column].apply(
217            lambda x: parse_range(x) if isinstance(x, str) else x)
218    return csv_data
219
220
221def create_terminal_data_cache(terminal_data, run_id, seed):
222    """
223    Create a cache dictionary from terminal data for quick access to resource values.
224    This function initializes a random number generator with a given seed,
225    iterates through the terminal data, and populates a cache dictionary with keys
226    based on cargo type, terminal ID, and resource name. If the resource value is a range,
227    it stores a random integer from that range; otherwise, it stores the value directly.
228    Args:
229        terminal_data (pd.DataFrame): The DataFrame containing terminal data.
230        run_id (str): The identifier for the current run, used for logging.
231        seed (int): The seed value for the random number generator.
232    Returns:
233        dict: A dictionary where keys are tuples of (cargo_type, terminal_id, resource_name)
234              and values are the corresponding resource values or random integers from ranges.
235    """
236    # Create the cache dictionary
237    initialize_rng(seed)
238    cache = {}
239    # Populate the cache
240    for index, row in terminal_data.iterrows():
241        cargo_type = row['Cargo']
242        terminal_id = row['Terminal']
243        # Skip 'Cargo' and 'Terminal' columns
244        for resource_name in row.index[2:]:
245            value = row[resource_name]
246            key = (cargo_type, terminal_id, resource_name)
247            # If value is a tuple-like string, convert it to tuple and store a random range value; otherwise, store as int
248            if isinstance(value, tuple):
249                # Convert to tuple and store a random value from range
250                cache[key] = rng_random.randint(value[0], value[1])
251            else:
252                if resource_name in ['train arrival rate', 'truck arrival rate']:
253                    cache[key] = value
254                # Store integer or directly as it is
255                else:
256                    cache[key] = int(value)
257
258    return cache
259
260
261def get_value_by_terminal(terminal_data_cache, cargo_type, terminal_id, resource_name):
262    """
263    Retrieve a value from the terminal data cache based on cargo type, terminal ID, and resource name.
264    This function checks the cache for a specific key and returns the corresponding value.
265    Args:
266        terminal_data_cache (dict): The cache dictionary containing terminal data.
267        cargo_type (str): The type of cargo.
268        terminal_id (str): The ID of the terminal.
269        resource_name (str): The name of the resource.
270    Returns:        
271        The value associated with the specified cargo type, terminal ID, and resource name,
272        or None if the key does not exist in the cache.
273    """
274    key = (cargo_type, terminal_id, resource_name)
275    return terminal_data_cache.get(key)
276
277
278def create_terminal_tuple_cache(terminal_data, run_id, seed):
279    """
280    Create a cache dictionary from terminal data for quick access to resource values as tuples.
281    This function initializes a random number generator with a given seed,
282    iterates through the terminal data, and populates a cache dictionary with keys
283    based on cargo type, terminal ID, and resource name. If the resource value is a range,
284    it stores the range as a tuple; otherwise, it stores the value as a tuple of itself.
285    Args:
286        terminal_data (pd.DataFrame): The DataFrame containing terminal data.   
287        run_id (str): The identifier for the current run, used for logging.
288        seed (int): The seed value for the random number generator.
289    Returns:
290        dict: A dictionary where keys are tuples of (cargo_type, terminal_id, resource_name)
291              and values are tuples of the corresponding resource values or ranges.
292    """
293    # Create the cache dictionary
294    initialize_rng(seed)
295    truck_load_cache = {}
296
297    # Populate the cache
298    for index, row in terminal_data.iterrows():
299        cargo_type = row['Cargo']
300        terminal_id = row['Terminal']
301        # Skip 'Cargo' and 'Terminal' columns
302        for resource_name in row.index[2:]:
303            value = row[resource_name]
304            key = (cargo_type, terminal_id, resource_name)
305            # If value is a tuple-like string, convert it to tuple and store a random range value; otherwise, store as int
306            if isinstance(value, tuple):
307                # Convert to tuple and store a random value from range
308                truck_load_cache[key] = (value[0], value[1])
309            else:
310                if resource_name in ['train arrival rate', 'truck arrival rate']:
311                    truck_load_cache[key] = (value, value)
312                # Store integer or directly as it is
313                else:
314                    truck_load_cache[key] = (int(value), int(value))
315
316    return truck_load_cache
317
318
319def get_value_from_terminal_tuple(terminal_tuple_cache, cargo_type, terminal_id, resource_name):
320    """
321    Retrieve a value from the terminal tuple cache based on cargo type, terminal ID, and resource name.
322    This function checks the cache for a specific key and returns a random integer from the range stored in the tuple.
323    Args:
324        terminal_tuple_cache (dict): The cache dictionary containing terminal data as tuples.
325        cargo_type (str): The type of cargo.        
326        terminal_id (str): The ID of the terminal.
327        resource_name (str): The name of the resource.
328    Returns:
329        int: A random integer from the range stored in the tuple for the specified cargo type, terminal ID, and resource name.
330        If the key does not exist in the cache, it returns None.
331    """
332    # initialize_rng(seed)
333    key = (cargo_type, terminal_id, resource_name)
334    tups = terminal_tuple_cache.get(key)
335    return rng_random.randint(tups[0], tups[1])
336
337
338def get_values_by_terminal_random_sample(terminal_data, cargo_type, terminal_id, resource_name_input, num_samples, seed):
339    """
340    Get a list of random samples from the terminal data based on cargo type and terminal ID.
341    This function retrieves a specific resource value from the terminal data for a given cargo type and terminal ID,
342    and generates a list of random samples from that value. If the value is a range, it samples from that range;
343    otherwise, it returns a list with the value repeated for the specified number of samples.  
344    Args:
345        terminal_data (pd.DataFrame): The DataFrame containing terminal data.
346        cargo_type (str): The type of cargo.
347        terminal_id (str): The ID of the terminal.
348        resource_name_input (str): The name of the resource to sample from.
349        num_samples (int): The number of random samples to generate.
350        seed (int): The seed value for the random number generator.
351    Returns:
352        list: A list of random samples from the specified resource value.
353        If the resource value is a range, it returns random integers from that range;
354        otherwise, it returns a list with the value repeated for the specified number of samples.
355    """
356    random.seed(seed)
357
358    row = terminal_data[(terminal_data['Terminal'] == terminal_id) &
359                        (terminal_data['Cargo'] == cargo_type)].iloc[0]
360
361    value = row[resource_name_input]
362
363    try:
364        return [random.randint(int(value[0]), int(value[1])) for _ in range(num_samples)]
365    except:
366        return [value] * num_samples
367
368
369#######################
370# Helper functions
371#######################
372
373def is_daytime(time, start=6, end=18):
374    """
375    Check if the given time is within the daytime range.
376    Args:
377        time (int or float): The time to check, in hours (0-23).
378        start (int): The start of the daytime range (default is 6).
379        end (int): The end of the daytime range (default is 18).
380    Returns:
381        bool: True if the time is within the daytime range, False otherwise.
382    """
383    return start <= time % 24 <= end
384
385#######################
386# Random
387#######################
388
389
390def normal_random_with_sd(mu, sigma, seed, scale_factor=1):
391    """
392    Generate a random value from a truncated normal distribution with specified mean and standard deviation.
393    The distribution is truncated to ensure the value is within a specified range based on the mean and standard deviation.
394    Args:
395        mu (float): The mean of the normal distribution.
396        sigma (float): The standard deviation of the normal distribution.
397        seed (int): Random seed for reproducibility.
398        scale_factor (float): Factor to scale the standard deviation for truncation (default is 1).
399    Returns:
400        float: A random value from the truncated normal distribution, ensuring it is within the range [a, b].
401    """
402    np.random.seed(seed)
403    a = max(0, mu - scale_factor * sigma)
404    b = max(0, mu + scale_factor * sigma)
405    if a <= 0:
406        a = 0
407    lower, upper = (a - mu) / sigma, (b - mu) / sigma
408    distribution = truncnorm(lower, upper, loc=mu, scale=sigma)
409    val = distribution.rvs()
410    if val < a:
411        return a
412    elif val > b:
413        return b
414    return distribution.rvs()
415
416
417def normal_random_with_limit(a, b, seed):
418    """
419    Generate a random value from a truncated normal distribution between a and b.
420    Args:
421        a (float): Lower limit of the range.
422        b (float): Upper limit of the range.
423        seed (int): Random seed for reproducibility.
424    Returns:
425        float: A random value from the truncated normal distribution.
426    """
427    np.random.seed(seed)
428    mean = (a + b) / 2
429    std_dev = (b - a) / 6
430    lower_bound = (a - mean) / std_dev
431    upper_bound = (b - mean) / std_dev
432    distribution = truncnorm(lower_bound, upper_bound, loc=mean, scale=std_dev)
433    return distribution.rvs()