Set line and column when an error is founded in a configuration file.

.github/workflows/tests.yml

@@ -32,6 +32,7 @@ jobs:
         pip install accelerator-toolbox
         pip install pyaml
         pip install flake8 pytest
+        pip install ruamel.yaml
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
     - name: Lint with flake8
       run: |

pyaml/__init__.py

@@ -15,8 +15,9 @@
 import logging.config
 import os
 from pyaml.exception import PyAMLException
+from pyaml.configuration.config_exception import PyAMLConfigException
 
-__all__ = [__version__, PyAMLException]
+__all__ = [__version__, PyAMLException, PyAMLConfigException]
 
 
 config_file = os.getenv("PYAML_LOG_CONFIG", "pyaml_logging.conf")

pyaml/configuration/__init__.py

@@ -3,11 +3,12 @@
 """
 
 from pathlib import Path
+from typing import Union
 
 ROOT = {"path": Path.cwd().resolve()}
 
 
-def set_root_folder(path: str | Path):
+def set_root_folder(path: Union[str, Path]):
     """
     Set the root path for configuration files.
     """

pyaml/configuration/config_exception.py

@@ -9,24 +9,28 @@ class PyAMLConfigException(PyAMLException):
         message -- explanation of the error
     """
 
-    def __init__(self, config_key = None, parent_exception:Union["PyAMLConfigException", "PyAMLException"] = None):
+    def __init__(self, config_key = None, parent_exception:Exception = None):
         self.parent_keys = []
         self.config_key = config_key
         self.parent_exception = parent_exception
         message = "An exception occurred while building object."
         if parent_exception is not None:
-            if isinstance(self.parent_exception, PyAMLConfigException) and parent_exception.config_key is not None:
+            if isinstance(parent_exception, PyAMLConfigException) and parent_exception.config_key is not None:
                 self.parent_keys.append(parent_exception.config_key)
                 self.parent_keys.extend(parent_exception.parent_keys)
                 if config_key is not None:
                     message = f"An exception occurred while building key '{config_key}.{parent_exception.get_keys()}': {parent_exception.get_original_message()}"
                 else:
                     message = f"An exception occurred while building object in '{parent_exception.get_keys()}': {parent_exception.get_original_message()}"
             else:
+                if isinstance(parent_exception, PyAMLException):
+                    parent_message = parent_exception.message
+                else:
+                    parent_message = str(parent_exception)
                 if config_key is not None:
-                    message = f"An exception occurred while building key '{config_key}': {parent_exception.message}"
+                    message = f"An exception occurred while building key '{config_key}': {parent_message}"
                 else:
-                    message = f"An exception occurred while building object: {parent_exception.message}"
+                    message = f"An exception occurred while building object: {parent_message}"
         super().__init__(message)
 
     def get_keys(self) -> str:
@@ -42,7 +46,9 @@ def get_original_message(self):
         if self.parent_exception is not None:
             if isinstance(self.parent_exception, PyAMLConfigException):
                 return self.parent_exception.get_original_message()
-            else:
+            elif isinstance(self.parent_exception, PyAMLException):
                 return self.parent_exception.message
+            else:
+                return str(self.parent_exception)
         else:
             return self.message

pyaml/configuration/factory.py

@@ -43,17 +43,22 @@ def remove_strategy(self, strategy: BuildStrategy):
 
     def build_object(self, d:dict):
         """Build an object from the dict"""
+        location = d.pop('__location__', None)
+        location_str = ""
+        if location:
+            line, col = location
+            location_str = f" at line {line}, column {col}."
 
         if not isinstance(d,dict):
-            raise PyAMLException("Unexpected object " + str(d))
+            raise PyAMLException(f"Unexpected object {str(d)}{location_str}")
         if not "type" in d:
-            raise PyAMLException("No type specified for " + str(type(d)) + ":" + str(d))
+            raise PyAMLException(f"No type specified for {str(type(d))}:{str(d)}{location_str}")
         type_str = d.pop("type")
 
         try:
             module = importlib.import_module(type_str)
         except ModuleNotFoundError as ex:
