Loads and normalizes serialization configuration from YAML.
The config file specifies which properties to include for:
- Population (person.properties)
- Geography (geographical_unit.properties)
- Venues (venue.properties, per-type)
- Subsets (subset properties)
- Relationships (activity_map, hierarchies)
Source code in may/serialization/serialization_config.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
89
90
91
92
93
94
95 | class SerializationConfig:
"""
Loads and normalizes serialization configuration from YAML.
The config file specifies which properties to include for:
- Population (person.properties)
- Geography (geographical_unit.properties)
- Venues (venue.properties, per-type)
- Subsets (subset properties)
- Relationships (activity_map, hierarchies)
"""
def __init__(self, config_file):
"""
Initialize SerializationConfig.
Args:
config_file: Path to YAML configuration file
"""
self.config_file = pr.resolve(config_file)
self.config = None
# Parsed configuration sections
self.population_properties = []
self.geography_include_coordinates = True
self.geography_properties = []
self.venue_global_settings = {}
self.venue_type_properties = {}
self.relationships = {}
self.output_settings = {}
self._load_config()
def _load_config(self):
"""Load and parse the YAML configuration file."""
if not os.path.exists(self.config_file):
raise FileNotFoundError(
f"Serialization config not found: {self.config_file}"
)
logger.info(f"Loading serialization config from {self.config_file}")
self.config = load_yaml(self.config_file)
if not self.config:
raise ValueError(f"Empty serialization config: {self.config_file}")
population = self.config.get("population", {})
self.population_properties = population.get("properties", [])
logger.info(
f"Population: {len(self.population_properties)} additional properties to serialize"
)
if self.population_properties:
logger.info(f" Properties: {self.population_properties}")
geography = self.config.get("geography", {})
self.geography_include_coordinates = geography.get("include_coordinates", True)
self.geography_properties = geography.get("properties", [])
logger.info(
f"Geography: coordinates={self.geography_include_coordinates}, "
f"{len(self.geography_properties)} additional properties"
)
venues = self.config.get("venues", {})
self.venue_global_settings = venues.get("global", {})
for venue_type, type_config in venues.get("types", {}).items():
properties = type_config.get("properties", [])
self.venue_type_properties[venue_type] = properties
if properties:
logger.info(
f"Venue '{venue_type}': {len(properties)} properties to serialize"
)
logger.info(f" Properties: {properties}")
else:
logger.debug(
f"Venue '{venue_type}': minimal serialization (core attributes only)"
)
self.relationships = self.config.get("relationships", {})
self.output_settings = self.config.get("output", {})
|
__init__(config_file)
Initialize SerializationConfig.
Parameters:
| Name |
Type |
Description |
Default |
config_file
|
|
Path to YAML configuration file
|
required
|
Source code in may/serialization/serialization_config.py
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47 | def __init__(self, config_file):
"""
Initialize SerializationConfig.
Args:
config_file: Path to YAML configuration file
"""
self.config_file = pr.resolve(config_file)
self.config = None
# Parsed configuration sections
self.population_properties = []
self.geography_include_coordinates = True
self.geography_properties = []
self.venue_global_settings = {}
self.venue_type_properties = {}
self.relationships = {}
self.output_settings = {}
self._load_config()
|