Coverage for lode / reader / config_manager.py: 84%

1# config_magarer.py - STRATEGIE CON CONFIG UNIFICATA 

2 

3import yaml 

4from pathlib import Path 

5from abc import ABC, abstractmethod 

6from rdflib import URIRef, Graph, Node 

7from lode.models import * 

8 

9class ConfigManager(ABC): 

10 """Strategia basata su config YAML unico""" 

11 

12 def __init__(self): 

13 self.config = self._load_config() 

14 self._type_mapping_cache = None 

15 self._property_mapping_cache = None 

16 

17 @abstractmethod 

18 def create_logic(self, graph: Graph, cache: dict): 

19 """Factory method per creare Logic specifica""" 

20 pass 

21 

22 @property 

23 @abstractmethod 

24 def config_name(self) -> str: 

25 """Nome del file config specifico: 'owl' | 'skos' | 'rdfs'""" 

26 pass 

27 

28 def _load_config(self) -> dict: 

29 config_dir = Path(__file__).parent / 'config' 

30 

31 with open(config_dir / 'base.yaml') as f: 

32 base = yaml.safe_load(f) 

33 

34 with open(config_dir / f'{self.config_name}.yaml') as f: 

35 specific = yaml.safe_load(f) 

36 

37 return self._deep_merge(base, specific) 

38 

39 def _deep_merge(self, base: dict, override: dict) -> dict: 

40 result = base.copy() 

41 for key, value in override.items(): 

42 if key in ('name', 'inherits'): 

43 continue 

44 if key in result and isinstance(result[key], dict) and isinstance(value, dict): 

45 result[key] = self._deep_merge(result[key], value) 

46 else: 

47 result[key] = value 

48 return result 

49 

50 # def _load_config(self) -> dict: 

51 # """Carica config da file YAML unico""" 

52 # config_path = Path(__file__).parent / 'config' / 'base.yaml' 

53 

54 # with open(config_path, 'r') as f: 

55 # config = yaml.safe_load(f) 

56 

57 # return config 

58 

59 @abstractmethod 

60 def create_viewer(self, reader): 

61 """Factory method per creare Viewer specifico""" 

62 pass 

63 

64 # ========== NAMESPACE MAP ========== 

65 NAMESPACES = { 

66 'owl': 'http://www.w3.org/2002/07/owl#', 

67 'rdfs': 'http://www.w3.org/2000/01/rdf-schema#', 

68 'rdf': 'http://www.w3.org/1999/02/22-rdf-syntax-ns#', 

69 'skos': 'http://www.w3.org/2004/02/skos/core#', 

70 'xsd': "http://www.w3.org/2001/XMLSchema#", 

71 'swrl': 'http://www.w3.org/2003/11/swrl#' 

72 } 

73 

74 def _parse_uri(self, uri_str: str) -> URIRef: 

75 """owl:Class -> URIRef""" 

76 if ':' not in uri_str: 

77 return URIRef(uri_str) 

78 

79 prefix, local = uri_str.split(':', 1) 

80 return URIRef(self.NAMESPACES[prefix] + local) 

81 

82 def _parse_class(self, class_name: str): 

83 import lode.models as _models 

84 cls = getattr(_models, class_name, None) 

85 if cls is None: 

86 raise KeyError(class_name) 

87 return cls 

88 

89 def _parse_value(self, value): 

90 if value == 'Literal': 

91 return 'Literal' 

92 elif value is True or value is False: 

93 return value 

94 elif isinstance(value, str): 

95 import lode.models as _models 

96 cls = getattr(_models, value, None) 

97 if cls is not None: 

98 return cls 

99 return value 

100 

101 # ========== CONFIG ACCESSORS ========== 

102 

103 def get_type_mapping(self) -> dict[URIRef, dict]: 

104 """rdf:type -> {target_class, setters, ...}""" 

105 if self._type_mapping_cache is None: 

106 self._type_mapping_cache = { 

107 self._parse_uri(uri): self._parse_config(cfg) 

108 for uri, cfg in self.config['mapper'].items() 

109 if cfg.get('is') == 'class' 

110 } 

111 return self._type_mapping_cache 

112 

113 def get_property_mapping(self) -> dict[URIRef, dict]: 

114 """predicate -> {target_classes, setters, handler, ...}""" 

115 if self._property_mapping_cache is None: 

116 self._property_mapping_cache = { 

117 self._parse_uri(uri): self._parse_config(cfg) 

118 for uri, cfg in self.config['mapper'].items() 

119 if cfg.get('is') == 'predicate' 

120 } 

121 return self._property_mapping_cache 

122 

123 def _parse_config(self, cfg: dict) -> dict: 

124 """Parse un blocco di config""" 

125 parsed = cfg.copy() 

126 

127 if 'target_class' in parsed: 

128 parsed['target_class'] = self._parse_class(parsed['target_class']) 

129 

130 if 'inferred_class' in parsed: 

