solid_attenuator.ioc_lfe_at2l0_calc.at2l0.SystemGroup

class solid_attenuator.ioc_lfe_at2l0_calc.at2l0.SystemGroup(*args, **kwargs)

PV group for attenuator system-spanning information.

This system group implementation is specific to AT2L0.

SystemGroup pvproperties

Attribute	Suffix	Docs	Type	Notes	Alarm Group
active_config	ActiveConfiguration_RBV	Active configuration array	int	Read-only Inherited from SystemGroupBase Startup
active_config_bitmask	ActiveConfigurationBitmask_RBV	Active configuration represented as an integer	int	Read-only Inherited from SystemGroupBase
apply_config	ApplyConfiguration	Apply the calculated configuration.	ENUM (bo)	Inherited from SystemGroupBase Startup Put
best_config	BestConfiguration_RBV	Best configuration as an array (1 if inserted)	int	Read-only Inherited from SystemGroupBase
best_config_bitmask	BestConfigurationBitmask_RBV	Best configuration as an integer	int	Read-only Inherited from SystemGroupBase
best_config_error	BestConfigError_RBV	Calculated transmission error	float (ao)	Read-only Inherited from SystemGroupBase
calc_mode	CalcMode	Mode for selecting floor or ceiling transmission estimation	ENUM (bo)	Inherited from SystemGroupBase
calculated_transmission	T_CALC	Calculated transmission (all blades)	float (ao)	Read-only Inherited from SystemGroupBase
calculated_transmission_3omega	T_3OMEGA	Calculated 3omega transmission (all blades)	float	Read-only Inherited from SystemGroupBase
cancel_apply	Cancel	Stop trying to apply the configuration.	ENUM (bo)	Inherited from SystemGroupBase
desired_transmission	DesiredTransmission	Desired transmission value	float	Inherited from SystemGroupBase
energy_actual	ActualPhotonEnergy_RBV		float	Read-only Inherited from SystemGroupBase Startup
energy_custom	CustomPhotonEnergy		float	Inherited from SystemGroupBase
energy_source	EnergySource	Choose the source of photon energy	ENUM (bo)	Inherited from SystemGroupBase
filter_moving	FiltersMoving_RBV	Filter motion status as an array (1 if moving)	int	Read-only Inherited from SystemGroupBase
filter_moving_bitmask	FiltersMovingBitmask_RBV	Filter motion status as an integer	int	Read-only Inherited from SystemGroupBase
last_energy	LastPhotonEnergy_RBV	Energy that was used for the calculation.	float	Read-only Inherited from SystemGroupBase
last_mode	LastCalcMode_RBV	Last calculation mode	ENUM (bo)	Read-only Inherited from SystemGroupBase
last_transmission	LastTransmission_RBV	Last desired transmission value	float	Read-only Inherited from SystemGroupBase
mirror_in	MIRROR_IN	The inspection mirror is in	ENUM (bo)	Read-only Inherited from SystemGroupBase
moving	Moving_RBV	Moving to a new configuration.	ENUM (bo)	Read-only Inherited from SystemGroupBase
run	Run	Run calculation	ENUM (bo)	Inherited from SystemGroupBase Put
transmission_3omega_actual	Actual3OmegaTransmission_RBV		float	Read-only Inherited from SystemGroupBase
transmission_actual	ActualTransmission_RBV		float	Read-only Inherited from SystemGroupBase

Methods

calculate_stuck_transmission()	The effective normalized transmission of all stuck filters.
calculate_transmission()	Total transmission through all filter blades.
calculate_transmission_3omega()	Total 3rd harmonic transmission through all filter blades.
check_materials()	Ensure the materials specified are OK according to the order.
get_filters([stuck, inactive, normal])	Get filters matching the specified criteria, sorted by index.
group_read(instance)	Generic read called for channels without get defined
group_write(instance, value)	Generic write called for channels without put defined
motor_has_moved(blade_index, raw_state)	Callback indicating a motor has moved.
move_blade_step(state)	Caller is requesting to move blades in or out.
move_blades(*[, timeout_threshold])	Move to the calculated configuration.
run_calculation(energy, …)

Attributes

active_filters	A dictionary of all filters that are marked as inactive.
all_filter_materials	All filter materials in a list.
default_values
filters	All filters in the system.
first_filter	The first filter index in the system.
material_order	Material prioritization.
stuck_filters	A dictionary of all filters that are stuck in a particular state.
type_map
type_map_read_only

pvproperty methods

active_config.startup(self, instance, async_lib)

Source code: active_config.startup

@active_config.startup
async def active_config(self, instance, async_lib):
    motor_pvnames = self.parent.monitor_pvnames['motors']
    monitor_list = sum((pvlist for pvlist in motor_pvnames.values()),
                       [])
    all_status = {pv: False for pv in monitor_list}

    async def update_connection_status(pv, status):
        all_status[pv] = (status == 'connected')
        await util.alarm_if(instance, not all(all_status.values()),
                            AlarmStatus.LINK)

    async for event, context, data in monitor_pvs(*monitor_list,
                                                  async_lib=async_lib):
        if event == 'connection':
            await update_connection_status(context.name, data)
            continue

        pvname = context.pv.name
        if pvname not in motor_pvnames['get']:
            continue

        idx = self.parent.monitor_pvnames['motors']['get'].index(pvname)
        await self.motor_has_moved(self.parent.first_filter + idx,
                                   data.data[0])

apply_config.startup(self, instance, async_lib)

Source code: apply_config.startup

@apply_config.startup
async def apply_config(self, instance, async_lib):
    def put_thread():
        while True:
            pv, value = self._pv_put_queue.get()
            try:
                pv.write([value], wait=False)
            except Exception:
                self.log.exception('Failed to put value: %s=%s', pv, value)

    ctx = util.get_default_thread_context()

    self._set_pvs = ctx.get_pvs(
        *self.parent.monitor_pvnames['motors']['set'], timeout=None)

    self._pv_put_queue = self.async_lib.ThreadsafeQueue()
    self._put_thread = threading.Thread(target=put_thread, daemon=True)
    self._put_thread.start()

energy_actual.startup(self, instance, async_lib)

Update beam energy and calculated values.

Source code: energy_actual.startup

@energy_actual.startup
async def energy_actual(self, instance, async_lib):
    """Update beam energy and calculated values."""
    async def update_connection_status(status):
        await util.alarm_if(
            instance, status != 'connected', AlarmStatus.LINK
        )

    await update_connection_status('disconnected')
    pvname = self.parent.monitor_pvnames['ev']
    async for event, context, data in monitor_pvs(pvname,
                                                  async_lib=async_lib):
        if event == 'connection':
            self.log.info('%s %s', context, data)
            await update_connection_status(data)
            continue

        eV = data.data[0]
        self.log.debug('Photon energy changed: %s', eV)

        if instance.value != eV:
            delta = instance.value - eV
            if abs(delta) > 1000:
                self.log.info("Photon energy changed to %s eV.", eV)
            await instance.write(eV)

    return eV