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)

Takes the node, link, and demand descriptions, builds the matching supply chain node, link, and demand objects, and gathers them all into one dictionary that represents the network.

Each of nodes, links, and demands must be all one kind — either all dicts (the description style) or all ready-made objects (Node / Link / Demand instances). You cannot mix the two in a single list, because the fresh simpy.Environment made for the dict items would not match the one the ready-made objects were built with. If any list contains ready-made objects, you must pass env yourself, and each object's own env must be that same environment.

Parameters:
  • nodes (list) –

    The nodes (sites): dicts, or Node objects.

  • links (list) –

    The links between them: dicts, or Link objects.

  • demands (list) –

    The customer demands: dicts, or Demand objects.

  • env (Environment, default: None ) –

    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 (set) –

    The IDs used so far, to catch duplicates.

  • num_suppliers (int) –

    Number of raw-material suppliers (finite and infinite).

  • num_manufacturers (int) –

    Number of manufacturers.

  • num_distributors (int) –

    Number of distributors and warehouses.

  • num_retailers (int) –

    Number of retailers, stores and shops.

Raises:
  • ValueError –

    If the SimPy Environment object is not provided or if there are duplicate IDs in nodes, links, or demands.

  • ValueError –

    If any of nodes / links / demands mixes dicts and pre-built domain objects.

  • ValueError –

    If a pre-built object's env does not match the provided env.

  • ValueError –

    If an invalid node type is encountered.

  • ValueError –

    If an invalid source or sink node is specified in a link.

  • ValueError –

    If a customer demand names a demand_node (the site its customers order from) that is not among the nodes.

Returns:
  • dict –

    A dictionary representing the supply chain network.

