Save the configuration to the shared configuration JSON file. Holds _config_lock so concurrent writers to self.config can't change the dict mid-serialization (would raise "dictionary changed size during iteration"). Serializes a shallow snapshot to keep the lock wind
(self)
| 1550 | try: |
| 1551 | with open(self.netkbfile, 'w', newline='') as file: |
| 1552 | writer = csv.writer(file) |
| 1553 | writer.writerow(headers) |
| 1554 | logger.info(f"Network knowledge base CSV file created at {self.netkbfile}") |
| 1555 | except IOError as e: |
| 1556 | logger.error(f"Error writing to netkbfile: {e}") |
| 1557 | except Exception as e: |
| 1558 | logger.error(f"Unexpected error while writing to netkbfile: {e}") |
| 1559 | else: |
| 1560 | logger.info(f"Network knowledge base CSV file already exists at {self.netkbfile}") |
| 1561 | except Exception as e: |
| 1562 | logger.error(f"Unexpected error in initialize_csv: {e}") |
| 1563 | |
| 1564 | |
| 1565 | def load_config(self): |
| 1566 | """Load the configuration from the shared configuration JSON file.""" |
| 1567 | try: |
| 1568 | logger.info("Loading configuration...") |
| 1569 | if os.path.exists(self.shared_config_json): |
| 1570 | # Check if file is empty before attempting to parse |
| 1571 | if os.path.getsize(self.shared_config_json) == 0: |
| 1572 | logger.warning("Configuration file is empty, creating new one with default values...") |
| 1573 | self.save_config() |
| 1574 | return |
| 1575 | with open(self.shared_config_json, 'r') as f: |
| 1576 | config = json.load(f) |
| 1577 | config = self._normalize_config_keys(config) |
| 1578 | self.config.update(config) |
| 1579 | self.config = self._normalize_config_keys(self.config) |
| 1580 | for key, value in self.config.items(): |
| 1581 | setattr(self, key, value) |
| 1582 | self._remove_legacy_attributes() |
no test coverage detected