218
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 | 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.
Each of ``nodes``, ``links``, and ``demands`` must be homogeneous — either
entirely dicts (netlist style) or entirely pre-built domain objects
(``Node`` / ``Link`` / ``Demand`` instances). Mixing the two within a single
list is rejected, because a fresh ``simpy.Environment`` created for the
dict items would not match the one the object items were built against.
When any list contains pre-built objects, ``env`` must be passed explicitly
and each object's own ``env`` must match it.
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 mixed dict-and-object lists up-front. Scanning every element means
# we no longer fall for the old first-element-only trap: a list like
# ``[dict, Node_instance, ...]`` used to pass the env-required check
# (because ``nodes[0]`` was a dict) and then silently fall into the
# ``isinstance(node, Node)`` branch at a later index with a fresh env,
# dropping the object's real env on the floor.
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
# Set rather than list: ``in`` is O(1) and ``.add``/``.remove`` are O(1) too,
# so building a network with N nodes is O(N) instead of O(N^2). The
# ``check_duplicate_id`` helper duck-types its insert call, so passing a
# set just works.
used_ids = set()
# Counters keyed by the category name used in ``_NODE_DISPATCH``. A dict
# replaces the four separate ``num_*`` locals so dispatch is a single
# ``counters[category] += 1`` instead of an if/elif ladder.
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__`` does not accept ``node_type`` — every
# other constructor does. The ``drop_node_type`` flag is the one
# place this asymmetry is encoded.
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
|