Source code in src/SupplyNetPy/Components/utilities.py
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def create_sc_net(nodes: list, links: list, demands: list, env:simpy.Environment = None):
    """
    Takes the node, link, and demand descriptions, builds the matching supply
    chain node, link, and demand objects, and gathers them all into one
    dictionary that represents the network.

    Each of ``nodes``, ``links``, and ``demands`` must be all one kind — either
    all dicts (the description style) or all ready-made objects (``Node`` /
    ``Link`` / ``Demand`` instances). You cannot mix the two in a single list,
    because the fresh ``simpy.Environment`` made for the dict items would not
    match the one the ready-made objects were built with. If any list contains
    ready-made objects, you must pass ``env`` yourself, and each object's own
    ``env`` must be that same environment.

    Parameters:
        nodes (list): The nodes (sites): dicts, or `Node` objects.
        links (list): The links between them: dicts, or `Link` objects.
        demands (list): The customer demands: dicts, or `Demand` objects.
        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 (set): The IDs used so far, to catch duplicates.
        num_suppliers (int): Number of raw-material suppliers (finite and infinite).
        num_manufacturers (int): Number of manufacturers.
        num_distributors (int): Number of distributors and warehouses.
        num_retailers (int): Number of retailers, stores and shops.

    Raises:
        ValueError: If the SimPy Environment object is not provided or if there are duplicate IDs in nodes, links, or demands.
        ValueError: If any of ``nodes`` / ``links`` / ``demands`` mixes dicts and pre-built domain objects.
        ValueError: If a pre-built object's ``env`` does not match the provided ``env``.
        ValueError: If an invalid node type is encountered.
        ValueError: If an invalid source or sink node is specified in a link.
        ValueError: If a customer demand names a ``demand_node`` (the site its customers order from) that is not among the nodes.

    Returns:
        dict: A dictionary representing the supply chain network.
    """
    # Reject lists that mix dicts and pre-built objects, right at the start. We
    # check every element, not just the first one. An earlier version only
    # looked at element 0, so a list like ``[dict, Node_instance, ...]`` slipped
    # past the "env required" check (because ``nodes[0]`` was a dict) and then
    # built the later object with a fresh env, throwing away the env that object
    # was actually created with.
    def _check_homogeneous(items, obj_cls, list_name):
        has_dict = any(isinstance(x, dict) for x in items)
        has_obj = any(isinstance(x, obj_cls) for x in items)
        if has_dict and has_obj:
            global_logger.error(
                f"{list_name} list mixes dicts and {obj_cls.__name__} instances."
            )
            raise ValueError(
                f"{list_name} list mixes dicts and {obj_cls.__name__} instances; "
                f"use all dicts or all {obj_cls.__name__} instances."
            )

    _check_homogeneous(nodes, Node, "nodes")
    _check_homogeneous(links, Link, "links")
    _check_homogeneous(demands, Demand, "demands")

    # ``env`` is required whenever ANY list contains a pre-built domain object,
    # regardless of the element's position. Scanning each list (rather than
    # indexing [0]) also avoids an IndexError on legitimately empty lists.
    any_object = (
        any(isinstance(x, Node) for x in nodes)
        or any(isinstance(x, Link) for x in links)
        or any(isinstance(x, Demand) for x in demands)
    )
    if any_object and env is None:
        global_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.error("Nodes, links, and demands cannot be empty")
        raise ValueError("Nodes, links, and demands cannot be empty")
    if(env is None):
        env = simpy.Environment()

    # When the user supplied both ``env`` and pre-built objects, each object's
    # own ``env`` must match — otherwise the returned supplychainnet would
    # combine processes from two different environments and nothing would run
    # consistently. The old code simply ignored the object's env; this check
    # surfaces the mismatch loudly.
    def _check_env_match(items, obj_cls, list_name):
        for i, x in enumerate(items):
            if isinstance(x, obj_cls) and getattr(x, "env", None) is not env:
                global_logger.error(
                    f"{list_name}[{i}] ({getattr(x, 'ID', '<no ID>')}) was built against a "
                    f"different simpy.Environment than the one passed to create_sc_net."
                )
                raise ValueError(
                    f"{list_name}[{i}] env does not match the env passed to create_sc_net."
                )

    _check_env_match(nodes, Node, "nodes")
    _check_env_match(links, Link, "links")
    _check_env_match(demands, Demand, "demands")
    supplychainnet = {"nodes":{},"links":{},"demands":{}} # create empty supply chain network
    # Use a set, not a list: checking membership and adding/removing items are
    # all fast (constant time), so building a network with many nodes stays
    # fast. ``check_duplicate_id`` works with either a set or a list.
    used_ids = set()
    # Counts of each node category, keyed by the names used in
    # ``_NODE_DISPATCH``. Using one dict (instead of four separate ``num_*``
    # variables) lets us count with a single ``counters[category] += 1`` rather
    # than a long if/else chain.
    counters = {"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"]
            try:
                nt = NodeType(node["node_type"])
            except ValueError:
                used_ids.remove(node["ID"])
                global_logger.error(f"Invalid node type {node['node_type']}")
                raise ValueError("Invalid node type")
            cls, counter_key, drop_node_type = _NODE_DISPATCH[nt]
            # ``Manufacturer.__init__`` is the only constructor that does not
            # accept ``node_type``, so for it we drop that key. The
            # ``drop_node_type`` flag records this one difference.
            kwargs = {k: v for k, v in node.items() if not (drop_node_type and k == "node_type")}
            supplychainnet["nodes"][f"{node_id}"] = cls(env=env, **kwargs)
            counters[counter_key] += 1
        elif isinstance(node, Node):
            check_duplicate_id(used_ids, node.ID, "node ID")
            node_id = node.ID
            supplychainnet["nodes"][f"{node_id}"] = node
            try:
                nt = NodeType(node.node_type)
            except ValueError:
                used_ids.remove(node.ID)
                global_logger.error(f"Invalid node type {node.node_type}")
                raise ValueError("Invalid node type")
            counters[_NODE_DISPATCH[nt][1]] += 1
    for link in links:
        if isinstance(link, dict):
            check_duplicate_id(used_ids, link["ID"], "link ID")
            source = None
            sink = None
            node_ids = supplychainnet["nodes"].keys()
            if(link["source"] in node_ids):
                source_id = link["source"]
                source = supplychainnet["nodes"][f"{source_id}"]
            if(link["sink"] in node_ids):
                sink_id = link["sink"]
                sink = supplychainnet["nodes"][f"{sink_id}"]
            if(source is None or sink is None):
                global_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):
            check_duplicate_id(used_ids, link.ID, "link ID")
            supplychainnet["links"][f"{link.ID}"] = link
    # Every link is known now, so each site's supplier-selection policy can
    # check its settings against the links (SelectProbabilistic: every link
    # has a probability). Mistakes then show before the run, not at the first
    # order. A manufacturer does not use its policy yet, so it is skipped.
    for node in supplychainnet["nodes"].values():
        check_links = getattr(getattr(node, "selection_policy", None), "check_links", None)
        if check_links is not None and not isinstance(node, Manufacturer):
            check_links()
    for d in demands:
        if isinstance(d, dict):
            check_duplicate_id(used_ids, d["ID"], "demand ID")
            demand_node = None # the site this customer demand's customers order from
            node_ids = supplychainnet["nodes"].keys()
            if d['demand_node'] in node_ids:
                demand_node_id = d['demand_node']
                demand_node = supplychainnet["nodes"][f"{demand_node_id}"]
            if(demand_node is None):
                global_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):
            check_duplicate_id(used_ids, d.ID, "demand ID")
            supplychainnet["demands"][f"{d.ID}"] = d

    supplychainnet["env"] = env
    supplychainnet["num_of_nodes"] = sum(counters.values())
    supplychainnet["num_of_links"] = len(links)
    supplychainnet.update(counters)
    return supplychainnet

