Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add validation of a Device/Channel/EOM from another Device/Channel/EOM, and comparison of RegisterLayouts #558

Closed
wants to merge 2 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
201 changes: 201 additions & 0 deletions pulser-core/pulser/channels/comparison_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,201 @@
# Copyright 2020 Pulser Development Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Defines functions for comparison of Channels."""
from __future__ import annotations

import numbers
from collections.abc import Callable
from dataclasses import asdict
from operator import gt, lt, ne
from typing import TYPE_CHECKING, Any

import numpy as np

from pulser.channels.eom import RydbergEOM

if TYPE_CHECKING:
from pulser.channels.base_channel import Channel
from pulser.channels.eom import BaseEOM


def _compare_with_None(
comparison_ops: dict[str, Callable],
leftvalues: dict[str, float | None],
rightvalues: dict[str, float | None],
) -> dict[str, Any]:
"""Compare dict of values using a dict of comparison operators.

If comparison operator returns a boolean, the value returned is:
- True if right value is None and left value is defined.
- False if left value is None and right value is defined.
- None if left and right values are None.
Implemented for lt, gt, min, max.

Args:
comparison_ops: Associate keys to compare with comparison operator
leftvalues: Dict of values on the left of the comparison operator
rightvalues: Dict of values on the right of the comparison operator

Returns:
Dictionary having the keys of the comparison operator, associating
None if both values are None for this key, and the result of the
comparison otherwise.
"""
# Error if some keys of comparison operators are not dict of left or right
if not (
comparison_ops.keys() <= leftvalues.keys()
and comparison_ops.keys() <= rightvalues.keys()
):
raise ValueError(
"Keys in comparison_ops should be in left values and right values."
)
# Compare using +inf and -inf to replace None values
return {
key: comparison_op(
*(
value
or (
float("+inf")
if comparison_op in [lt, min]
else float("-inf")
)
for value in (leftvalues[key], rightvalues[key])
)
)
if (leftvalues[key], rightvalues[key]) != (None, None)
else None
for (key, comparison_op) in comparison_ops.items()
}


def _validate_obj_from_best(
obj_dict: dict, best_obj_dict: dict, comparison_ops: dict
) -> bool:
"""Validates an object by comparing it with a better one.

Attributes:
obj_dict: Dict of attributes and values of the object to compare.
best_obj_dict: Dict of attributes and values of the best object.
comparison_ops: Dict of attributes and comparison operators to
use to compare the object and the best object.

Returns:
True if the comparison works, raises a ValueError otherwise.
"""
# If the two values are almost equal then there is no need to compare them
comparison_ops_keys = list(comparison_ops.keys())
for key in comparison_ops_keys:
if (
obj_dict[key] is not None
and best_obj_dict[key] is not None
and isinstance(obj_dict[key], numbers.Number)
and isinstance(best_obj_dict[key], numbers.Number)
and np.isclose(obj_dict[key], best_obj_dict[key], rtol=1e-14)
):
comparison_ops.pop(key)
is_wrong_effective_ch = _compare_with_None(
comparison_ops, obj_dict, best_obj_dict
)
# Validates if no True in the dictionary of the comparisons
if not (True in is_wrong_effective_ch.values()):
return True
is_wrong_effective_index = list(is_wrong_effective_ch.values()).index(True)
is_wrong_key = list(is_wrong_effective_ch.keys())[is_wrong_effective_index]
raise ValueError(
f"{is_wrong_key} cannot be"
+ (
" below "
if comparison_ops[is_wrong_key] == lt
else (
" above "
if comparison_ops[is_wrong_key] == gt
else " different than "
)
)
a-corni marked this conversation as resolved.
Show resolved Hide resolved
+ f"{best_obj_dict[is_wrong_key]}."
)


def validate_channel_from_best(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you think of turning this into a method? We could have something like Channel.is_replaceable_by(better_channel: Channel).

The same would go for the EOM, probably.

Copy link
Collaborator Author

@a-corni a-corni Jul 19, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes that was also an idea I had, that seemed quite pleasant to me. Also pleasant because it can be redefined/ extended for LaserChannel and DMM.

channel: Channel, best_channel: Channel
) -> bool:
"""Checks that a channel can be realized from another one.

Attributes:
channel: The channel to check.
best_channel: The channel that should have better properties.
"""
if type(channel) != type(best_channel):
raise ValueError(
a-corni marked this conversation as resolved.
Show resolved Hide resolved
"Channels do not have the same types, "
f"{type(channel)} and {type(best_channel)}"
)
if channel.eom_config:
if best_channel.eom_config:
validate_eom_from_best(channel.eom_config, best_channel.eom_config)
else:
raise ValueError(
"eom_config cannot be defined in channel as the best_channel"
" does not have one."
)
best_ch_att = asdict(best_channel)
ch_att = asdict(channel)

