SupplyNetPy Components.core Module

def __init__(self, node:object, periodic_update:bool=False, period:float=1):
    """
    Initialize the statistics object.

    Parameters:
        node (object): The node for which statistics are tracked.
        periodic_update (bool, optional): Whether to update statistics periodically. Default is False.
        period (float, optional): Time interval for periodic updates. Default is 1.

    Attributes:
        node (object): The node to which this statistics object belongs.
        name (str): Name of the statistics object.
        demand_placed (list): Orders this node placed, as [number of orders, units].
        fulfillment_received (list): Orders this node received (they arrived), as [number of orders, units].
        demand_received (list): Orders placed with this node (replenishment orders and customer orders), as [number of orders, units].
        demand_fulfilled (list): Orders this node fulfilled (sent out in full), as [number of orders, units].
        shortage (list): How many units this node was missing when orders arrived, as [number of orders, units]. A running total that is never reduced. ``orders_shortage`` is kept as an old name for it (§6.6).
        backorder (list): Orders this node owes right now, as [number of orders, units].
        disruption_lost_sales (list): Customer orders this site lost because it was disrupted when they arrived, as [number of orders, units]. Kept apart from ``shortage``, which is for units the site did not have.
        start_time (float): When counting began: the time these statistics were created, or the last ``reset``.
        elapsed_time (float): Time since ``start_time``. Worked out when read; every average below divides by it.
        lost_sales (list): Customer units this site never delivered, as [number of orders, units]: the whole order of a customer with no tolerance who found too few units, the rest of an order when the customer takes only what is there, an order too big for the site ever to hold, and what a waiting customer was still owed when their tolerance ran out. Unlike ``shortage``, it counts only units that were never delivered. Customers turned away by a disruption are in ``disruption_lost_sales`` instead.
        waited (list): Orders that had to wait at this site, from customers and from other sites, as [number of orders, units].
        waiting_time (float): Total time those orders waited, until their units left the inventory.
        avg_waiting_time (float): Average wait per order fulfilled, with orders served at once counting as 0: ``waiting_time / demand_fulfilled[0]``. Worked out when read.
        blocked (list): Replenishment orders this site could not place because every supplier it has (or the link to it) was disrupted, as [number of orders, units]. They are not lost: the site orders again when a supplier or a link recovers. A site with no supplier at all records nothing here; that is a mistake in how the network is built, and a warning says so.
        dropped_qty (float): Units of deliveries that did not fit in the inventory and that no waiting customer could take. The site does not pay for them.
        inventory_level_time (float): Running sum of inventory level × time (units × time units), up to now. Holding 10 units for 3 time units and then 4 units for 2 gives 38. The holding cost is this sum times the holding cost per unit per time unit.
        avg_inventory_level (float): Average inventory level over time, each level counted for as long as it lasted: ``inventory_level_time / elapsed_time`` (7.6 in the example above). Worked out when read.
        zero_inventory_time (float): Time the inventory level was 0, up to now.
        zero_inventory_share (float): Share of time the inventory level was 0: ``zero_inventory_time / elapsed_time``. Worked out when read.
        disruptions (int): Number of disruptions of this node.
        disrupted_time (float): Time this node spent disrupted, including a disruption still going on when it is read.
        availability (float): Share of time this node was not disrupted: ``1 - disrupted_time / elapsed_time``. Worked out when read.
        inventory_level (float): The inventory level at the last update.
        inventory_waste (float): Expired units thrown away so far.
        inventory_carry_cost (float): Holding cost so far.
        inventory_spend_cost (float): What this node paid for the units it received.
        transportation_cost (float): Transportation cost of this node's orders.
        node_cost (float): Everything this node spent (the sum of its costs).
        revenue (float): Sell price times the units this node fulfilled.
        profit (float): Revenue minus node cost.
        _info_keys (list): Keys to include in the info dictionary.
        _stats_keys (list): Keys to include in the statistics dictionary.

    Returns:
        None
    """
    # Clone the class-level declarative lists onto the instance so that
    # per-node extensions (e.g. ``Supplier`` appending "total_material_cost"
    # to its own ``self.stats._stats_keys``) do not leak into the shared
    # class attribute and pollute other Statistics instances.
    self._stats_keys = list(type(self)._stats_keys)
    self._cost_components = list(type(self)._cost_components)
    self.node = node # the node to which this statistics object belongs
    self.name = f"{self.node.ID} statistics"
    self.demand_placed = [0,0] # orders this node placed: [number of orders, units]
    self.fulfillment_received = [0,0] # orders that arrived at this node
    self.demand_received = [0,0] # orders placed with this node
    self.demand_fulfilled = [0,0] # orders this node sent out in full
    # ``shortage`` (was ``orders_shortage`` before §6.6) — renamed to
    # match ``supplychainnet["shortage"]`` so the same vocabulary applies
    # at both the node and network level. ``orders_shortage`` is kept as
    # a read-only alias / accepted ``update_stats`` kwarg for back-compat.
    self.shortage = [0,0] # units missing when orders arrived (running total)
    self.start_time = self.node.env.now # when counting began (moved by reset)
    self.lost_sales = [0,0] # customer units never delivered (other than through a disruption)
    self.disruption_lost_sales = [0,0] # customer orders lost because the site was disrupted when they arrived
    self.backorder = [0,0] # orders this node owes right now
    self.waited = [0,0] # orders that had to wait at this site
    self.waiting_time = 0 # total time those orders waited
    self.blocked = [0,0] # replenishment orders not placed because no supplier could take them
    self.dropped_qty = 0 # delivered units that did not fit in the inventory
    self.disruptions = 0 # number of disruptions of this node
    # Time spent disrupted, kept as the total of disruptions that have
    # ended plus the start of the one going on now (if any), so that
    # ``disrupted_time`` can include the ongoing part when it is read.
    self._disrupted_time = 0
    self._disrupted_since = None
    self.inventory_level = 0 # inventory level at the last update
    self.inventory_waste = 0 # expired units thrown away so far
    self.inventory_carry_cost = 0 # holding cost so far
    self.inventory_spend_cost = 0 # what this node paid for the units it received
    self.transportation_cost = 0 # transportation cost of this node's orders
    # Units destroyed in disruptions (for example, a flood at a
    # warehouse). ``destroyed_qty`` counts the units; ``destroyed_value``
    # is the cost of disposing of them, and is part of ``node_cost`` (see
    # _cost_components).
    self.destroyed_qty = 0
    self.destroyed_value = 0
    self.node_cost = 0 # everything this node spent
    self.revenue = 0 # sell price times the units this node fulfilled
    self.profit = 0 # revenue minus node cost

    if(periodic_update):
        self.node.env.process(self.update_stats_periodically(period=period))