check_duplicate_id

check_duplicate_id(used_ids, new_id, entity_type='ID')

Checks if new_id is already in used_ids. If it is, logs an error and raises ValueError; otherwise inserts it.

Accepts either a set (preferred — O(1) membership and insert) or a list (kept for backward compatibility with external callers — O(n) on both operations). create_sc_net passes a set; the helper duck-types the insert call so existing user code that builds its own list keeps working unchanged.

Parameters:
  • used_ids (set | list) –

    Container of already-used IDs. Mutated in place.

  • new_id (str) –

    The new ID to check and insert.

  • entity_type (str, default: 'ID' ) –

    Type of the entity for which the ID is being checked (e.g., "node ID", "link ID").

Returns:
  • –

    None

Raises:
  • ValueError –

    If new_id is already in used_ids.

Source code in src/SupplyNetPy/Components/utilities.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def check_duplicate_id(used_ids, new_id, entity_type="ID"):
    """
    Checks if ``new_id`` is already in ``used_ids``. If it is, logs an error
    and raises ``ValueError``; otherwise inserts it.

    Accepts either a ``set`` (preferred — O(1) membership and insert) or a
    ``list`` (kept for backward compatibility with external callers — O(n) on
    both operations). ``create_sc_net`` passes a ``set``; the helper duck-types
    the insert call so existing user code that builds its own ``list`` keeps
    working unchanged.

    Parameters:
        used_ids (set | list): Container of already-used IDs. Mutated in place.
        new_id (str): The new ID to check and insert.
        entity_type (str): Type of the entity for which the ID is being checked
            (e.g., ``"node ID"``, ``"link ID"``).

    Returns:
        None

    Raises:
        ValueError: If ``new_id`` is already in ``used_ids``.
    """
    if new_id in used_ids:
        global_logger.error(f"Duplicate {entity_type} {new_id}")
        raise ValueError(f"Duplicate {entity_type}")
    # A ``set`` adds items with ``.add`` and a ``list`` with ``.append``. We
    # use whichever method the passed-in container has, instead of checking its
    # exact type, so other set-like or list-like containers work too.
    insert = getattr(used_ids, "add", None) or used_ids.append
    insert(new_id)

simulate_sc_net

simulate_sc_net(supplychainnet, sim_time, logging=True, log_window=None)

Simulate the supply chain network up to a given time, and work out the network-wide results.

Parameters:
  • supplychainnet (dict) –

    A supply chain network.

  • sim_time (int) –

    The time at which the simulation stops. The run stops just before this time, so events due at exactly sim_time (for example a customer order or an expiry at that time) do not happen. To include them, pass a slightly later time, such as sim_time + 1.

  • logging (bool, default: True ) –

    Whether to enable logging for the whole run. Defaults to True. Mutually exclusive with log_window — pass log_window to confine logging to a sub-interval instead.

  • log_window (tuple[float, float], default: None ) –

    (start, stop) window during which logging is enabled; the simulation runs silently outside the window. Splits the prior overloaded-tuple use of logging into a dedicated parameter (§6.2).

Returns:
  • supplychainnet( dict ) –

    The same dictionary, with these results added. Counts are given as [number of orders, units].

    • available_inv: total inventory level of all sites at the end (infinite suppliers left out).
    • avg_available_inv: the sum, over the sites that record their inventory level (record_inv_levels=True), of each one's average recorded inventory level; 0 if none does.
    • inventory_carry_cost: total holding cost of all sites.
    • inventory_spend_cost: total paid by all sites for the units they received.
    • inventory_waste: total expired units thrown away.
    • transportation_cost: total transportation cost of replenishment orders, plus the delivery costs of customer orders.
    • revenue: total revenue of all sites.
    • total_cost: total node cost of all sites and customer demands.
    • profit: revenue minus total cost.
    • demand_by_customers: customer orders placed.
    • fulfillment_received_by_customers: customer orders that reached the customers.
    • demand_by_site: replenishment orders placed by sites.
    • fulfillment_received_by_site: replenishment orders that arrived at sites.
    • total_demand: all orders placed (the two "demand" counts above added).
    • total_fulfillment_received: all orders received (the two "fulfillment" counts above added).
    • shortage: total shortage recorded at the sites.
    • lost_sales: total customer units the sites never delivered (other than through disruptions).
    • disruption_lost_sales: total customer orders the sites lost because they were disrupted.
    • backorders: total backorders at the sites at the end.
    • avg_cost_per_order: total cost divided by the number of orders in total_demand (0 if none).
    • avg_cost_per_item: total cost divided by the units in total_demand (0 if none); "item" here means unit.
