import importlib
from io import IOBase
import urllib.parse
import urllib.request
import base64
import copy
from typing import Union
import random
import string
import numpy as np
from atomiq.components.primitives import Component
import artiq
from artiq.language.types import TFloat, TInt32, TBool
from artiq.experiment import kernel, rpc, portable
[docs]
def normalize_component_name(name) -> str:
return name.strip("$&@").replace("/","_")
[docs]
def get_module(path, relative_path=None):
module_name = '.'.join([relative_path, path]) if relative_path is not None else path
module = importlib.import_module(module_name)
return module
[docs]
def get_class_by_name(name, relative_path=None):
"""
Resolve class by name
:param name: (str) "%s.%s" % (module.name, class.name)
:return: (class)
"""
assert name is not None
module_path, class_name = name.rsplit('.', 1)
module__ = get_module(module_path, relative_path=relative_path)
class_ = getattr(module__, class_name)
return class_
[docs]
def rec_getattr(obj, attr):
"""Get object's attribute. May use dot notation.
>>> class C(object): pass
>>> a = C()
>>> a.b = C()
>>> a.b.c = 4
>>> rec_getattr(a, 'b.c')
4
"""
if '.' not in attr:
return getattr(obj, normalize_component_name(attr))
else:
L = attr.split('.')
return rec_getattr(getattr(obj, L[0]), '.'.join(L[1:]))
[docs]
def random_ascii_string(length):
return ''.join(random.choice(string.ascii_lowercase) for i in range(length))
[docs]
def file_from_url(url: str) -> IOBase:
"""
Load content from a file specified by a URL.
This can be every type of URL supported by pythons urllib
(e.g. http://, file://, etc ). Giving the basic auth credentials in the URL in the form
http://user:password@hostname:port/path is supported.
Returns:
file handle on the file
"""
parsed = urllib.parse.urlparse(url)
if parsed.username and parsed.password:
request = urllib.request.Request(parsed._replace(netloc=parsed.netloc.split('@')[1]).geturl())
base64string = base64.b64encode(bytes('%s:%s' % (parsed.username, parsed.password), 'ascii'))
request.add_header("Authorization", "Basic %s" % base64string.decode('utf-8'))
f_handle = urllib.request.urlopen(request)
else:
f_handle = urllib.request.urlopen(url)
return f_handle
[docs]
def component_dict(component: Union[str, tuple]) -> dict:
"""
Extract the required information (id, type) from the entries of a component list defined
in an AtomiqBlock/AtomiqExperiment
We allow entries in a components list of AtomiqExperiment/AtomiqBlock to be either just strings
specifying the id of the component or tuples of the form (id, type) to require a specific
type of the component with the given id. The type can also be a superclass of the specified
component.
Args:
component: entry of the component list
Returns:
dict: dictionary with keys "id" and and "type"
"""
# we allow for tuples ("comp_id", class, ...) or just a string "comp_id" in the components list
res = {"type": Component}
if type(component) is tuple:
res.update({"id": component[0], "type": component[1]})
else:
res.update({"id": component})
return res
[docs]
def component_data(component_dict: dict) -> Union[str, tuple]:
if "type" in component_dict and component_dict["type"] != Component:
return (component_dict["id"], component_dict["type"])
else:
return component_dict["id"]
[docs]
def block_dict(blockdef: Union[str, tuple]) -> dict:
"""
Extract the required information (class, map) from the entries of a block list defined
in an AtomiqBlock/AtomiqExperiment
We allow entries in a blocks list of AtomiqExperiment/AtomiqBlock to be either just strings
specifying the class of the AtomiqBlock to be imported or tuples of the form (class, ,map)
to specify a mapping of the components in the experiment to the components in the block.
Args:
blockdef: entry of the blocks list
Returns:
dict: dictionary with keys "class", "map", and "alias"
"""
res = {"map": {}, "alias": None}
if type(blockdef) is tuple:
res.update({"class": blockdef[0]})
if len(blockdef) > 1:
res.update({"map": blockdef[1]})
if len(blockdef) > 2:
res.update({"alias": blockdef[2]})
else:
res.update({"class": blockdef})
return res
[docs]
def replace_member(obj, target: str, replacement: str):
"""
Replace a member (target) of an object (obj) with a different member (replacement). This is useful to change
member functions of components at construction of the object depending on the configuration.
when running the legacy artiq compiler, every object has its own class and we can/have to modify the class.
Otherwise, we just alter the object itself.
"""
obj = obj.__class__ if "compiler" in dir(artiq) else obj
setattr(obj, target, getattr(obj, replacement))
[docs]
@kernel
def identity_float(x: TFloat) -> TFloat:
return x
[docs]
@kernel
def identity_int32(x: TInt32) -> TInt32:
return x
[docs]
@kernel
def identity_float_int32(x: TFloat) -> TInt32:
return round(x)
[docs]
@portable
def isnan(x: TFloat) -> TBool:
"""
Check whether a float is NaN.
We cannot use `math.isnan` or `numpy.isnan` in the core, so we use IEEE 754 and do it by hand.
"""
return x != x
[docs]
@portable
def less_than_uint32(x:TInt32, y:TInt32) -> TBool:
"""
Helper function to compare two unsigned int32 bit register `x` and `y` values which are a int32 (x<y).
This is a necessary workaround as the ARTIQ kernel does not support unsigned integers.
"""
if not (x ^ y) & 0x80000000:
# same sign, comparison as usual
return x < y
else:
# if x > 0, x is the smaller value so return True
return x > 0
[docs]
@portable
def unsigned_bitshift32(x:TInt32, bits:TInt32) -> TInt32:
"""
Implemets a right bitshift operation on an signed `int32` (as supported by the artiq kernel) which encodes an
unsigned `uint32` value.
"""
bits &= 31
mask = ~(~np.int32(0) << (32 - bits))
return np.int32((x >> bits) & mask)
[docs]
def master_ipc_call(experiment, action: str):
"""
Return a method to call the artiq master through internal IPC communication
Args:
experiment: The artiq/atomiq experiment object
action: action in the master process to call with this function
Returns:
Method that calles the requested RPC endpoint in the master
"""
# we copy an existing IPC calling function and modify its closure to perform the action we want
caller = copy.deepcopy(experiment.scheduler._submit)
caller.__closure__[0].cell_contents = action
return caller
[docs]
def dummy_decorator(func):
return func