-            raise PyAMLException(f"Module referenced in type cannot be founded: '{type_str}'") from ex
+            raise PyAMLException(f"Module referenced in type cannot be founded: '{type_str}'{location_str}") from ex
 
         # Try plugin strategies first
         for strategy in self._strategies:
@@ -63,18 +68,18 @@ def build_object(self, d:dict):
                     self.register_element(obj)
                     return obj
             except Exception as e:
-                raise PyAMLException("Custom strategy failed") from e
+                raise PyAMLException(f"Custom strategy failed{location_str}") from e
 
         # Default loading strategy
         # Get the config object
         config_cls = getattr(module, "ConfigModel", None)
         if config_cls is None:
-            raise ValueError(f"ConfigModel class '{type_str}.ConfigModel' not found")
+            raise PyAMLException(f"ConfigModel class '{type_str}.ConfigModel' not found{location_str}")
 
         # Get the class name
         cls_name = getattr(module, "PYAMLCLASS", None)
         if cls_name is None:
-            raise ValueError(f"PYAMLCLASS definition not found in '{type_str}'")
+            raise PyAMLException(f"PYAMLCLASS definition not found in '{type_str}'{location_str}")
 
         try:
 
@@ -84,16 +89,14 @@ def build_object(self, d:dict):
             # Construct and return the object
             elem_cls = getattr(module, cls_name, None)
             if elem_cls is None:
-                raise ValueError(
-                    f"Unknown element class '{type_str}.{cls_name}'"
-                )
+                raise PyAMLException(f"Unknown element class '{type_str}.{cls_name}'")
 
             obj = elem_cls(cfg)
             self.register_element(obj)
             return obj
 
         except Exception as e:
-            raise PyAMLConfigException(f'{type_str}.{cls_name}') from e
+            raise PyAMLException(f'{type_str}.{cls_name}') from e
 
 
     def depth_first_build(self, d):
@@ -110,7 +113,7 @@ def depth_first_build(self, d):
                   except PyAMLException as pyaml_ex:
                       raise PyAMLConfigException(f"[{index}]", pyaml_ex) from pyaml_ex
                   except Exception as ex:
-                      raise PyAMLConfigException(f"[{index}]") from ex
+                      raise PyAMLConfigException(f"[{index}]", ex) from ex
               else:
                   l.append(e)
           return l

pyaml/configuration/fileloader.py

@@ -1,12 +1,13 @@
 # PyAML config file loader
 import logging
-
-import yaml
 import json
 from typing import Union
-from . import get_root_folder
 from pathlib import Path
 
+from ruamel.yaml import YAML
+from ruamel.yaml.comments import CommentedMap, CommentedSeq
+
+from . import get_root_folder
 from .. import PyAMLException
 
 logger = logging.getLogger(__name__)
@@ -37,15 +38,15 @@ def hasToExpand(self,value):
         return isinstance(value,str) and any(value.endswith(suffix) for suffix in self.suffixes)
 
     # Recursively expand a dict
-    def expandDict(self,d:dict):            
+    def expand_dict(self,d:dict):
         for key, value in d.items():
             if self.hasToExpand(value):
                 d[key] = load(value)
             else:
                 self.expand(value)
 
     # Recursively expand a list
-    def expandList(self,l:list):            
+    def expand_list(self,l:list):
         for idx,value in enumerate(l):
             if self.hasToExpand(value):
                 l[idx] = load(value)
@@ -55,11 +56,29 @@ def expandList(self,l:list):
     # Recursively expand an object
     def expand(self,obj: Union[dict,list]):
         if isinstance(obj,dict):
-            self.expandDict(obj)
+            self.expand_dict(obj)
         elif isinstance(obj,list):
-            self.expandList(obj)
+            self.expand_list(obj)
+        self.tag_node_positions(obj)
         return obj
 