Source code in src/SupplyNetPy/Components/utilities.py
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def simulate_sc_net(supplychainnet, sim_time, logging=True, log_window=None):
    """
    Simulate the supply chain network up to a given time, and work out the network-wide results.

    Parameters:
        supplychainnet (dict): A supply chain network.
        sim_time (int): The time at which the simulation stops. The run stops just
            before this time, so events due at exactly ``sim_time`` (for example a
            customer order or an expiry at that time) do not happen. To include
            them, pass a slightly later time, such as ``sim_time + 1``.
        logging (bool): Whether to enable logging for the whole run. Defaults
            to ``True``. Mutually exclusive with ``log_window`` — pass
            ``log_window`` to confine logging to a sub-interval instead.
        log_window (tuple[float, float], optional): ``(start, stop)`` window
            during which logging is enabled; the simulation runs silently
            outside the window. Splits the prior overloaded-tuple use of
            ``logging`` into a dedicated parameter (§6.2).

    Returns:
        supplychainnet (dict): The same dictionary, with these results added. Counts are given as
            ``[number of orders, units]``.

            - ``available_inv``: total inventory level of all sites at the end (infinite suppliers left out).
            - ``avg_available_inv``: the sum, over the sites that record their inventory level
              (``record_inv_levels=True``), of each one's average recorded inventory level; 0 if none does.
            - ``inventory_carry_cost``: total holding cost of all sites.
            - ``inventory_spend_cost``: total paid by all sites for the units they received.
            - ``inventory_waste``: total expired units thrown away.
            - ``transportation_cost``: total transportation cost of replenishment orders, plus the delivery
              costs of customer orders.
            - ``revenue``: total revenue of all sites.
            - ``total_cost``: total node cost of all sites and customer demands.
            - ``profit``: revenue minus total cost.
            - ``demand_by_customers``: customer orders placed.
            - ``fulfillment_received_by_customers``: customer orders that reached the customers.
            - ``demand_by_site``: replenishment orders placed by sites.
            - ``fulfillment_received_by_site``: replenishment orders that arrived at sites.
            - ``total_demand``: all orders placed (the two "demand" counts above added).
            - ``total_fulfillment_received``: all orders received (the two "fulfillment" counts above added).
            - ``shortage``: total shortage recorded at the sites.
            - ``lost_sales``: total customer units the sites never delivered (other than through disruptions).
            - ``disruption_lost_sales``: total customer orders the sites lost because they were disrupted.
            - ``backorders``: total backorders at the sites at the end.
            - ``avg_cost_per_order``: total cost divided by the number of orders in ``total_demand`` (0 if none).
            - ``avg_cost_per_item``: total cost divided by the units in ``total_demand`` (0 if none); "item"
              here means unit.
    """
    logger = global_logger
    env = supplychainnet["env"]

    # For backward compatibility: the original API let you pass
    # ``logging=(start, stop)`` to ask for a logging window. We still accept
    # that, but convert it to ``log_window`` here so the rest of this function
    # only deals with one form of each argument.
    if isinstance(logging, tuple) and len(logging) == 2:
        if log_window is not None:
            message = "simulate_sc_net: both logging=(start, stop) and log_window= were given; log_window is used."
            logger.warning(message)
            warnings.warn(message, UserWarning, stacklevel=2)
        else:
            # FutureWarning, not DeprecationWarning: Python shows FutureWarning
            # to everyone, while DeprecationWarning is hidden outside the main script.
            message = "simulate_sc_net: logging=(start, stop) is deprecated; use log_window=(start, stop) instead."
            logger.warning(message)
            warnings.warn(message, FutureWarning, stacklevel=2)
            log_window = logging
        logging = True

    if(sim_time<=env.now):
        message = (f"simulate_sc_net: sim_time ({sim_time}) is not later than the current simulation time "
                   f"({env.now}), so nothing more is simulated. To start again, create a new network with "
                   f"create_sc_net(); to run this one further, give a later sim_time.")
        logger.warning(message)
        warnings.warn(message, UserWarning, stacklevel=2)
        logger.info(f"Performance measures for the supply chain network are calculated and returned.")
    elif log_window is not None:
        assert isinstance(log_window, tuple) and len(log_window) == 2, "log_window must be a (start, stop) tuple"
        assert log_window[0] < log_window[1], "log_window start should be less than stop"
        assert log_window[0] >= 0, "log_window start should be greater than or equal to 0"
        assert log_window[1] <= sim_time, "log_window stop should be less than or equal to simulation time"
        log_start, log_stop = log_window
        global_logger.disable_logging()
        env.run(log_start)
        global_logger.enable_logging()
        env.run(log_stop)
        global_logger.disable_logging()
        if(sim_time > log_stop):
            env.run(sim_time)
    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

    # Network-wide totals, added up from every site and customer demand
    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, units]
    total_fulfillment_received_by_customers = [0, 0] # [orders, units]
    total_demand_by_site = [0, 0] # [orders, units]
    total_fulfillment_received_by_site = [0, 0] # [orders, units]
    total_demand_placed = [0, 0] # [orders, units]
    total_fulfillment_received = [0, 0] # [orders, units]
    total_shortage = [0, 0] # [orders, units]
    total_lost_sales = [0, 0] # [orders, units]
    total_disruption_lost_sales = [0, 0] # [orders, units]
    total_backorders = [0, 0] # [orders, units]

    for key, node in supplychainnet["nodes"].items():
        if(node.node_type == NodeType.INFINITE_SUPPLIER): # skip infinite suppliers
            continue
        node.stats.update_stats() # update stats for the node
        total_available_inv += node.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.shortage[0]
        total_shortage[1] += node.stats.shortage[1]
        total_lost_sales[0] += node.stats.lost_sales[0]
        total_lost_sales[1] += node.stats.lost_sales[1]
        total_disruption_lost_sales[0] += node.stats.disruption_lost_sales[0]
        total_disruption_lost_sales[1] += node.stats.disruption_lost_sales[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] # units
        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.shortage[0]
        total_shortage[1] += node.stats.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["lost_sales"] = total_lost_sales
    supplychainnet["disruption_lost_sales"] = total_disruption_lost_sales
    supplychainnet["backorders"] = total_backorders
    # Average cost per order and per unit (the second key keeps its old name, avg_cost_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 log_window is not None:
        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 supplychainnet

