Source code for pyarts.workspace.workspace

import os
import sys
from ast import parse, Call, Name, Expression, Expr, FunctionDef, \
    Starred, Module, Str
from inspect import getsource, getclosurevars, ismodule, isclass
from copy import copy
from typing import Callable

import pyarts.arts as cxx
from pyarts.workspace.utility import unindent as unindent


# Set the default basename of Arts
try:
    filename = sys.modules["__main__"].__file__
    basename, _ = os.path.splitext(os.path.basename(filename))
    cxx.globals.parameters.out_basename = basename
except:
    pass


_InternalWorkspace = getattr(cxx, "_Workspace")
Agenda = cxx.Agenda


[docs] class DelayedAgenda: """ Helper class to delay the parsing of an Agenda until a workspace exist """
[docs] def __init__(self, *args): self.args = [[*args]]
[docs] def append_agenda_methods(self, other): self.args.extend(other.args)
def __call__(self, ws): a = cxx.Agenda(ws) for args in self.args: a.append_agenda_methods(continue_parser_function(ws, *args)) a.name = "<unknown>" return a
[docs] def Include(ws, path): """ Parse and execute the .arts file at path onto the current workspace ws The Arts parser is invoked on the file at path. The methods and commands of this file are executed """ if isinstance(path, Agenda): path.execute(ws) else: Agenda(ws, path).execute(ws)
[docs] def arts_agenda(func=None, *, ws=None, allow_callbacks=False, set_agenda=False): """ Decorator to parse a Python method as ARTS agenda This decorator can be used to define ARTS agendas using python function syntax. The function should have one arguments which is assumed to be a :class:`Workspace` instance. All expressions inside the function must be calls to ARTS WSMs. The function definition results in an Agenda object that can be copied into an ARTS agenda. Example:: ws = pyarts.workspace.Workspace() @pyarts.workspace.arts_agenda(ws=ws) def iy_main_agenda_clearsky(ws): ws.ppathCalc() ws.iyClearsky() ws.VectorSet(ws.geo_pos, []) ws.iy_main_agenda = iy_main_agenda_clearsky Or, by passing `set_agenda=True`, write the agenda directly to the workspace (allowing initiating automatic checking if the agenda is a defined workspace agenda):: ws = pyarts.workspace.Workspace() @pyarts.workspace.arts_agenda(ws=ws, set_agenda=True) def iy_main_agenda(ws): ws.ppathCalc() ws.iyClearsky() ws.VectorSet(ws.geo_pos, []) When the decorator is used with the `allow_callbacks` keyword argument set to True, arbitrary Python code can be executed within the callback. Note, however, that ARTS ignores exceptions occurring within the callback, so care must be taken that potentially silenced errors don't interfere with simulation results. .. warning:: Using `allow_callbacks=True` breaks the Agenda input-output control. It is therefore considered undefined behavior if you manipulate workspace variables that are neither in- nor output of the Agenda using callbacks. *Do this at your own risk*. Example:: ws = pyarts.workspace.Workspace() @pyarts.workspace.arts_agenda(ws=ws, allow_callbacks=True) def python_agenda(ws): print("Python says 'hi'.") A special `INCLUDE(path)` directive can be part of the function definitions to use the Arts parser of .arts files to be invoked on the file. All methods and invokations that are part of the .arts file are appended in place to the agenda To predefine an agenda before a workspace is created, for instance for use with multiple workspaces later, use the :class:`DelayedAgenda` class. A :class:`DelayedAgenda` is created by omitting the ws keyword argument:: @pyarts.workspace.arts_agenda def iy_main_agenda_clearsky(ws): ws.ppathCalc() ws.iyClearsky() ws.VectorSet(ws.geo_pos, []) ws = pyarts.workspace.Workspace() ws.iy_main_agenda = iy_main_agenda_clearsky """ def agenda_decorator(func): return parse_function(func, ws, allow_callbacks=allow_callbacks, set_agenda=set_agenda) if func is None: return agenda_decorator else: return parse_function(func, ws, False, False)
[docs] def parse_function(func, arts, allow_callbacks, set_agenda): """ Parse python method as ARTS agenda Args: func: The function object to parse. allow_callbacks: Whether to allow callbacks in the agenda. Return: An 'Agenda' object containing the code in the given function. """ source = getsource(func) source = unindent(source) ast = parse(source) context = copy(func.__globals__) dellist = [] for key in context: if key.startswith("__"): continue if (not ismodule(context[key]) and not isclass(context[key]) and not isinstance(context[key], Callable)): dellist.append(key) for key in dellist: del context[key] nls, _, _, _ = getclosurevars(func) context.update(nls) if arts is None: return DelayedAgenda(context, ast, allow_callbacks, set_agenda) return continue_parser_function(arts, context, ast, allow_callbacks, set_agenda)
[docs] def continue_parser_function(arts, context, ast, allow_callbacks, set_agenda): assert isinstance(arts, _InternalWorkspace), f"Expects Workspace, got {type(arts)}" func_ast = ast.body[0] if not isinstance(func_ast, FunctionDef): raise Exception("ARTS agenda definition can only decorate function definitions.") args = func_ast.args.args try: arg_name = func_ast.args.args[0].arg except: raise Exception("Agenda definition needs workspace arguments.") context.update({arg_name : arts}) # # Helper functions # callback_body = [] def callback_make_fun(body): """ Helper function that creates a wrapper function around python code to be executed withing an ARTS agenda. """ if sys.version_info >= (3, 8): # https://bugs.python.org/issue35894#msg334808 m = Module(body, []) else: m = Module(body) def callback(ws): # FIXME: Is mutex required here? context[arg_name].swap(ws) # FIXME: This is required, of course eval(compile(m , "<unknown>", 'exec'), context) context[arg_name].swap(ws) # FIXME: But is this required or thrown away? return callback def eval_argument(expr): """ Evaluate argument of workspace method call. """ if not hasattr(expr, "lineno"): setattr(expr, "lineno", 0) try: ret = eval(compile(Expression(expr), "<unknown>", 'eval'), context) except NameError: try: from ast import unparse errstr = f"the local Python variable `{unparse(expr)}`" except ImportError: errstr = "a local Python variable" raise NameError( f"You seem to want to pass {errstr} into a WSM.\n" "This breaks scoping rules. You can only pass literals into WSMs." ) return ret illegal_statement_exception = Exception(""" Pure ARTS agenda definitions may only contain calls to WSMs of the workspace argument '{arg_name}' or INCLUDE statements. If you want to allow Python callbacks you need to use the '@arts_agenda' decorator with the 'allow_callbacks' keyword argument set to 'True'. WARNING: This will break the Agenda input-output control. To ensure proper scoping, you need to explicitly `ws.Touch` every WSV you modify. That includes output variables from WSMs you might call. Everything else is undefined behaviour. ;-) """) workspace_methods = [str(x.name) for x in cxx.globals.get_md_data()] agenda = Agenda(arts) for e in func_ast.body: if not isinstance(e, Expr): if allow_callbacks: callback_body += [e] continue else: raise illegal_statement_exception else: call = e.value if not isinstance(call, Call): if isinstance(call, Str): continue elif allow_callbacks: callback_body += [e] continue else: raise illegal_statement_exception # Include statement if type(call.func) == Name: if call.func.id != "INCLUDE": if allow_callbacks: callback_body += [e] else: raise illegal_statement_exception else: args = [] for a in call.args: args.append(eval_argument(a)) include_agenda = Agenda(arts, *args) if len(callback_body) > 0: agenda.add_callback_method(callback_make_fun(callback_body)) callback_body = [] agenda.append_agenda_methods(include_agenda) else: att = call.func.value if not hasattr(att, 'id') or not att.id == arg_name: callback_body += [e] continue # Extract method name. name = call.func.attr # m is not a workspace method if name not in workspace_methods: if allow_callbacks: callback_body += [e] continue else: raise ValueError( f"{name} is not a know ARTS WSM." ) args = [] kwargs = dict() for a in call.args: # Handle starred expression if type(a) == Starred: bs = eval_argument(a.value) for b in bs: args.append(b) else: args.append(eval_argument(a)) # Extract keyword arguments for k in call.keywords: if k.arg is None: d = eval(compile(Expression(k.value), "<unknown>", 'eval'), context) kwargs.update(d) else: kwargs[k.arg] = eval(compile(Expression(k.value), "<unknown>", 'eval'), context) # Add function to agenda if len(callback_body) > 0: agenda.add_callback_method(callback_make_fun(callback_body)) callback_body = [] agenda.add_workspace_method(name, *args, **kwargs) # Check if there's callback code left to add to the agenda. if len(callback_body) > 0: agenda.add_callback_method(callback_make_fun(callback_body)) callback_body = [] agenda.name = func_ast.name if set_agenda: setattr(arts, func_ast.name, agenda) return agenda
_group_types = [eval(f"cxx.{x.name}") for x in list(cxx.globals.get_wsv_groups())]
[docs] class Workspace(_InternalWorkspace): """ A wrapper for the C++ workspace object """ def __getattribute__(self, attr): if attr.startswith("__"): object.__getattribute__(self, attr) return super().__getattribute__(attr) def __getattr__(self, attr): if super()._hasattr_check_(attr): return super()._getattr_unchecked_(attr) raise AttributeError( f"'Workspace' object has no attribute '{attr}'") def __setattr__(self, attr, value): if self._hasattr_check_(attr): if isinstance(value, DelayedAgenda): value = value(self) self._getattr_unchecked_(attr).initialize_if_not() self._getattr_unchecked_(attr).value = type( self._getattr_unchecked_(attr).value)(value) else: if type(value) in _group_types: self._setattr_unchecked_(attr, value) elif isinstance(value, DelayedAgenda): self._setattr_unchecked_(attr, value(self)) else: raise AttributeError( f"'Workspace' object has no attribute '{attr}'") def __delattr__(self, attr): if attr == '__class__': raise AttributeError("You cannot delete __class__") getattr(self, attr).delete_level() def __copy__(self): x = super().__copy__() out = Workspace() out.swap(x) return out def __deepcopy__(self, *args): x = super().__deepcopy__(*args) out = Workspace() out.swap(x) return out def __getstate__(self): return {"Workspace": super().__getstate__()} def __setstate__(self, d): super().__setstate__(d["Workspace"])