-
Notifications
You must be signed in to change notification settings - Fork 21
Added JSON serialisation for Lambda functions #1695
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
base: main
Are you sure you want to change the base?
Changes from all commits
2e7bd34
61d5990
f96223a
19b978b
39c1b1a
93d64c2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Large diffs are not rendered by default.
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| # This file is for defining the blueprint object | ||
|
|
||
| import functools as ft | ||
| import inspect | ||
| import json | ||
| import re | ||
| import warnings | ||
|
|
@@ -81,9 +82,15 @@ def __init__( | |
| # Make special functions live in the funlist but transfer their names | ||
| # to the namelist | ||
| # Infer names from signature if not given, i.e. allow for '' names | ||
| # for ii, name in enumerate(namelist): | ||
| # if isinstance(funlist[ii], str): | ||
| # namelist[ii] = funlist[ii] | ||
| # elif name == "": | ||
| # namelist[ii] = funlist[ii].__name__ | ||
| for ii, name in enumerate(namelist): | ||
| if isinstance(funlist[ii], str): | ||
| namelist[ii] = funlist[ii] | ||
| if name == "": | ||
| namelist[ii] = funlist[ii] | ||
| elif name == "": | ||
| namelist[ii] = funlist[ii].__name__ | ||
|
Comment on lines
+85
to
95
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. also, what happens now in the else of why keeping the commented out code part? |
||
|
|
||
|
|
@@ -93,8 +100,8 @@ def __init__( | |
| argslist[ii] = (args,) | ||
| self._argslist = argslist | ||
|
|
||
| self._namelist = namelist | ||
| namelist = self._make_names_unique(namelist) | ||
| self._namelist = namelist | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this looks like a good catch! thank you! |
||
|
|
||
| # initialise markers | ||
| if marker1 is None: | ||
|
|
@@ -265,6 +272,55 @@ def description(self): | |
| desc[segkey]["durations"] = self._durslist[sn] | ||
| if desc[segkey]["function"] == "waituntil": | ||
| desc[segkey]["arguments"] = {"waittime": self._argslist[sn]} | ||
| elif desc[segkey]["function"] == "function PulseAtoms.arb_func": | ||
| # Special handling for arb_func serialization | ||
| func_obj, kwargs_dict = self._argslist[sn] | ||
|
|
||
| # Serialize the function | ||
| if hasattr(func_obj, "__name__") and func_obj.__name__ != "<lambda>": | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. could you move the logic of extracting code a function into a separate function in order not to mix it with the logic of creating blueprint description? |
||
| # Regular function - store name and try to get source | ||
| func_name = func_obj.__name__ | ||
| try: | ||
| func_source = inspect.getsource(func_obj) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. isn't it extremely fragile? What is a function is using functions from math module or numpy, would those imports be also part of the source? also, i can't imagine that noone tried to solve this problem before - can't more functions/methods from the python standard library or perhaps even an external package be used to support this rather than implementing a custom function code serialization here? |
||
| except (OSError, TypeError): | ||
| func_source = None | ||
| desc[segkey]["arguments"] = { | ||
| "func_type": "named_function", | ||
| "func_name": func_name, | ||
| "func_source": func_source, | ||
| "kwargs": kwargs_dict, | ||
| } | ||
| else: | ||
| # Lambda function - store source code | ||
| # First check if the lambda has a __func_source__ attribute | ||
| # (for dynamically created lambdas) | ||
| if hasattr(func_obj, "__func_source__"): | ||
| func_source = func_obj.__func_source__ | ||
| else: | ||
| # Fall back to inspect.getsource() with regex parsing | ||
| try: | ||
| func_source = inspect.getsource(func_obj) | ||
| # Extract just the lambda part using regex | ||
| import re | ||
|
|
||
| # Match 'lambda' followed by parameters, colon, and expression | ||
| # This handles nested parentheses and complex expressions | ||
| lambda_match = re.search( | ||
| r"lambda\s+[^:]*:\s*[^\n,;]+", func_source | ||
| ) | ||
| if lambda_match: | ||
| func_source = lambda_match.group(0).strip() | ||
| else: | ||
| func_source = "lambda t, **kwargs: 0" | ||
| except (OSError, TypeError): | ||
| # Fallback: create a generic lambda string | ||
| func_source = "lambda t, **kwargs: 0" # Default fallback | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. shouldn't it rather be None, like in case the source code of named function is not possible to retireve? This would make the behavior consistent while having a default fallback can become confusing |
||
|
|
||
| desc[segkey]["arguments"] = { | ||
| "func_type": "lambda", | ||
| "func_source": func_source, | ||
| "kwargs": kwargs_dict, | ||
| } | ||
| else: | ||
| sig = signature(self._funlist[sn]) | ||
| desc[segkey]["arguments"] = dict( | ||
|
|
@@ -275,6 +331,7 @@ def description(self): | |
| desc["marker2_abs"] = self.marker2 | ||
| desc["marker1_rel"] = self._segmark1 | ||
| desc["marker2_rel"] = self._segmark2 | ||
| desc["SR"] = self._SR | ||
|
|
||
| return desc | ||
|
|
||
|
|
@@ -312,7 +369,76 @@ def blueprint_from_description(cls, blue_dict): | |
| if seg_dict["function"] == "waituntil": | ||
| arguments = blue_dict[seg]["arguments"].values() | ||
| arguments = (list(arguments)[0][0],) | ||
| bp_seg.insertSegment(i, "waituntil", arguments) | ||
| bp_seg.insertSegment(i, "waituntil", arguments, name=seg_dict["name"]) | ||
| elif seg_dict["function"] == "function PulseAtoms.arb_func": | ||
| # Special handling for arb_func reconstruction | ||
| args_dict = blue_dict[seg]["arguments"] | ||
|
|
||
| if args_dict.get("func_type") == "lambda": | ||
| # Reconstruct lambda function | ||
| func_source = args_dict["func_source"] | ||
| try: | ||
| # Create lambda function from source | ||
| func_obj = eval(func_source) | ||
| except (SyntaxError, NameError) as e: | ||
| # Fallback: create a zero function | ||
| print( | ||
| f"Warning: Could not reconstruct lambda function '{func_source}'. Using zero function. Error: {e}" | ||
| ) | ||
|
|
||
| def zero_function(t, **kwargs): | ||
| return 0 | ||
|
|
||
| func_obj = zero_function | ||
|
|
||
| kwargs_dict = args_dict["kwargs"] | ||
| arguments = (func_obj, kwargs_dict) | ||
| elif args_dict.get("func_type") == "named_function": | ||
| # Reconstruct named function | ||
| func_name = args_dict["func_name"] | ||
| func_source = args_dict.get("func_source") | ||
| kwargs_dict = args_dict["kwargs"] | ||
|
|
||
| # Try to reconstruct from source first | ||
| func_obj = None | ||
| if func_source: | ||
| try: | ||
| # Execute the function source in a local namespace | ||
| local_ns = {} | ||
| exec(func_source, globals(), local_ns) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is dangerous to make globals available, both from security perspective but also from the perspective of writing code that would prepare some globals to exist before deserializing certain blueprints and their arbitrary functions. Would it be possible to limit support to self-sufficient functions? at least to start with |
||
| if func_name in local_ns: | ||
| func_obj = local_ns[func_name] | ||
| except Exception as e: | ||
| print( | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. next to printing please use logging.warning so that this information can also show as a proper warning, and be included in logs (if those are used) |
||
| f"Warning: Could not reconstruct named function '{func_name}' from source. Error: {e}" | ||
| ) | ||
|
|
||
| # Fallback: try to find function in globals | ||
| if func_obj is None: | ||
| try: | ||
| func_obj = globals()[func_name] | ||
| except KeyError: | ||
| print( | ||
| f"Warning: Could not find function '{func_name}' in globals. Using zero function." | ||
| ) | ||
|
|
||
| def zero_function(t, **kwargs): | ||
| return 0 | ||
|
|
||
| func_obj = zero_function | ||
|
|
||
| arguments = (func_obj, kwargs_dict) | ||
| else: | ||
| # Legacy format or fallback | ||
| arguments = tuple(blue_dict[seg]["arguments"].values()) | ||
|
|
||
| bp_seg.insertSegment( | ||
| i, | ||
| knowfunctions[seg_dict["function"]], | ||
| arguments, | ||
| name=re.sub(r"\d", "", seg_dict["name"]), | ||
| dur=seg_dict["durations"], | ||
| ) | ||
| else: | ||
| arguments = tuple(blue_dict[seg]["arguments"].values()) | ||
| bp_seg.insertSegment( | ||
|
|
@@ -329,6 +455,8 @@ def blueprint_from_description(cls, blue_dict): | |
| listmarker2 = blue_dict["marker2_rel"] | ||
| bp_sum._segmark1 = [tuple(mark) for mark in listmarker1] | ||
| bp_sum._segmark2 = [tuple(mark) for mark in listmarker2] | ||
| if "SR" in blue_dict: | ||
| bp_sum._SR = blue_dict["SR"] | ||
| return bp_sum | ||
|
|
||
| @classmethod | ||
|
|
@@ -664,7 +792,6 @@ def insertSegment(self, pos, func, args=(), dur=None, name=None, durs=None): | |
|
|
||
| if pos < -1: | ||
| raise ValueError("Position must be strictly larger than -1") | ||
|
|
||
| if name is None or name == "": | ||
| if func == "waituntil": | ||
| name = "waituntil" | ||
|
|
@@ -674,7 +801,6 @@ def insertSegment(self, pos, func, args=(), dur=None, name=None, durs=None): | |
| if len(name) > 0: | ||
| if name[-1].isdigit(): | ||
| raise ValueError("Segment name must not end in a number") | ||
|
|
||
| if pos == -1: | ||
| self._namelist.append(name) | ||
| self._namelist = self._make_names_unique(self._namelist) | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why are there two duplicated(?) plots instead of one in this example? or is it github mis-rendering?