def __init__(self, link):
    """
    Initialize the statistics object.

    Parameters:
        link (Link): The link whose statistics these are.

    Returns:
        None
    """
    self.link = link
    self.name = f"{link.ID} statistics"
    self._clear_totals()
    self.backorder = [0, 0] # this buyer's orders waiting at the supplier right now
    self.start_time = link.env.now # when counting began (moved by reset)
    # Time spent disrupted, kept as the total of disruptions that have
    # ended plus the start of the one going on now (if any), so that
    # ``disrupted_time`` can include the ongoing part when it is read.
    self._disrupted_since = None

def __init__(self,
             ID: str,
             name: str,
             extraction_quantity: float,
             extraction_time: float,
             mining_cost: float,
             cost: float) -> None:
    """
    Initialize the raw material object.

    Parameters:
        ID (str): ID of the raw material (alphanumeric)
        name (str): name of the raw material
        extraction_quantity (float): units mined in each round of mining
        extraction_time (float): time units each round of mining takes
        mining_cost (float): cost of mining one unit
        cost (float): the raw-material supplier's sell price per unit

    Attributes:
        _info_keys (list): list of keys to include in the info dictionary
        _stats_keys (list): list of keys to include in the statistics dictionary
        ID (str): ID of the raw material (alphanumeric)
        name (str): name of the raw material
        extraction_quantity (float): units mined in each round of mining
        extraction_time (float): time units each round of mining takes
        mining_cost (float): cost of mining one unit
        cost (float): the raw-material supplier's sell price per unit

    Returns:
        None
    """
    validate_positive("Extraction quantity", extraction_quantity)
    validate_non_negative("Extraction time", extraction_time)
    validate_non_negative("Mining Cost", mining_cost)
    validate_positive("Cost", cost)
    self.ID = ID # ID of the raw material (alphanumeric)
    self.name = name # name of the raw material
    self.extraction_quantity = extraction_quantity # units mined in each round of mining
    self.extraction_time = extraction_time # time units each round of mining takes
    self.mining_cost = mining_cost # cost of mining one unit
    self.cost = cost # the raw-material supplier's sell price per unit

