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

    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 (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 (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 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 an invalid demand node is specified in a demand.

Returns:
  • dict

    A dictionary representing the supply chain network.

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
267
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
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): 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 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 an invalid demand node is specified in a demand.

    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
    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
            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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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 for a given time period, and calculate performance measures.

Parameters:
  • supplychainnet (dict) –

    A supply chain network.

  • sim_time (int) –

    Simulation time.

  • 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 ) –

    Updated dict with listed performance measures.

Source code in src/SupplyNetPy/Components/utilities.py
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
458
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
def simulate_sc_net(supplychainnet, sim_time, logging=True, log_window=None):
    """
    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.
        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): Updated dict with listed performance measures.
    """
    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:
            logger.warning("simulate_sc_net: both logging=tuple and log_window= were provided; log_window takes precedence.")
        else:
            logger.warning("simulate_sc_net: logging=(start, stop) is deprecated; use log_window=(start, stop) instead.")
            log_window = logging
        logging = True

    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 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

    # 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(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_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.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["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 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 :func:networkx.multipartite_layout) so the drawing reads left-to-right (suppliers → manufacturers → distributors → retailers → demand). The old default, spectral_layout, often placed nodes badly for small networks.

Parameters:
  • supplychainnet (dict) –

    The supply chain network containing nodes and edges.

Returns:
  • None

Source code in src/SupplyNetPy/Components/utilities.py
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def visualize_sc_net(supplychainnet):
    """
    Visualize the supply chain network as a graph.

    Lays the nodes out in columns by type (using
    :func:`networkx.multipartite_layout`) so the drawing reads left-to-right
    (suppliers → manufacturers → distributors → retailers → demand). The old
    default, ``spectral_layout``, often placed nodes badly for small networks.

    Parameters:
        supplychainnet (dict): The supply chain network containing nodes and edges.

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

    # Tier each node by its NodeType so multipartite_layout can column it.
    # ``demands`` are not in ``nodes`` but are still useful to visualise; add
    # them as the right-most tier with edges from their 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 edge_id, edge in edges.items():
        G.add_edge(edge.source.ID, edge.sink.ID, weight=round(edge.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()

print_node_wise_performance

print_node_wise_performance(nodes_object_list)

Print the per-node performance table to stdout.

Thin wrapper around :func:format_node_wise_performance; kept as a library convenience for the existing notebooks/scripts that call it directly. Programmatic consumers should prefer :func: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
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
def print_node_wise_performance(nodes_object_list):
    """
    Print the per-node performance table to stdout.

    Thin wrapper around :func:`format_node_wise_performance`; kept as a
    library convenience for the existing notebooks/scripts that call it
    directly. Programmatic consumers should prefer
    :func:`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_sc_net_info

get_sc_net_info(supplychainnet)

Get supply chain network information.

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
172
173
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 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
    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 edge_id, edge in supplychainnet["links"].items():
        parts.append(process_info_dict(edge.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
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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():
        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 "")