visualize_sc_net

visualize_sc_net(supplychainnet)

Visualize the supply chain network as a graph.

Lays the nodes out in columns by type (using networkx.multipartite_layout) so the drawing reads left-to-right (raw-material suppliers → manufacturers → distributors and warehouses → retailers → customer demands). Each link is labelled with one draw of its lead time. The old default, spectral_layout, often placed nodes badly for small networks.

Parameters:
  • supplychainnet (dict) –

    The supply chain network: its nodes, links and customer demands.

Returns:
  • –

    None

Source code in src/SupplyNetPy/Components/utilities.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def visualize_sc_net(supplychainnet):
    """
    Visualize the supply chain network as a graph.

    Lays the nodes out in columns by type (using
    `networkx.multipartite_layout`) so the drawing reads left-to-right
    (raw-material suppliers → manufacturers → distributors and warehouses →
    retailers → customer demands). Each link is labelled with one draw of its
    lead time. The old default, ``spectral_layout``, often placed nodes badly
    for small networks.

    Parameters:
        supplychainnet (dict): The supply chain network: its nodes, links and customer demands.

    Returns:
        None
    """
    G = nx.DiGraph()
    nodes = supplychainnet["nodes"]
    links = supplychainnet["links"]

    # Put each node in a column by its NodeType, for multipartite_layout.
    # Customer demands are kept apart from ``nodes`` in the dictionary, but are
    # still drawn: as the right-most column, each joined to the site its
    # customers order from (its ``demand_node``).
    for node_id, node in nodes.items():
        tier = _TIER_INDEX.get(str(node.node_type).lower(), 2)
        G.add_node(node_id, subset=tier, level=node.node_type)

    for link_id, link in links.items():
        G.add_edge(link.source.ID, link.sink.ID, weight=round(link.lead_time(), 2))

    for demand_id, demand in supplychainnet.get("demands", {}).items():
        G.add_node(demand_id, subset=_TIER_INDEX["demand"], level="demand")
        target = getattr(demand, "demand_node", None)
        if target is not None and target.ID in nodes:
            G.add_edge(target.ID, demand_id)

    pos = nx.multipartite_layout(G, subset_key="subset")
    nx.draw(G, pos, node_color='#CCCCCC', with_labels=True, arrows=True)
    labels = nx.get_edge_attributes(G, 'weight')
    nx.draw_networkx_edge_labels(G, pos, edge_labels=labels)
    plt.title("Supply chain network")
    plt.show()