# Error if attributes in channel and best_channel compare to True
comparison_ops = {
"addressing": ne,
"max_abs_detuning": gt,
"max_amp": gt,
"min_retarget_interval": lt,
"fixed_retarget_t": lt,
"max_targets": gt,
"clock_period": lt,
"min_duration": lt,
"max_duration": gt,
"mod_bandwidth": gt,
Comment on lines +157 to +166
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm afraid things are not this simple for the quantities that affect the sequence when it is being reconstructed. For example, mod_bandwidth influences things like the wait times, so if you are rebuilding the sequence with a device that has a different modulation bandwidth (even if higher), it may turn out different.

I suggest you take a look at how Sequence.switch_device() works (particularly when strict=True) - where, by the way, I see these functions fitting very naturally

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I now see the mistake with my approach: it completely forgets the original goal of this validation, which is to check whether sequence from the the cloud can be performed with the available_devices. As a reminder:

        if emulator is None:
            available_devices = self.fetch_available_devices()
            # TODO: Could be better to check if the devices are
            # compatible, even if not exactly equal
            if sequence.device not in available_devices.values():
                raise ValueError(
                    "The device used in the sequence does not match any "
                    "of the devices currently available through the remote "
                    "connection."
                )

I only wanted to check if the values put in Device/Channel/EOM matched with Device/Channel/EOM, but that is not the topic indeed. The checks should garantee that the shape of the curve will be the same. So indeed the mod bandwidth should be the same. The clock_period should be the same as well (or a divider of the clock period of the best channel).

Here is a list of things that are different in Sequence.switch_device:

  • takes into account whether the channels are reusable or not when matching channels of the devices.
  • basis should be equal.
  • if strict, the eom configs have to be exactly the same. To me, that's right for all the attributes except for controlled_beams: we can make a list of the controlled beams config ( RED, BLUE, (BLUE, RED) ) used in each eom block of the sequence, and check if they are among the controlled beams of the new channel (if the current device has controlled beams (RED, BLUE, (BLUE, RED)) but only the RED is controlled in the EOM blocks of the Sequence, a Channel with controlled beams (RED) would work strictly).
  • if strict, fixed_retarget_t should match exactly. Given its definition "Time taken to change the target (in ns)." that's correct.
  • if strict, min_retarget_interval should match exactly. As this is for testing only, and as its definition is "Minimum time required between the ends of two target instructions (in ns).", I think that can be changed into new_channel.min_retarget_interval <= old_channel.min_retarget_interval.
  • tests all the other properties of device by implementing the sequence with it.
  • Misses the comparison of layout/register of the Sequence with the calibrated_layouts of the new device (if there is a matching layout, the register of the sequence should have this new layout)

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To add another remark, the difference between Sequence.switch_device and the validation functions is that in Sequence.switch_device, the default operator is ne, whereas in the validation functions they are None, meaning that no comparisons can be done. If I were to keep these validation functions, I would build the dictionary of comparison operators by first assigning to all the attributes of device/channel/eom ne, and then specify a different operator for some of them.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • if strict, the eom configs have to be exactly the same. To me, that's right for all the attributes except for controlled_beams: we can make a list of the controlled beams config ( RED, BLUE, (BLUE, RED) ) used in each eom block of the sequence, and check if they are among the controlled beams of the new channel (if the current device has controlled beams (RED, BLUE, (BLUE, RED)) but only the RED is controlled in the EOM blocks of the Sequence, a Channel with controlled beams (RED) would work strictly).

I agree with this, the problem is checking it... The decision of which beams are controlled depends on the value of detuning_on, which may even be parametrized, so we can't always know in advance. And even when we can, we currently don't store the controlled beams in in each block so we have to reverse engineer it, which is a pain...

}
return _validate_obj_from_best(ch_att, best_ch_att, comparison_ops)


def validate_eom_from_best(eom: BaseEOM, best_eom: BaseEOM) -> bool:
"""Checks that an EOM config can be realized from another one.

Attributes:
a-corni marked this conversation as resolved.
Show resolved Hide resolved
eom: The EOM config to check.
best_eom: The EOM config that should have better properties.
"""
best_eom_att = asdict(best_eom)
eom_att = asdict(eom)

# Error if attributes in eom and best_eom compare to True
comparison_ops = {"mod_bandwidth": gt}
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should add the custom_buffer_time

