Builds multiple social networks from a YAML config, storing each under
its own key in person.properties.
Source code in may/social_networks/social_networks.py
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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
81
82
83
84
85
86
87
88 | class SocialNetworkBuilder:
"""
Builds multiple social networks from a YAML config, storing each under
its own key in person.properties.
"""
def __init__(self, world, config: dict):
self.world = world
self.config = config
self._validate(config)
def _validate(self, config: dict) -> None:
for entry in config.get("networks", []):
self._validate_network_config(entry)
def _validate_network_config(self, entry: dict) -> None:
name = entry.get("name", "<unnamed>")
for key in _REQUIRED_KEYS:
if key not in entry:
raise ValueError(f"Network '{name}' missing required key '{key}'")
net_type = entry["network_type"]
if net_type not in _NETWORK_TYPES:
raise ValueError(
f"Network '{name}': unknown network_type '{net_type}'. "
f"Supported: {list(_NETWORK_TYPES)}"
)
pool_type = entry["pool_type"]
if pool_type not in pool_type_builders:
raise ValueError(
f"Network '{name}': unknown pool_type '{pool_type}'. "
f"Registered: {sorted(pool_type_builders)}"
)
@classmethod
def from_yaml(cls, world, yaml_path: str) -> "SocialNetworkBuilder":
config = load_yaml(yaml_path)
return cls(world, config)
def build_all(self) -> None:
for entry in self.config.get("networks", []):
network_name = entry.get("name", entry["storage_key"])
logger.info(
f"Building network '{network_name}' "
f"(network_type={entry['network_type']}, "
f"pool_type={entry['pool_type']}, "
f"storage_key={entry['storage_key']})"
)
if entry["network_type"] == "activity_peers":
build_activity_peers(self.world, entry)
elif entry["network_type"] == "intra_geo_unit":
build_intra_geo_unit(self.world, entry)
elif entry["network_type"] == "local_social_network":
build_local_social_network(self.world, entry)
elif entry["network_type"] == "spatial_social_network":
build_spatial_social_network(self.world, entry)
else:
build_bounded_distance(self.world, entry)
logger.info(f" Stored '{network_name}' → '{entry['storage_key']}'")
|