get_node_wise_performance

get_node_wise_performance(nodes_object_list)

Collect per-node performance statistics into a structured form.

A library should return data rather than print it, so this function gathers each node's stats.get_statistics() dict and returns the rows as a list of dicts keyed by the metric name. Callers who want a table can feed the result to pandas.DataFrame.from_records; callers who want the plain-text layout can pass it to format_node_wise_performance or print_node_wise_performance.

Parameters:
  • nodes_object_list (list) –

    List of supply chain node objects.

Returns:
  • –

    list[dict]: One dict per metric, with the key "Performance Metric" holding the metric name and one entry per node holding the value. Returns an empty list when nodes_object_list is empty.

Source code in src/SupplyNetPy/Components/utilities.py
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
def get_node_wise_performance(nodes_object_list):
    """
    Collect per-node performance statistics into a structured form.

    A library should return data rather than print it, so this function gathers
    each node's ``stats.get_statistics()`` dict and returns the rows as a list
    of dicts keyed by the metric name. Callers who want a table can feed the
    result to ``pandas.DataFrame.from_records``; callers who want the plain-text
    layout can pass it to `format_node_wise_performance` or
    `print_node_wise_performance`.

    Parameters:
        nodes_object_list (list): List of supply chain node objects.

    Returns:
        list[dict]: One dict per metric, with the key ``"Performance Metric"``
            holding the metric name and one entry per node holding the value.
            Returns an empty list when ``nodes_object_list`` is empty.
    """
    if not nodes_object_list:
        return []
    return _performance_rows(nodes_object_list)

format_node_wise_performance

format_node_wise_performance(nodes_object_list, col_width: int = 25) -> str

Format the per-node performance table as a fixed-width text block.

Builds a string only — it does not print. The lines are collected in a list and joined once at the end, which keeps it fast even for many nodes.

Parameters:
  • nodes_object_list (list) –

    List of supply chain node objects.

  • col_width (int, default: 25 ) –

    Smallest width of each column. Defaults to 25. A column is made wider when one of its entries is too long for it.

Returns:
  • str( str ) –

    Multi-line table; an empty string when no nodes are provided.

Source code in src/SupplyNetPy/Components/utilities.py
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
def format_node_wise_performance(nodes_object_list, col_width: int = 25) -> str:
    """
    Format the per-node performance table as a fixed-width text block.

    Builds a string only — it does not print. The lines are collected in a list
    and joined once at the end, which keeps it fast even for many nodes.

    Parameters:
        nodes_object_list (list): List of supply chain node objects.
        col_width (int): Smallest width of each column. Defaults to 25. A
            column is made wider when one of its entries is too long for it.

    Returns:
        str: Multi-line table; an empty string when no nodes are provided.
    """
    if not nodes_object_list:
        return ""
    return _format_performance(nodes_object_list, col_width)

print_node_wise_performance

print_node_wise_performance(nodes_object_list)

Print the per-node performance table to stdout.

Thin wrapper around format_node_wise_performance; kept as a library convenience for the existing notebooks/scripts that call it directly. Programmatic consumers should prefer get_node_wise_performance (returns a list of dicts) so the table can be rendered however the caller prefers.

Parameters:
  • nodes_object_list (list) –

    List of supply chain node objects.

Returns:
  • –

    None

Source code in src/SupplyNetPy/Components/utilities.py
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
def print_node_wise_performance(nodes_object_list):
    """
    Print the per-node performance table to stdout.

    Thin wrapper around `format_node_wise_performance`; kept as a
    library convenience for the existing notebooks/scripts that call it
    directly. Programmatic consumers should prefer
    `get_node_wise_performance` (returns a list of dicts) so the table
    can be rendered however the caller prefers.

    Parameters:
        nodes_object_list (list): List of supply chain node objects.

    Returns:
        None
    """
    if not nodes_object_list:
        print("No nodes provided.")
        return
    print(format_node_wise_performance(nodes_object_list))
get_link_wise_performance(links)

Collect the statistics of each link (link.stats.get_statistics()) into a structured form, in the same layout as get_node_wise_performance.

