An object representing a custom EE Function.
| 19 | # - as a Function: encode its invocation (which may also involve encoding its |
| 20 | # definition, if that hasn't happened yet). |
| 21 | class CustomFunction(function.Function, encodable.Encodable): |
| 22 | """An object representing a custom EE Function.""" |
| 23 | _body: Any |
| 24 | _signature: dict[str, Any] |
| 25 | |
| 26 | def __init__(self, signature: dict[str, Any], body: Any): |
| 27 | """Creates a function defined by a given expression with unbound variables. |
| 28 | |
| 29 | The expression is created by evaluating the given function |
| 30 | using variables as placeholders. |
| 31 | |
| 32 | Args: |
| 33 | signature: The function signature. If any of the argument names are |
| 34 | null, their names will be generated deterministically, based on |
| 35 | the body. |
| 36 | body: The Python function to evaluate. |
| 37 | """ |
| 38 | variables = [CustomFunction.variable(arg['type'], arg['name']) |
| 39 | for arg in signature['args']] |
| 40 | |
| 41 | if body(*variables) is None: |
| 42 | raise ee_exception.EEException('User-defined methods must return a value') |
| 43 | |
| 44 | # The signature of the function. |
| 45 | self._signature = CustomFunction._resolveNamelessArgs( |
| 46 | signature, variables, body) |
| 47 | |
| 48 | # The expression to evaluate. |
| 49 | self._body = body(*variables) |
| 50 | |
| 51 | def encode(self, encoder: Callable[[Any], Any]) -> dict[str, Any]: |
| 52 | return { |
| 53 | 'type': 'Function', |
| 54 | 'argumentNames': [x['name'] for x in self._signature['args']], |
| 55 | 'body': encoder(self._body) |
| 56 | } |
| 57 | |
| 58 | def encode_cloud_value(self, encoder: Callable[[Any], Any]) -> dict[str, Any]: |
| 59 | return { |
| 60 | 'functionDefinitionValue': { |
| 61 | 'argumentNames': [x['name'] for x in self._signature['args']], |
| 62 | 'body': encoder(self._body) |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | def encode_invocation(self, encoder: Callable[[Any], Any]) -> dict[str, Any]: |
| 67 | return self.encode(encoder) |
| 68 | |
| 69 | def encode_cloud_invocation( |
| 70 | self, encoder: Callable[[Any], Any] |
| 71 | ) -> dict[str, Any]: |
| 72 | return {'functionReference': encoder(self)} |
| 73 | |
| 74 | def getSignature(self) -> dict[str, Any]: |
| 75 | """Returns a description of the interface provided by this function.""" |
| 76 | return self._signature |
| 77 | |
| 78 | @staticmethod |