if isinstance(eom, RydbergEOM):
if isinstance(best_eom, RydbergEOM):
comparison_ops.update(
{
"limiting_beam": ne,
"max_limiting_amp": gt,
a-corni marked this conversation as resolved.
Show resolved Hide resolved
"intermediate_detuning": ne,
"controlled_beams": gt,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should add multiple_beam_control

}
)
best_eom_att["controlled_beams"] = set(
best_eom_att["controlled_beams"]
)
eom_att["controlled_beams"] = set(eom_att["controlled_beams"])
else:
raise ValueError(
"EOM config is RydbergEOM whereas best EOM config is not."
)
return _validate_obj_from_best(eom_att, best_eom_att, comparison_ops)
119 changes: 119 additions & 0 deletions pulser-core/pulser/devices/comparison_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
# Copyright 2020 Pulser Development Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Defines functions for comparison of Channels."""
from __future__ import annotations

import itertools
from dataclasses import asdict
from operator import gt, lt, ne
from typing import TYPE_CHECKING, cast

from pulser.channels.comparison_tools import (
_validate_obj_from_best,
validate_channel_from_best,
)
from pulser.devices import Device, VirtualDevice

if TYPE_CHECKING:
from pulser.devices._device_datacls import BaseDevice


def _exist_good_configuration(
possible_configurations: dict[str, list[str]]
) -> bool:
# If one value is an empty list then no configuration can work
if any(
[
len(possible_values) == 0
for possible_values in possible_configurations.values()
]
):
return False
print(list(possible_configurations.values()))
print(itertools.product(*list(possible_configurations.values())))
for config in itertools.product(*list(possible_configurations.values())):
# True if each value in the list is different
print(config)
set(config)
a-corni marked this conversation as resolved.
Show resolved Hide resolved
if len(set(config)) == len(config):
return True
return False


def validate_device_from_best(
device: BaseDevice, best_device: BaseDevice
) -> bool:
"""Checks that a device can be realized from another one.

Attributes:
device: The device to check.
best_device: The device that should have better properties.
"""
if type(device) != type(best_device):
raise ValueError(
"Devices do not have the same types, "
f"{type(device)} and {type(best_device)}"
)
equivalent_channels: dict[str, list[str]] = {
ch_name: [] for ch_name in device.channels.keys()
}
for ch_name, channel in device.channels.items():
for best_ch_name, best_channel in best_device.channels.items():
try:
validate_channel_from_best(channel, best_channel)
except ValueError:
continue
equivalent_channels[ch_name].append(best_ch_name)
if not _exist_good_configuration(equivalent_channels):
raise ValueError(
"No configuration could be found where each channel of the device"
" could be realized with one channel of the best device."
)

if isinstance(device, Device) and device.calibrated_register_layouts:
equivalent_layouts: dict[str, list[str]] = {
str(id): []
for id in range(len(device.calibrated_register_layouts))
}
for id, layout in enumerate(device.calibrated_register_layouts):
for id_best, best_layout in enumerate(
cast(Device, best_device).calibrated_register_layouts
):
if best_layout > layout:
equivalent_layouts[str(id)].append(str(id_best))
if not _exist_good_configuration(equivalent_layouts):
raise ValueError(
"No configuration could be found where each calibrated layouts"
" of the device could be realized with a calibrated layout of"
" the best device."
)

best_device_att = asdict(best_device)
device_att = asdict(device)

# Error if attributes in device and best_device compare to True
comparison_ops = {
"dimensions": gt,
"rydberg_level": ne,
"min_atom_distance": lt,
"max_atom_num": gt,
"max_radial_distance": gt,
"interaction_coeff_xy": ne,
"supports_slm_mask": gt,
"max_layout_filling": gt,
}
if isinstance(device, VirtualDevice):
comparison_ops["reusable_channels"] = gt

return _validate_obj_from_best(device_att, best_device_att, comparison_ops)
28 changes: 28 additions & 0 deletions pulser-core/pulser/register/register_layout.py
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,34 @@ def __eq__(self, other: Any) -> bool:
return False
return self._safe_hash() == other._safe_hash()

def __gt__(self, other: RegisterLayout) -> bool:
if not isinstance(other, RegisterLayout):
raise TypeError("Right operand should be of type RegisterLayout.")
return (
set(tuple(self._coords[i]) for i in range(self.number_of_traps))
> set(
tuple(other._coords[i]) for i in range(other.number_of_traps)
)
and self.dimensionality == other.dimensionality
)

def __ge__(self, other: Any) -> bool:
return self.__eq__(other) or self.__gt__(other)

def __lt__(self, other: RegisterLayout) -> bool:
if not isinstance(other, RegisterLayout):
raise TypeError("Right operand should be of type RegisterLayout.")
return (
set(tuple(self._coords[i]) for i in range(self.number_of_traps))
< set(
tuple(other._coords[i]) for i in range(other.number_of_traps)
)
and self.dimensionality == other.dimensionality
)

def __le__(self, other: Any) -> bool:
return self.__eq__(other) or self.__lt__(other)

def __hash__(self) -> int:
return hash(self._safe_hash())

Expand Down
Loading