The utility module offers various functions to create, simulate, access information, and visualize the supply chain network more effectively.
SupplyNetPy.Components.utilities
create_sc_net
create_sc_net(nodes: list, links: list, demands: list, env: Environment = None)This functions inputs the nodes, links and demand netlists and creates supply chain nodes, links and demand objects. It then creates a supply chain network by putting all the objects in a dictionary.
| Parameters: | 
 | 
|---|
| Attributes: | 
 | 
|---|
| Raises: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def create_sc_net(nodes: list, links: list, demands: list, env:simpy.Environment = None):
    """
    This functions inputs the nodes, links and demand netlists and creates supply chain nodes, links and demand objects. 
    It then creates a supply chain network by putting all the objects in a dictionary.
    Parameters:
        nodes (list): A netlist of nodes in the supply chain network.
        links (list): A netlist of links between the nodes.
        demand (list): A netlist of demand nodes in the supply chain network.
        env (simpy.Environment, optional): A SimPy Environment object. If not provided, a new environment will be created.
    Attributes:
        global_logger (GlobalLogger): The global logger instance used for logging messages.
        supplychainnet (dict): A dictionary representing the supply chain network.
        used_ids (list): A list to keep track of used IDs to avoid duplicates.
        num_suppliers (int): Counter for the number of suppliers.
        num_manufacturers (int): Counter for the number of manufacturers.
        num_distributors (int): Counter for the number of distributors.
        num_retailers (int): Counter for the number of retailers.
    Raises:
        ValueError: If the SimPy Environment object is not provided or if there are duplicate IDs in nodes, links, or demands.
        ValueError: If an invalid node type is encountered.
        ValueError: If an invalid source or sink node is specified in a link.
        ValueError: If an invalid demand node is specified in a demand.
    Returns:
        dict: A dictionary representing the supply chain network.
    """
    if (isinstance(nodes[0],Node) or isinstance(links[0],Link) or isinstance(demands[0],Demand)) and env is None:
        global_logger.logger.error("Please provide SimPy Environment object env")
        raise ValueError("A SimPy Environment object is required!")
    if len(nodes)==0 or len(links)==0 or len(demands)==0:
        global_logger.logger.error("Nodes, links, and demands cannot be empty")
        raise ValueError("Nodes, links, and demands cannot be empty")
    if(env is None):
        env = simpy.Environment()
    supplychainnet = {"nodes":{},"links":{},"demands":{}} # create empty supply chain network
    used_ids = []
    num_suppliers = 0
    num_manufacturers = 0
    num_distributors = 0
    num_retailers = 0
    for node in nodes:
        if isinstance(node, dict):
            check_duplicate_id(used_ids, node["ID"], "node ID")
            node_id = node['ID']
            if node["node_type"].lower() in ["supplier", "infinite_supplier"]:
                supplychainnet["nodes"][f"{node_id}"] = Supplier(env=env, **node)
                num_suppliers += 1
            elif node["node_type"].lower() in ["manufacturer", "factory"]:
                node_ex = {key: node[key] for key in node if key != 'node_type'} # excluding key 'node_type', Manufacturer do not take it
                supplychainnet["nodes"][f"{node_id}"] = Manufacturer(env=env, **node_ex)
                num_manufacturers += 1
            elif node["node_type"].lower() in ["distributor", "warehouse"]:
                supplychainnet["nodes"][f"{node_id}"] = InventoryNode(env=env, **node)
                num_distributors += 1
            elif node["node_type"].lower() in ["retailer", "store", "shop"]:
                supplychainnet["nodes"][f"{node_id}"] = InventoryNode(env=env, **node)
                num_retailers += 1
            else:
                used_ids.remove(node["ID"])
                global_logger.logger.error(f"Invalid node type {node['node_type']}")
                raise ValueError("Invalid node type")
        elif isinstance(node, Node):
            if(node.ID in used_ids):
                global_logger.logger.error(f"Duplicate node ID {node.ID}")
                raise ValueError("Duplicate node ID")
            used_ids.append(node.ID)
            node_id = node.ID
            supplychainnet["nodes"][f"{node_id}"] = node
            if node.node_type.lower() in ["supplier", "infinite_supplier"]:
                num_suppliers += 1
            elif node.node_type.lower() in ["manufacturer", "factory"]:
                num_manufacturers += 1
            elif node.node_type.lower() in ["distributor", "warehouse"]:
                num_distributors += 1
            elif node.node_type.lower() in ["retailer", "store", "shop"]:
                num_retailers += 1
            else:
                used_ids.remove(node.ID)
                global_logger.logger.error(f"Invalid node type {node.node_type}")
                raise ValueError("Invalid node type")
    for link in links:
        if isinstance(link, dict):
            check_duplicate_id(used_ids, link["ID"], "link ID")
            source = None
            sink = None
            nodes = supplychainnet["nodes"].keys()
            if(link["source"] in nodes):
                source_id = link["source"]
                source = supplychainnet["nodes"][f"{source_id}"]
            if(link["sink"] in nodes):
                sink_id = link["sink"]
                sink = supplychainnet["nodes"][f"{sink_id}"]
            if(source is None or sink is None):
                global_logger.logger.error(f"Invalid source or sink node {link['source']} {link['sink']}")
                raise ValueError("Invalid source or sink node")
            exclude_keys = {'source', 'sink'}
            params = {k: v for k, v in link.items() if k not in exclude_keys}
            link_id = params['ID']
            supplychainnet["links"][f"{link_id}"] = Link(env=env,source=source,sink=sink,**params)
        elif isinstance(link, Link):
            if(link.ID in used_ids):
                global_logger.logger.error(f"Duplicate link ID {link.ID}")
                raise ValueError("Duplicate node ID")
            used_ids.append(link.ID)
            supplychainnet["links"][f"{link.ID}"] = link
    for d in demands:
        if isinstance(d, dict):
            check_duplicate_id(used_ids, d["ID"], "demand ID")
            demand_node = None # check for which node the demand is
            nodes = supplychainnet["nodes"].keys()
            if d['demand_node'] in nodes:
                demand_node_id = d['demand_node']
                demand_node = supplychainnet["nodes"][f"{demand_node_id}"]
            if(demand_node is None):
                global_logger.logger.error(f"Invalid demand node {d['demand_node']}")
                raise ValueError("Invalid demand node")
            exclude_keys = {'demand_node','node_type'}
            params = {k: v for k, v in d.items() if k not in exclude_keys}
            demand_id = params['ID']
            supplychainnet["demands"][f"{demand_id}"] = Demand(env=env,demand_node=demand_node,**params)
        elif isinstance(d, Demand):
            if(d.ID in used_ids):
                global_logger.logger.error(f"Duplicate demand ID {d.ID}")
                raise ValueError("Duplicate demand ID")
            used_ids.append(d.ID)
            supplychainnet["demands"][f"{d.ID}"] = d
    supplychainnet["env"] = env
    supplychainnet["num_of_nodes"] = num_suppliers + num_manufacturers + num_distributors + num_retailers
    supplychainnet["num_of_links"] = len(links)
    supplychainnet["num_suppliers"] = num_suppliers
    supplychainnet["num_manufacturers"] = num_manufacturers
    supplychainnet["num_distributors"] = num_distributors
    supplychainnet["num_retailers"] = num_retailers
    return supplychainnetcheck_duplicate_id