Each link records the replenishment orders its buyer placed over it, what arrived, how long the orders waited and travelled, what they cost, and the link's disruptions (see LinkStatistics). To see one site's links, pass node.suppliers (what the site ordered, and from whom) or node.buyers (who ordered from the site, and what they got). For every link in a network, pass supplychainnet["links"].values().

Parameters:
  • links (list) –

    The links to include.

Returns:
  • –

    list[dict]: One dict per metric, with the key "Performance Metric" holding the metric name and one entry per link (headed by the link's name) holding the value. Returns an empty list when links is empty.

Source code in src/SupplyNetPy/Components/utilities.py
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
def get_link_wise_performance(links):
    """
    Collect the statistics of each link (``link.stats.get_statistics()``) into
    a structured form, in the same layout as `get_node_wise_performance`.

    Each link records the replenishment orders its buyer placed over it, what
    arrived, how long the orders waited and travelled, what they cost, and the
    link's disruptions (see `LinkStatistics`). To see one site's links, pass
    ``node.suppliers`` (what the site ordered, and from whom) or
    ``node.buyers`` (who ordered from the site, and what they got). For every
    link in a network, pass ``supplychainnet["links"].values()``.

    Parameters:
        links (list): The links to include.

    Returns:
        list[dict]: One dict per metric, with the key ``"Performance Metric"``
            holding the metric name and one entry per link (headed by the
            link's name) holding the value. Returns an empty list when
            ``links`` is empty.
    """
    links = list(links)
    if not links:
        return []
    return _performance_rows(links)
format_link_wise_performance(links, col_width: int = 25) -> str

Format the statistics of each link as a fixed-width text table, in the same layout as format_node_wise_performance. Each column is headed by the link's name (such as "D1 to R1"), with its ID added when two links have the same name. It builds a string only and does not print.

Parameters:
  • links (list) –

    The links to include, for example node.suppliers.

  • col_width (int, default: 25 ) –

    Smallest width of each column. Defaults to 25. A column is made wider when one of its entries is too long for it.

Returns:
  • str( str ) –

    Multi-line table; an empty string when no links are given.

Source code in src/SupplyNetPy/Components/utilities.py
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
def format_link_wise_performance(links, col_width: int = 25) -> str:
    """
    Format the statistics of each link as a fixed-width text table, in the
    same layout as `format_node_wise_performance`. Each column is headed by
    the link's name (such as "D1 to R1"), with its ID added when two links
    have the same name. It builds a string only and does not print.

    Parameters:
        links (list): The links to include, for example ``node.suppliers``.
        col_width (int): Smallest width of each column. Defaults to 25. A
            column is made wider when one of its entries is too long for it.

    Returns:
        str: Multi-line table; an empty string when no links are given.
    """
    links = list(links)
    if not links:
        return ""
    return _format_performance(links, col_width)
print_link_wise_performance(links)

Print the statistics of each link as a table (see format_link_wise_performance). Nothing is printed by the library unless this is called. Use get_link_wise_performance to get the values as data instead.

Parameters:
  • links (list) –

    The links to include, for example node.suppliers or node.buyers.

Returns:
  • –

    None

Source code in src/SupplyNetPy/Components/utilities.py
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
def print_link_wise_performance(links):
    """
    Print the statistics of each link as a table (see
    `format_link_wise_performance`). Nothing is printed by the library unless
    this is called. Use `get_link_wise_performance` to get the values as data
    instead.

    Parameters:
        links (list): The links to include, for example ``node.suppliers`` or
            ``node.buyers``.

    Returns:
        None
    """
    links = list(links)
    if not links:
        print("No links provided.")
        return
    print(format_link_wise_performance(links))

reset_stats

reset_stats(supplychainnet)

Start the statistics of the whole network again from zero: every node, every customer demand and every link. This is what a warm-up uses. A warm-up is an opening stretch of a run that is left out of the statistics, because a run starts in an unusual state (nothing in transit, no orders waiting). Run for the warm-up period, reset, and carry on running:

scm.simulate_sc_net(sc, sim_time=31)   # warm-up
scm.reset_stats(sc)
scm.simulate_sc_net(sc, sim_time=T + 1)

The simulation carries on unchanged and only the counting starts again. Each node, customer demand and link keeps what describes the present (see Statistics.reset and LinkStatistics.reset). The network results in supplychainnet are not cleared here; simulate_sc_net works them out again from the statistics when it next stops.

Parameters:
  • supplychainnet (dict) –

    The supply chain network, as returned by create_sc_net.

Returns:
  • –

    None

Source code in src/SupplyNetPy/Components/utilities.py
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
def reset_stats(supplychainnet):
    """
    Start the statistics of the whole network again from zero: every node,
    every customer demand and every link. This is what a warm-up uses. A
    warm-up is an opening stretch of a run that is left out of the
    statistics, because a run starts in an unusual state (nothing in transit,
    no orders waiting). Run for the warm-up period, reset, and carry on
    running:

    ```python
    scm.simulate_sc_net(sc, sim_time=31)   # warm-up
    scm.reset_stats(sc)
    scm.simulate_sc_net(sc, sim_time=T + 1)
    ```

    The simulation carries on unchanged and only the counting starts again.
    Each node, customer demand and link keeps what describes the present (see
    ``Statistics.reset`` and ``LinkStatistics.reset``). The network results in
    ``supplychainnet`` are not cleared here; ``simulate_sc_net`` works them
    out again from the statistics when it next stops.

    Parameters:
        supplychainnet (dict): The supply chain network, as returned by
            `create_sc_net`.

    Returns:
        None
    """
    for group in ("nodes", "demands", "links"):
        for item in supplychainnet.get(group, {}).values():
            item.stats.reset()

get_sc_net_info

get_sc_net_info(supplychainnet)

Describe the supply chain network: how many nodes and links it has, the settings of each node, link and customer demand, and the results of the last simulation (if any). Everything is also logged to the screen.

Parameters:
  • supplychainnet (dict) –

    A dictionary representing the supply chain network.

Attributes:
  • logger (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.

Source code in src/SupplyNetPy/Components/utilities.py
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_sc_net_info(supplychainnet):
    """
    Describe the supply chain network: how many nodes and links it has, the
    settings of each node, link and customer demand, and the results of the
    last simulation (if any). Everything is also logged to the screen.

    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
    global_logger.enable_logging(log_to_screen=True)
    # Collect the lines in a list and join them once at the end. Repeatedly
    # growing a string with ``+=`` is slow, because Python rebuilds the whole
    # string each time.
    parts = ["Supply chain configuration: "]
    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():
            parts.append(f"{key}: {supplychainnet[key]}")
            logger.info(f"{key}: {supplychainnet[key]}")
    logger.info(f"Nodes in the network: {list(supplychainnet['nodes'].keys())}")
    parts.append("Nodes in the network:")
    for node_id, node in supplychainnet["nodes"].items():
        parts.append(process_info_dict(node.get_info(), logger).rstrip("\n"))
    logger.info(f"Edges in the network: {list(supplychainnet['links'].keys())}")
    parts.append("Edges in the network:")
    for link_id, link in supplychainnet["links"].items():
        parts.append(process_info_dict(link.get_info(), logger).rstrip("\n"))
    logger.info(f"Demands in the network: {list(supplychainnet['demands'].keys())}")
    parts.append("Demands in the network:")
    for demand_id, demand in supplychainnet["demands"].items():
        parts.append(process_info_dict(demand.get_info(), logger).rstrip("\n"))
    keys = supplychainnet.keys() - {'nodes', 'links', 'demands', 'env', 'num_of_nodes', 'num_of_links', 'num_suppliers','num_manufacturers', 'num_distributors', 'num_retailers'}
    parts.append("Supply chain network performance:")
    logger.info("Supply chain network performance:")
    for key in sorted(keys):
        parts.append(f"{key}: {supplychainnet[key]}")
        logger.info(f"{key}: {supplychainnet[key]}")
    return "\n".join(parts) + "\n"

process_info_dict

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 (Logger) –

    The logger instance used for logging messages.

Returns:
  • str –

    A string representation of the processed information.

Source code in src/SupplyNetPy/Components/utilities.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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.
    """
    # Collect the lines in a list and join them once at the end. Repeatedly
    # adding to a string with ``+=`` is slow, because Python has to rebuild the
    # whole string every time — which adds up for networks with 100+ nodes that
    # each produce dozens of lines.
    parts = []
    for key, value in info_dict.items():
        # A policy as text is only its name. Show its settings too, so that,
        # for example, two retailers that split their orders 70/30 and 50/50
        # can be told apart.
        if isinstance(value, (InventoryReplenishment, SupplierSelectionPolicy)):
            value = _describe_policy(value)
        if isinstance(value, object):
            value = str(value)
        if callable(value):
            value = value.__name__
        parts.append(f"{key}: {value}")
        logger.info(f"{key}: {value}")
    return "\n".join(parts) + ("\n" if parts else "")