def __init__(self,
             ID: str,
             name: str,
             manufacturing_cost: float,
             manufacturing_time: float,
             sell_price: float,
             raw_materials: list,
             batch_size: int,
             buy_price: float = 0) -> None:
    """
    Initialize the product object.

    Performs input validation for positive and non-negative values, and ensures raw materials are provided.

    Parameters:
        ID (str): ID of the product (alphanumeric)
        name (str): Name of the product
        manufacturing_cost (float): Manufacturing cost per unit
        manufacturing_time (float): Time units to make one production batch
        sell_price (float): Sell price per unit
        buy_price (float, optional): Buy price per unit (default is 0)
        raw_materials (list): List of (raw material, quantity) tuples: the raw materials needed to make one unit
        batch_size (int): The most units made in one production batch

    Attributes:
        _info_keys (list): List of keys to include in the info dictionary
        _stats_keys (list): List of keys to include in the statistics dictionary
        ID (str): ID of the product
        name (str): Name of the product
        manufacturing_cost (float): Manufacturing cost per unit
        manufacturing_time (float): Time units to make one production batch
        sell_price (float): Sell price per unit
        buy_price (float): Buy price per unit (default is 0)
        raw_materials (list): List of (raw material, quantity) tuples needed to make one unit
        batch_size (int): The most units made in one production batch

    Returns:
        None

    Raises:
        ValueError: If validations fail for positive values, non-negative values, or empty raw materials list.
    """
    validate_positive("Manufacturing cost", manufacturing_cost)
    validate_non_negative("Manufacturing time", manufacturing_time)
    validate_positive("Sell price", sell_price)
    validate_non_negative("Buy price", buy_price)
    validate_positive("Units per cycle", batch_size)
    if raw_materials is None or len(raw_materials) == 0:
        global_logger.error("Raw materials cannot be empty.")
        raise ValueError("Raw materials cannot be empty.")
    for raw_mat in raw_materials:
        if not isinstance(raw_mat[0], RawMaterial):
            raise ValueError("Invalid raw material.")
        if raw_mat[1] <= 0:
            raise ValueError("Invalid quantity for raw material.")

    self.ID = ID # ID of the product (alphanumeric)
    self.name = name # name of the product
    self.manufacturing_cost = manufacturing_cost # manufacturing cost of the product (per unit)
    self.manufacturing_time = manufacturing_time # time units to make one production batch
    self.sell_price = sell_price # sell price per unit
    self.buy_price = buy_price # buy price per unit (default: 0)
    self.raw_materials = raw_materials # (raw material, quantity) tuples needed to make one unit
    self.batch_size = batch_size # the most units made in one production batch

def __init__(self,
             env: simpy.Environment,
             node: object,
             params: dict) -> None:
    """
    Initialize the replenishment policy object.

    Parameters:
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy

    Attributes:
        _info_keys (list): list of keys to include in the info dictionary
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy

    Returns:
        None
    """
    if not isinstance(env, simpy.Environment):
        raise ValueError("Invalid environment. Provide a valid SimPy environment.")
    self.env = env  # simulation environment
    self.node = node  # node to which this policy applies
    self.params = params  # parameters for the replenishment policy
    self._warn_unused_params()

