Refactor element attachment with holder-specific dispatch

pyaml/apidoc/gen_api.py

@@ -1,7 +1,6 @@
 import importlib
 import inspect
 import sys
-from pathlib import Path
 
 # List of PyAML modules to include in the API reference
 # for i in `find . | grep ".py" | grep -v "cache" | grep -v "__init__"`;

pyaml/bpm/bpm.py

@@ -151,6 +151,9 @@ def attach(
         obj._peer = peer
         return obj
 
+    def _fill_device(self, holder) -> None:
+        holder._fill_bpm(self)
+
     def get_pos_devices(self) -> list[str | None]:
         """
         Get device handles used for position reading

pyaml/common/holders/element_holder.py

@@ -7,11 +7,7 @@
 from abc import ABCMeta, abstractmethod
 from typing import TYPE_CHECKING
 
-from ...arrays.bpm_array import BPMArray
-from ...arrays.cfm_magnet_array import CombinedFunctionMagnetArray
 from ...arrays.element_array import ElementArray
-from ...arrays.magnet_array import MagnetArray
-from ...arrays.serialized_magnet_array import SerializedMagnetsArray
 from ...bpm.bpm import BPM
 from ...diagnostics.tune_monitor import BetatronTuneMonitor
 from ...magnet.cfm_magnet import CombinedFunctionMagnet
@@ -37,14 +33,17 @@
 
 if TYPE_CHECKING:
     from ...accelerator import Accelerator
+    from ...configuration.unbound_element import UnboundElement
     from ...tuning_tools.bba import BBA
     from ...tuning_tools.chromaticity import Chromaticity
     from ...tuning_tools.chromaticity_response_matrix import ChromaticityResponseMatrix
     from ...tuning_tools.dispersion import Dispersion
+    from ...tuning_tools.measurement_tool import MeasurementTool
     from ...tuning_tools.orbit import Orbit
     from ...tuning_tools.orbit_response_matrix import OrbitResponseMatrix
     from ...tuning_tools.tune import Tune
     from ...tuning_tools.tune_response_matrix import TuneResponseMatrix
+    from ...tuning_tools.tuning_tool import TuningTool
 
 
 class ElementHolder(metaclass=ABCMeta):
@@ -145,7 +144,40 @@ def post_init(self):
             e.post_init()
 
     def fill_device(self, elements: list[Element]):
-        raise PyAMLException("ElementHolder.fill_device() is not subclassed")
+        for element in elements:
+            element._fill_device(self)
+
+    @abstractmethod
+    def _fill_magnet(self, magnet: Magnet) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_combined_function_magnet(self, magnet: CombinedFunctionMagnet) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_bpm(self, bpm: BPM) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_rf_plant(self, rf_plant: RFPlant) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_betatron_tune_monitor(self, monitor: BetatronTuneMonitor) -> None:
+        pass
+
+    @abstractmethod
+    def _fill_tool(self, tool: "TuningTool | MeasurementTool") -> None:
+        pass
+
+    @abstractmethod
+    def _fill_unbound_element(self, element: "UnboundElement") -> None:
+        pass
 
     # Aggregators
 

pyaml/configuration/unbound_element.py

@@ -37,6 +37,9 @@ def __repr__(self):
             self._module_name,
         )
 
+    def _fill_device(self, holder) -> None:
+        holder._fill_unbound_element(self)
+
     def instantiate(self, holder) -> Element:
         """
         Instantiate the element represented by this UnboundElement.

pyaml/control/controlsystem.py

@@ -5,10 +5,8 @@
 from ..bpm.bpm import BPM
 from ..common.abstract import RWMapper
 from ..common.abstract_aggregator import ScalarAggregator
-from ..common.element import Element
 from ..common.exception import PyAMLException
 from ..common.holders.element_holder import ElementHolder
-from ..configuration.factory import Factory
 from ..configuration.unbound_element import UnboundElement
 from ..control.abstract_impl import (
     CSScalarAggregator,
@@ -120,100 +118,77 @@ def create_bpm_aggregators(self, bpms: list[BPM]) -> list[ScalarAggregator | Non
             aggv.add_devices(devs[1])
         return [agg, aggh, aggv]
 
-    def fill_device(self, elements: list[Element]):
-        """
-        Fill device of this control system with Element
-        coming from the configuration file
-
-        Parameters
-        ----------
-        elements : list[Element]
-            List of elements coming from the configuration
-            file to attach to this control system
-        """
-        for e in elements:
-            if isinstance(e, Magnet):
-                dev = self.get_device_access(e.model.get_device_names()[0])
-                current = RWHardwareScalar(e.model, dev) if e.model.has_hardware() else None
-                strength = RWStrengthScalar(e.model, dev) if e.model.has_physics() else None
-                # Create a unique ref for this control system
-                m = e.attach(self, strength, current)
-                self.magnet.add(m)
-
-            elif isinstance(e, CombinedFunctionMagnet):
-                devs = self.get_devices_access(e.model.get_device_names())
-                currents = RWHardwareArray(e.model, devs)
-                strengths = RWStrengthArray(e.model, devs)
-                # Create unique refs the cfm and
-                # each of its function for this control system
-                ms = e.attach(self, strengths, currents)
-                self.combined_function_magnet.add(ms[0])
-                for m in ms[1:]:
-                    self.magnet.add(m)
-
-            elif isinstance(e, SerializedMagnets):
-                devs = self.get_devices_access(e.model.get_device_names())
-                currents = []
-                strengths = []
-                # Create unique refs the series and each of its function for this
-                # control system
-                for i in range(e.get_nb_magnets()):
-                    current = RWHardwareScalar(e.model.get_sub_model(i), devs[i]) if e.model.has_hardware() else None
-                    strength = RWStrengthScalar(e.model.get_sub_model(i), devs[i]) if e.model.has_physics() else None
-                    currents.append(current)
-                    strengths.append(strength)
-                ms = e.attach(self, strengths, currents)
-                self.serialized_magnet.add(ms[0])
-                for m in ms[1:]:
-                    self.magnet.add(m)
-
-            elif isinstance(e, BPM):
-                pos_devs = self.get_devices_access(e.get_pos_devices())
-                tilt_devs = self.get_devices_access([e.get_tilt_device()])
-                offset_devs = self.get_devices_access(e.get_offset_devices())
-                positions = RBpmArray(pos_devs[0], pos_devs[1])
-                tilt = RWBpmTiltScalar(tilt_devs[0])
-                offsets = RWBpmOffsetArray(offset_devs[0], offset_devs[1])
-                e = e.attach(self, positions, offsets, tilt)
-                self.bpm.add(e)
-
-            elif isinstance(e, RFPlant):
-                attachedTrans: list[RFTransmitter] = []
-                if e.transmitters:
-                    for t in e.transmitters:
-                        vDev = self.get_device_access(t.voltage_name)
-                        pDev = self.get_device_access(t.phase_name)
-                        voltage = RWRFVoltageScalar(t, vDev)
-                        phase = RWRFPhaseScalar(t, pDev)
-                        nt = t.attach(self, voltage, phase)
-                        self.rf.transmitter.add(nt)
-                        attachedTrans.append(nt)
-
-                fDev = self.get_device_access(e.masterclock)
-                frequency = RWRFFrequencyScalar(e, fDev)
-                voltage = RWTotalVoltage(attachedTrans) if e.transmitters else None
-                ne = e.attach(self, frequency, voltage)
-                self.rf.add(ne)
-
-            elif isinstance(e, BetatronTuneMonitor):
-                # Built in tune monitor
-                tuneDevs = self.get_devices_access([e.tune_h, e.tune_v])
-                betatron_tune = RBetatronTuneArray(e, tuneDevs)
-                e = e.attach(self, betatron_tune)
-                self.add_betatron_tune_monitor(e)
-
-            elif isinstance(e, TuningTool) | isinstance(e, MeasurementTool):
-                self.add_tool(e.attach(self))
-
-            elif isinstance(e, UnboundElement):
-                if self.name() in e._control_modes:
-                    ne = e.instantiate(self)
-
-                    if isinstance(ne, ABetatronTuneMonitor):
-                        self.add_betatron_tune_monitor(ne)
-                    else:
-                        # Default to standard Element
-                        self.add_element(ne)
+    def _fill_magnet(self, magnet: Magnet) -> None:
+        device = self.get_device_access(magnet.model.get_device_names()[0])
+        current = RWHardwareScalar(magnet.model, device) if magnet.model.has_hardware() else None
+        strength = RWStrengthScalar(magnet.model, device) if magnet.model.has_physics() else None
+        self.magnet.add(magnet.attach(self, strength, current))
+
+    def _fill_combined_function_magnet(self, magnet: CombinedFunctionMagnet) -> None:
+        devices = self.get_devices_access(magnet.model.get_device_names())
+        currents = RWHardwareArray(magnet.model, devices)
+        strengths = RWStrengthArray(magnet.model, devices)
+        attached_magnets = magnet.attach(self, strengths, currents)
+        self.combined_function_magnet.add(attached_magnets[0])
+        for virtual_magnet in attached_magnets[1:]:
+            self.magnet.add(virtual_magnet)
+
+    def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None:
+        devices = self.get_devices_access(magnets.model.get_device_names())
+        currents = []
+        strengths = []
+        for index in range(magnets.get_nb_magnets()):
+            currents.append(
+                RWHardwareScalar(magnets.model.get_sub_model(index), devices[index]) if magnets.model.has_hardware() else None
+            )
+            strengths.append(
+                RWStrengthScalar(magnets.model.get_sub_model(index), devices[index]) if magnets.model.has_physics() else None
+            )
+        attached_magnets = magnets.attach(self, strengths, currents)
+        self.serialized_magnet.add(attached_magnets[0])
+        for magnet in attached_magnets[1:]:
+            self.magnet.add(magnet)
+
+    def _fill_bpm(self, bpm: BPM) -> None:
+        position_devices = self.get_devices_access(bpm.get_pos_devices())
+        tilt_devices = self.get_devices_access([bpm.get_tilt_device()])
+        offset_devices = self.get_devices_access(bpm.get_offset_devices())
+        positions = RBpmArray(position_devices[0], position_devices[1])
+        tilt = RWBpmTiltScalar(tilt_devices[0])
+        offsets = RWBpmOffsetArray(offset_devices[0], offset_devices[1])
+        self.bpm.add(bpm.attach(self, positions, offsets, tilt))
+
+    def _fill_rf_plant(self, rf_plant: RFPlant) -> None:
+        attached_transmitters: list[RFTransmitter] = []
+        if rf_plant.transmitters:
+            for transmitter in rf_plant.transmitters:
+                voltage_device = self.get_device_access(transmitter.voltage_name)
+                phase_device = self.get_device_access(transmitter.phase_name)
+                voltage = RWRFVoltageScalar(transmitter, voltage_device)
+                phase = RWRFPhaseScalar(transmitter, phase_device)
+                attached_transmitter = transmitter.attach(self, voltage, phase)
+                self.rf.transmitter.add(attached_transmitter)
+                attached_transmitters.append(attached_transmitter)
+        frequency_device = self.get_device_access(rf_plant.masterclock)
+        frequency = RWRFFrequencyScalar(rf_plant, frequency_device)
+        voltage = RWTotalVoltage(attached_transmitters) if rf_plant.transmitters else None
+        self.rf.add(rf_plant.attach(self, frequency, voltage))
+
+    def _fill_betatron_tune_monitor(self, monitor: BetatronTuneMonitor) -> None:
+        devices = self.get_devices_access([monitor.tune_h, monitor.tune_v])
+        self.add_betatron_tune_monitor(monitor.attach(self, RBetatronTuneArray(monitor, devices)))
+
+    def _fill_tool(self, tool: TuningTool | MeasurementTool) -> None:
+        self.add_tool(tool.attach(self))
+
+    def _fill_unbound_element(self, element: UnboundElement) -> None:
+        if self.name() not in element._control_modes:
+            return
+        attached_element = element.instantiate(self)
+        if isinstance(attached_element, ABetatronTuneMonitor):
+            self.add_betatron_tune_monitor(attached_element)
+        else:
+            self.add_element(attached_element)
 
 
 class ControlSystemAdapter(ControlSystem):

pyaml/diagnostics/tune_monitor.py

@@ -53,6 +53,9 @@ def __init__(
     def set_harmonic(self, h: int):
         self._h = float(h)
 
+    def _fill_device(self, holder) -> None:
+        holder._fill_betatron_tune_monitor(self)
+
     @property
     def tune_h(self) -> str | None:
         return self._tune_h

pyaml/lattice/abstract_impl.py

@@ -5,10 +5,7 @@
 
 from ..common import abstract
 from ..common.abstract_aggregator import ScalarAggregator
-from ..common.exception import PyAMLException
 from ..magnet.model import MagnetModel
-from ..rf.rf_plant import RFPlant
-from ..rf.rf_transmitter import RFTransmitter
 from .polynom_info import PolynomInfo
 
 # TODO handle serialized magnets for magnet array

pyaml/lattice/lattice_elements_linker.py

@@ -1,5 +1,4 @@
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
 from typing import Iterable
 
 import at