+    def tag_node_positions(self, obj):
+        """Recursively tag dicts with their __location__ (line, column)."""
+        if isinstance(obj, CommentedMap):
+            if hasattr(obj, 'lc'):
+                obj['__location__'] = (obj.lc.line + 1, obj.lc.col + 1)
+            for v in obj.values():
+                self.tag_node_positions(v)
+        elif isinstance(obj, dict):
+            for v in obj.values():
+                self.tag_node_positions(v)
+        elif isinstance(obj, CommentedSeq):
+            for i in obj:
+                self.tag_node_positions(i)
+        elif isinstance(obj, list):
+            for i in obj:
+                self.tag_node_positions(i)
+
     # Load a file
     def load(self) -> Union[dict,list]:
         raise Exception(str(self.path) + ": load() method not implemented")
@@ -73,11 +92,13 @@ def __init__(self, filename:str):
     def load(self) -> Union[dict,list]:
         logger.log(logging.DEBUG, f"Loading YAML file '{self.path}'")
         self.suffixes = [".yaml", ".yml"]
+        yaml = YAML()
+        yaml.preserve_quotes = True
         with open(self.path) as file:
             try:
-                return self.expand(yaml.load(file,yaml.CLoader)) # Use fast C loader
-            except yaml.YAMLError as e:
-                raise Exception(str(self.path) + ": " + str(e))
+                return self.expand(yaml.load(file)) # Use fast C loader
+            except Exception as e:
+                raise PyAMLException(str(self.path) + ": " + str(e)) from e
 
 # JSON loader
 class JSONLoader(Loader):
@@ -92,4 +113,4 @@ def  load(self) -> Union[dict,list]:
             try:
                 return self.expand(json.load(file))
             except json.JSONDecodeError as e:
-                raise Exception(str(self.path) + ": " + str(e))
+                raise PyAMLException(str(self.path) + ": " + str(e)) from e

pyproject.toml

@@ -44,7 +44,8 @@ dependencies = [
     "scipy>=1.7.3",
     "accelerator-toolbox>=0.6.1",
     "PyYAML>=6.0.2",
-    "pydantic>=2.11.7"
+    "pydantic>=2.11.7",
+    "ruamel.yaml"
 ]
 
 [project.optional-dependencies]

tests/config/bad_conf.yml

@@ -0,0 +1,25 @@
+type: pyaml.pyaml
+instruments:
+  - type: pyaml.instrument
+    name: sr
+    energy: 6e9
+    simulators:
+      - type: pyalkqln # Error here
+        lattice: sr/lattices/ebs.mat
+        name: design
+    data_folder: /data/store
+    arrays:
+      - type: pyaml.arrays.hcorrector
+        name: HCORR
+        elements:
+          - SH1A-C01-H
+          - SH1A-C02-H
+      - type: pyaml.arrays.vcorrector
+        name: VCORR
+        elements:
+          - SH1A-C01-V
+          - SH1A-C02-V
+    devices:
+      - sr/quadrupoles/QF1AC01.yaml
+      - sr/correctors/SH1AC01.yaml
+      - sr/correctors/SH1AC02.yaml

tests/test_factory_custom_build.py → tests/test_factory.py

@@ -1,4 +1,7 @@
+import pytest
+from pyaml import PyAMLConfigException
 from pyaml.configuration.factory import depthFirstBuild
+from pyaml.pyaml import PyAML, pyaml
 from tests.conftest import MockElement
 
 
@@ -23,3 +26,11 @@ def test_factory_with_custom_strategy():
     obj = depthFirstBuild(data)
     assert isinstance(obj, MockElement)
     assert obj.name == "custom_injected"
+
+
+def test_error_location():
+
+    with pytest.raises(PyAMLConfigException) as exc:
+        ml: PyAML = pyaml("tests/config/bad_conf.yml")
+
+    assert "at line 7, column 9" in str(exc.value)