def __init__(self, env, node, params):
    """ 
    Initialize the replenishment policy object.

    Parameters:
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy (s, S)

    Attributes:
        _info_keys (list): list of keys to include in the info dictionary
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy (s, S)
        name (str): replenishment policy name
        first_review_delay (float): time units to wait before the first check of the inventory position
        period (float): time units between checks; 0 means checking each time the inventory drops

    Returns:
        None
    """
    validate_non_negative("Reorder point (s)", params['s']) # s must not be negative
    validate_positive("Order-up-to level (S)", params['S']) # S must be positive
    if 's' not in params or 'S' not in params:
        raise ValueError("Parameters 's' and 'S' must be provided for the (s, S) replenishment policy.")
    if params['s'] > params['S']:
        raise ValueError("Reorder point (s) must be less than or equal to order-up-to level (S).")
    super().__init__(env, node, params)
    self.name = "min-max replenishment (s, S)"
    self.first_review_delay = params.get('first_review_delay', 0)
    self.period = params.get('period',0)

def __init__(self, env, node, params):
    """ 
    Initialize the (R, Q) replenishment policy object.

    Parameters:
        env (simpy.Environment): Simulation environment.
        node (object): Node to which this policy applies.
        params (dict): Replenishment policy parameters: R, Q, and the optional first_review_delay and period.

    Attributes:
        _info_keys (list): List of keys to include in the info dictionary.
        env (simpy.Environment): Simulation environment.
        node (object): Node to which this policy applies.
        params (dict): Replenishment policy parameters (R, Q, optional delays and period).
        name (str): Replenishment policy name.
        first_review_delay (float): Time units to wait before the first check of the inventory position.
        period (float): Time units between checks. 0 means checking each time the inventory drops.

    Returns:
        None
    """
    validate_non_negative("Reorder point (R)", params['R']) # R must not be negative
    validate_positive("Order quantity (Q)", params['Q'])  # Q must be positive
    super().__init__(env, node, params)
    self.name = "RQ replenishment (R, Q)"
    self.first_review_delay = params.get('first_review_delay', 0)
    self.period = params.get('period', 0)

def __init__(self, env, node, params):
    """ 
    Initialize the replenishment policy object.

    Parameters:
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy (T, Q), and optional parameters (safety_stock, first_review_delay).

    Attributes:
        _info_keys (list): list of keys to include in the info dictionary
        env (simpy.Environment): simulation environment
        node (object): node to which this policy applies
        params (dict): parameters for the replenishment policy (T, Q)
        name (str): replenishment policy name
        first_review_delay (float): time units to wait before the first replenishment order

    Returns:
        None
    """
    validate_positive("Replenishment period (T)", params['T'])  # T must be positive
    validate_positive("Replenishment quantity (Q)", params['Q'])  # Q must be positive
    super().__init__(env, node, params)
    self.name = "Periodic replenishment (T, Q)"
    self.first_review_delay = params.get('first_review_delay', 0)

def __init__(self, node, mode="fixed", params=None):
    """
    Initialize the supplier selection policy object.

    Parameters:
        node (object): Node for which the supplier selection policy is applied.
        mode (str): Supplier selection mode ("dynamic" or "fixed").
        params (dict): Settings for the policy (``supplier_selection_param`` of the site), or None.

    Attributes:
        _info_keys (list): List of keys to include in the info dictionary.
        node (object): Node for which the supplier selection policy is applied.
        mode (str): Supplier selection mode ("dynamic" or "fixed").
        params (dict): The settings given to the policy, or None.
        fixed_supplier (Link): In "fixed" mode, the link to the remembered supplier (None until the first choice).

    Returns:
        None

    Raises:
        ValueError: If the mode is not "dynamic" or "fixed".
        TypeError: If the node is not an instance of Node class.
    """
    if mode not in ["dynamic", "fixed"]:
        global_logger.error(f"Invalid mode: {mode}. Mode must be either 'dynamic' or 'fixed'.")
        raise ValueError("Mode must be either 'dynamic' or 'fixed'.")
    if not isinstance(node, Node):
        global_logger.error("Node must be an instance of Node class.")
        raise TypeError("Node must be an instance of Node class.")
    self.node = node
    self.mode = mode
    self.params = params
    self.fixed_supplier = None
    self._warn_unused_params()