check_duplicate_id(used_ids, new_id, entity_type='ID')Checks if the new_id is already in used_ids. If it is, logs an error and raises a ValueError.
| Parameters: | 
 | 
|---|
| Returns: | 
 | 
|---|
| Raises: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def check_duplicate_id(used_ids, new_id, entity_type="ID"):
    """
    Checks if the new_id is already in used_ids. If it is, logs an error and raises a ValueError.
    Parameters:
        used_ids (list): List of already used IDs.
        new_id (str): The new ID to check.
        entity_type (str): Type of the entity for which the ID is being checked (e.g., "node ID", "link ID").
    Attributes:
        None
    Returns:
        None
    Raises:
        ValueError: If the new_id is already in used_ids.
    """
    if new_id in used_ids:
        global_logger.logger.error(f"Duplicate {entity_type} {new_id}")
        raise ValueError(f"Duplicate {entity_type}")
    used_ids.append(new_id)simulate_sc_net
simulate_sc_net(supplychainnet, sim_time, logging=True)Simulate the supply chain network for a given time period, and calculate performance measures.
| Parameters: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def simulate_sc_net(supplychainnet, sim_time, logging=True):
    """
    Simulate the supply chain network for a given time period, and calculate performance measures.
    Parameters:
        supplychainnet (dict): A supply chain network.
        sim_time (int): Simulation time.
    Returns:
        supplychainnet (dict): Updated dict with listed performance measures.
    """
    logger = global_logger.logger
    env = supplychainnet["env"]
    if(sim_time<=env.now):
        logger.warning(f"You have already ran simulation for this network! \n To create a new network use create_sc_net(), or specify the simulation time grater than {env.now} to run it further.")
        logger.info(f"Performance measures for the supply chain network are calculated and returned.")
    elif isinstance(logging, tuple) and len(logging) == 2:
        assert logging[0] < logging[1], "Start logging time should be less than stop logging time"
        assert logging[0] >= 0, "Start logging time should be greater than or equal to 0"
        assert logging[1] <= sim_time, "Stop logging time should be less than or equal to simulation time"        
        log_start = logging[0]
        log_stop = logging[1]
        global_logger.disable_logging()
        env.run(log_start) # Run the simulation
        global_logger.enable_logging()
        env.run(log_stop) # Run the simulation
        global_logger.disable_logging()
        if(sim_time > log_stop):
            env.run(sim_time) # Run the simulation
    elif isinstance(logging, bool) and logging:
        global_logger.enable_logging()
        env.run(sim_time) # Run the simulation
    else:
        global_logger.disable_logging()
        env.run(sim_time) # Run the simulation
    # Let's create some variables to store stats
    total_available_inv = 0
    avg_available_inv = 0
    total_inv_carry_cost = 0
    total_inv_spend = 0
    total_inv_waste = 0
    total_transport_cost = 0
    total_revenue = 0
    total_cost = 0
    total_profit = 0
    total_demand_by_customers = [0, 0] # [orders, products]
    total_fulfillment_received_by_customers = [0, 0] # [orders, products]
    total_demand_by_site = [0, 0] # [orders, products]
    total_fulfillment_received_by_site = [0, 0] # [orders, products]
    total_demand_placed = [0, 0] # [orders, products]
    total_fulfillment_received = [0, 0] # [orders, products]
    total_shortage = [0, 0] # [orders, products]
    total_backorders = [0, 0] # [orders, products]
    for key, node in supplychainnet["nodes"].items():
        if("infinite" in node.node_type.lower()): # skip infinite suppliers
            continue
        node.stats.update_stats() # update stats for the node
        total_available_inv += node.inventory.inventory.level
        if len(node.inventory.instantaneous_levels)>0:
            avg_available_inv += sum([x[1] for x in node.inventory.instantaneous_levels])/len(node.inventory.instantaneous_levels) 
        total_inv_carry_cost += node.inventory.carry_cost
        total_inv_spend += node.stats.inventory_spend_cost
        total_inv_waste += node.stats.inventory_waste
        total_transport_cost += node.stats.transportation_cost
        total_cost += node.stats.node_cost
        total_revenue += node.stats.revenue
        total_demand_by_site[0] += node.stats.demand_placed[0]
        total_demand_by_site[1] += node.stats.demand_placed[1]
        total_fulfillment_received_by_site[0] += node.stats.fulfillment_received[0]
        total_fulfillment_received_by_site[1] += node.stats.fulfillment_received[1]
        total_shortage[0] += node.stats.orders_shortage[0]
        total_shortage[1] += node.stats.orders_shortage[1]
        total_backorders[0] += node.stats.backorder[0]
        total_backorders[1] += node.stats.backorder[1]
    for key, node in supplychainnet["demands"].items():
        node.stats.update_stats() # update stats for the node
        total_transport_cost += node.stats.transportation_cost
        total_cost += node.stats.node_cost
        total_revenue += node.stats.revenue
        total_demand_by_customers[0] += node.stats.demand_placed[0] # orders
        total_demand_by_customers[1] += node.stats.demand_placed[1] # products
        total_fulfillment_received_by_customers[0] += node.stats.fulfillment_received[0]
        total_fulfillment_received_by_customers[1] += node.stats.fulfillment_received[1]
        total_shortage[0] += node.stats.orders_shortage[0]
        total_shortage[1] += node.stats.orders_shortage[1]
        total_backorders[0] += node.stats.backorder[0]
        total_backorders[1] += node.stats.backorder[1]
    total_demand_placed[0] = total_demand_by_customers[0] + total_demand_by_site[0]
    total_demand_placed[1] = total_demand_by_customers[1] + total_demand_by_site[1]
    total_fulfillment_received[0] = total_fulfillment_received_by_customers[0] + total_fulfillment_received_by_site[0]
    total_fulfillment_received[1] = total_fulfillment_received_by_customers[1] + total_fulfillment_received_by_site[1]
    total_profit = total_revenue - total_cost
    supplychainnet["available_inv"] = total_available_inv
    supplychainnet["avg_available_inv"] = avg_available_inv
    supplychainnet["inventory_carry_cost"] = total_inv_carry_cost   
    supplychainnet["inventory_spend_cost"] = total_inv_spend
    supplychainnet["inventory_waste"] = total_inv_waste
    supplychainnet["transportation_cost"] = total_transport_cost
    supplychainnet["revenue"] = total_revenue
    supplychainnet["total_cost"] = total_cost
    supplychainnet["profit"] = total_profit
    supplychainnet["demand_by_customers"] = total_demand_by_customers
    supplychainnet["fulfillment_received_by_customers"] = total_fulfillment_received_by_customers
    supplychainnet["demand_by_site"] = total_demand_by_site
    supplychainnet["fulfillment_received_by_site"] = total_fulfillment_received_by_site
    supplychainnet["total_demand"] = total_demand_placed
    supplychainnet["total_fulfillment_received"] = total_fulfillment_received
    supplychainnet["shortage"] = total_shortage
    supplychainnet["backorders"] = total_backorders
    # Calculate average cost per order and per item
    if total_demand_placed[0] > 0:
        supplychainnet["avg_cost_per_order"] = total_cost / total_demand_placed[0]
    else:
        supplychainnet["avg_cost_per_order"] = 0
    if total_demand_placed[1] > 0:
        supplychainnet["avg_cost_per_item"] = total_cost / total_demand_placed[1]
    else:
        supplychainnet["avg_cost_per_item"] = 0
    if isinstance(logging, tuple):
        global_logger.enable_logging()
    max_key_length = max(len(key) for key in supplychainnet.keys()) + 1
    logger.info(f"Supply chain info:")
    for key in sorted(supplychainnet.keys()):
        logger.info(f"{key.ljust(max_key_length)}: {supplychainnet[key]}")
    return supplychainnetvisualize_sc_net