131 parsed['inferred_class'] = self._parse_class(parsed['inferred_class']) 

132 

133 if 'target_classes' in parsed: 

134 parsed['target_classes'] = [ 

135 self._parse_class(tc) for tc in parsed['target_classes'] 

136 ] 

137 

138 if 'setters' in parsed: 

139 parsed['setters'] = [ 

140 {k: self._parse_value(v)} if isinstance(item, dict) else item 

141 for item in parsed['setters'] 

142 for k, v in (item.items() if isinstance(item, dict) else [(item, None)]) 

143 ] 

144 

145 return parsed 

146 

147 def get_group_axioms(self) -> dict[URIRef, str]: 

148 """Group axioms: URI -> handler_name""" 

149 return { 

150 self._parse_uri(uri): handler 

151 for uri, handler in self.config.get('enricher', {}).items() 

152 } 

153 

154 def get_fallback_class(self) -> type | None: 

155 fallback = self.config.get('mapper', {}).get('fallback_class') 

156 if fallback: 

157 import lode.models as _models 

158 return getattr(_models, fallback, Statement) 

159 return None 

160 

161 # ========== HELPER METHODS ========== 

162 

163 def get_classifier_predicates(self) -> set[URIRef]: 

164 """Predicati con 'inferred_class'""" 

165 return { 

166 pred for pred, cfg in self.get_property_mapping().items() 

167 if 'inferred_class' in cfg 

168 } 

169 

170 def classify_by_predicate(self, uri: Node, graph: Graph) -> type | None: 

171 """Classifica guardando predicati. 

172 - inferred_class: tipo del soggetto (BNode restrictions, quantifiers, etc.) -> priorita' massima 

173 Se piu' predicati hanno inferred_class, vince la classe piu' specifica (issubclass). 

174 - target_classes con 1 elemento, nessun inferred_class, classify != false:  

175 tipo implicito del soggetto (fallback) 

176 - target_classes con 2+ elementi: ambiguo, gestito dall'handler -> None 

177 - classify: false nel config: il predicato non classifica il soggetto, 

178 serve solo per i setters in phase3 (es. facet XSD) 

179 """ 

180 fallback = None 

181 inferred = None 

182 for predicate, cfg in self.get_property_mapping().items(): 

183 if (uri, predicate, None) in graph: 

184 if 'inferred_class' in cfg: 

185 candidate = cfg['inferred_class'] 

186 if inferred is None or issubclass(candidate, inferred): 

187 inferred = candidate 

188 elif fallback is None and len(cfg.get('target_classes', [])) == 1: 

189 if cfg.get('classify', True): 

190 fallback = cfg['target_classes'][0] 

191 return inferred if inferred else fallback 

192 

193 def get_punning_priority(self): 

194 names = self.config.get('punning_priority', []) 

195 return [self._parse_class(n) for n in names] 

196 

197 

198# config_manager.py - aggiungi nei concrete managers 

199 

200class OwlConfigManager(ConfigManager): 

201 

202 @property 

203 def config_name(self) -> str: 

204 return 'owl' 

205 

206 def create_logic(self, graph: Graph, cache: dict): 

207 from lode.reader.logic import OwlLogic 

208 return OwlLogic(graph, cache, self) 

209 

210 def create_viewer(self, reader): 

211 from lode.viewer import OwlViewer 

212 return OwlViewer(reader) 

213 

214class RdfConfigManager(ConfigManager): 

215 @property 

216 def config_name(self) -> str: 

217 return 'rdf' 

218 

219 def create_logic(self, graph: Graph, cache: dict): 

220 from lode.reader.logic import RdfLogic 

221 return RdfLogic(graph, cache, self) 

222 

223 def create_viewer(self, reader): 

224 from lode.viewer import RdfViewer 

225 return RdfViewer(reader) 

226 

227 

228class SkosConfigManager(ConfigManager): 

229 @property 

230 def config_name(self) -> str: 

231 return 'skos' 

232 

233 def create_logic(self, graph: Graph, cache: dict): 

234 from lode.reader.logic import SkosLogic 

235 return SkosLogic(graph, cache, self) 

236 

237 def create_viewer(self, reader): 

238 from lode.viewer import SkosViewer 

239 return SkosViewer(reader) 

240 

241# ========== CONFIGURATIONS REGISTRY ========== 

242 

243CONFIGURATION_REGISTRY = { 

244 'OWL': OwlConfigManager, 

245 'SKOS': SkosConfigManager, 

246 'RDF': RdfConfigManager 

247} 

248 

249 

250def get_configuration(configuration_name: str) -> ConfigManager: 

251 key = configuration_name.upper() 

252 if key not in CONFIGURATION_REGISTRY: 

253 available = ', '.join(CONFIGURATION_REGISTRY.keys()) 

254 raise ValueError(f"Unknown configuration: '{configuration_name}'. Available: {available}") 

255 return CONFIGURATION_REGISTRY[key]()