Refactor create_<device> functions to raise errors to application level

[kabanaty]

Issue:

Currently, device functions like create_magnet() or create_wire() catch ValidationError internally and print the error. This prevents upstream scripts or applications from handling these errors appropriately. To improve modularity and error handling, we should refactor these functions to either raise a custom exception or return a structured error object while preserving the try/except logic in the application-level code.

• create_<device> functions no longer catch ValidationError internally
• Application-level code is responsible for handling failures
• Logging and error reporting should be moved to the caller
• Devices that fail validation should return None or raise a custom exception, based on context

Refactored code example:

reader.py

def create_bpm(area: str = None, name: str = None) -> Union[None, BPM, BPMCollection]:
    device_data = _device_data(area=area, device_type="bpms", name=name)
    if not device_data:
        return None
    if name:
        device_data.update({"name": name})
        return BPM(**device_data)
    else:
        return BPMCollection(**device_data)

tmit_loss.py

def create_bpms(self, bpms_elements):
        bpm_obj_dict = {}

        # Iterate through Dataframe of Elements and Areas
        for index, row in bpms_elements.iterrows():
            element = row["Element"]
            area = row["Area"]

            # Create an lcls-tools BPM object and append to dictionary
            # Key: Element Name
            # Value: lcls-tools BPM object
            try:
                bpm_obj_dict[element] = create_bpm(name=element, area=area)
            except ValidationError as e:
                logging.warning("Validation error creating %s. Continuing...", element)
                for err in e.errors():
                    loc = " -> ".join(str(i) for i in err["loc"])
                    logging.warning("%s: %s (%s)", loc, err["msg"], err["type"])
        
        return bpm_obj_dict

Log example:

2025-06-17 13:39:06,652 - INFO - Reserving BSA Buffer...
2025-06-17 13:39:11,932 - INFO - Reserved BSA Buffer 23
2025-06-17 13:39:11,933 - INFO - Creating device dictionary...
2025-06-17 13:39:22,740 - WARNING - Validation error creating BPMBP13. Continuing...
2025-06-17 13:39:22,740 - WARNING - controls_information -> PVs -> x: Field required (missing)
2025-06-17 13:39:22,740 - WARNING - controls_information -> PVs -> y: Field required (missing)

[robnagler]

    if not device_data:
        return None

Seems like it would be a good idea to raise here or even inside _device_data. This is what slicops.device_db does. What would be the use case of calling create_bpm with an unknown bpm?

logging.warning("Validation error creating %s. Continuing...", element)
``

Also seems like continuing is a mistake here. The caller of create_bpms will get a valid result (empty dict), which does not seem useful.

[kabanaty]

I'd like to handle ValidationErrors from device creation more effectively. Each wire scan instantiates around 150 devices, and when create_device() catches the error internally, it prevents the main application from responding appropriately.

Given the instability of the control system, it's common for devices (e.g., BPMs) to be temporarily disconnected. I'd like to avoid a full scan failure just because one device can't be created. Returning None works in that it keeps the script running, but forces repetitive if device is not None checks.

I'm looking for a more graceful failure-handling approach, ideally something that allows the scan to proceed while still surfacing which devices failed, without killing the process or losing context. Open to suggestions.

[robnagler]

I think I understand now. I do suggest a simplification:

def create_bpms(self, bpms_elements):
    rv = {}
    for r in bpm_elements.itertuples():
        try:
             rv[r.Element] = create_bpm(element=r.Element, area=r.Area)
        except ValidationError as e:
            logging.warning("create_bpms element=%s error=%s", r.Element,  e)
    return rv

Using itertuples fits the problem. I'm assuming bpms_elements is a DataFrame. Also, DataFrame returns a named tuple so you can remove syntax which clarifies meaning.

ValidationError implements __str__ so I would rely on that. This simplifies the error handling code so it doesn't dominate the function. If there is a standard in lcls_tools to do what you are doing, I would put all formatting in a function. The way it is written now is not robust to changes in the interface of ValidationError, that is, you'll have to change this error printing code in too many places should the interface change.

Another thing about having a single WARNING line is that it makes it easier to parse the logs with tools. You can see all error details with a single WARNING log line. When I see multiple WARNING lines, I think they are unrelated warnings.

I recommend not using temporary variables where possible.

I still wonder if you don't want to report an error to the user when some of the BPMs weren't available, but that could be done by checking for a non-existence of a bpm in the results of create_bpms. However, it would be nice to return the actual error so the user gets some context. Right now, the user would have check the log to see the detailed error. It depends on the context where create_bpms is used, though.

In the case of a loop iterator, a single letter variable is good enough. Again, by removing repetitive text in a line, it's easier to understand the salient parts.

My $.02 of course. :)

[kabanaty]

Thanks for your detailed suggestions. In scope of this issue, to catch the ValidationError in your suggested code block above, we would still need to remove the catch from reader.py, correct?

[robnagler]

Yes, and _device_data is problematic. Here's my suggestion for create_bpm:

def create_bpm(
    area: str,
    name: str = None
) -> Union[BPM, BPMCollection]:
    rv = _device_data(area=area, device_type="bpms", name=name)
    if not name:
    	return BPMCollection(**rv)
    rv.setdefault("name", name)
    return BPM(**rv)

By using early returns, the code is less indented, and easier to read. I changed area's type annotation to not be allowed to be None. setdefault eliminates the need for the comment. If the yaml is updated to include "name", then this code will do the right thing: use the YAML value.

And my suggestion for _device_data:

def _device_data(
    area: str,
    device_type: str = None,
    name: str = None,
) -> Union[None, Dict[str, Any]]:
    with open(_find_yaml_file(area=area), "rt") as device_file:
        rv = yaml.safe_load(device_file)
        if not device_type:
            return rv
        if (rv := rv.get(device_type)) is None:
            raise AssertionError(f"device_type={device_type} not found area={area}")
        if not name:
            return {device_type: rv}
        if (rv := rv.get(name)) is None:
            raise AssertionErro(f"device_name={name} not found device_type={device_type} area={area}")
        return rv

Again early exits make the code shorter, more readable. I used AssertionError (fault in the code) here. It's not clear in what context there wouldn't be a valid device_type or name.