visualize_sc_net(supplychainnet)Visualize the supply chain network as a graph.
| Parameters: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def visualize_sc_net(supplychainnet):
    """
    Visualize the supply chain network as a graph.
    Parameters:
        supplychainnet (dict): The supply chain network containing nodes and edges.
    Attributes:
        None
    Returns:
        None
    """
    G = nx.Graph()
    nodes = supplychainnet["nodes"]
    edges = supplychainnet["links"]
    # Add nodes to the graph
    for node_id, node in nodes.items():
        G.add_node(node_id, level=node.node_type)
    # Add edges to the graph
    for edge_id, edge in edges.items():
        from_node = edge.source.ID
        to_node = edge.sink.ID
        G.add_edge(from_node, to_node, weight=round(edge.lead_time(),2))
    # Generate the layout of the graph
    pos = nx.spectral_layout(G)
    # Draw the graph
    nx.draw(G, pos, node_color='#CCCCCC', with_labels=True)
    # Add edge labels
    labels = nx.get_edge_attributes(G, 'weight')
    nx.draw_networkx_edge_labels(G, pos, edge_labels=labels)
    # Set the title and display the graph
    plt.title("Supply chain network")
    plt.show()print_node_wise_performance
print_node_wise_performance(nodes_object_list)This function prints the performance metrics for each supply chain node provided in the nodes_object_list.
| Parameters: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def print_node_wise_performance(nodes_object_list):
    """
    This function prints the performance metrics for each supply chain node provided in the nodes_object_list.
    Parameters:
        nodes_object_list (list): List of supply chain node objects
    Returns: 
        None
    """
    if not nodes_object_list:
        print("No nodes provided.")
        return
    # Pre-fetch statistics from all nodes
    stats_per_node = {node.name: node.stats.get_statistics() for node in nodes_object_list}
    stat_keys = sorted(next(iter(stats_per_node.values())).keys())
    # Determine column widths
    col_width = 25
    header = "Performance Metric".ljust(col_width)
    for name in stats_per_node:
        header += name.ljust(col_width)
    print(header)
    # Print row-wise stats
    for key in stat_keys:
        row = key.ljust(col_width)
        for name in stats_per_node:
            value = stats_per_node[name].get(key, "N/A")
            row += str(value).ljust(col_width)
        print(row)get_sc_net_info
get_sc_net_info(supplychainnet)Get supply chain network information.
| Parameters: | 
 | 
|---|
| Attributes: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def get_sc_net_info(supplychainnet):
    """
    Get supply chain network information. 
    Parameters: 
        supplychainnet (dict): A dictionary representing the supply chain network.
    Attributes:
        logger (logging.Logger): The logger instance used for logging messages.
        sc_info (str): A string to accumulate the supply chain network information.
        info_keys (list): A list of keys to extract information from the supply chain network.
        keys (set): A set of keys in the supply chain network regarding performance of the network.
    Returns:
        str: A string containing the supply chain network information.
    """
    logger = global_logger.logger
    global_logger.enable_logging(log_to_screen=True)
    sc_info = "Supply chain configuration: \n"
    info_keys = ['num_of_nodes', 'num_of_links', 'num_suppliers','num_manufacturers', 'num_distributors', 'num_retailers']
    for key in info_keys:
        if key in supplychainnet.keys():
            sc_info += f"{key}: {supplychainnet[key]}\n"
            logger.info(f"{key}: {supplychainnet[key]}")
    logger.info(f"Nodes in the network: {list(supplychainnet['nodes'].keys())}")
    sc_info += "Nodes in the network:\n"
    for node_id, node in supplychainnet["nodes"].items():
        sc_info += process_info_dict(node.get_info(), logger)
    logger.info(f"Edges in the network: {list(supplychainnet['links'].keys())}")
    sc_info += "Edges in the network:\n"
    for edge_id, edge in supplychainnet["links"].items():
        sc_info += process_info_dict(edge.get_info(), logger)
    logger.info(f"Demands in the network: {list(supplychainnet['demands'].keys())}")                
    sc_info += "Demands in the network:\n"
    for demand_id, demand in supplychainnet["demands"].items():
        sc_info += process_info_dict(demand.get_info(), logger)    
    keys = supplychainnet.keys() - {'nodes', 'links', 'demands', 'env', 'num_of_nodes', 'num_of_links', 'num_suppliers','num_manufacturers', 'num_distributors', 'num_retailers'}
    sc_info += "Supply chain network performance:\n"
    logger.info("Supply chain network performance:")
    for key in sorted(keys):
        sc_info += f"{key}: {supplychainnet[key]}\n"
        logger.info(f"{key}: {supplychainnet[key]}")
    return sc_infoprocess_info_dict
process_info_dict(info_dict, logger)Processes the dictionary and logs the key-value pairs.
| Parameters: | 
 | 
|---|
| Returns: | 
 | 
|---|
Source code in src/SupplyNetPy/Components/utilities.py
              def process_info_dict(info_dict, logger):
    """
    Processes the dictionary and logs the key-value pairs.
    Parameters:
        info_dict (dict): The information dictionary to process.
        logger (logging.Logger): The logger instance used for logging messages.
    Attributes:
        None
    Returns:
        str: A string representation of the processed information.
    """
    info_string = ""
    for key, value in info_dict.items():
        if isinstance(value, object):
            value = str(value)
        if callable(value):
            value = value.__name__
        info_string += f"{key}: {value}\n"
        logger.info(f"{key}: {value}")
    return info_string