| # ------------------------------------------------------------- |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| # |
| # ------------------------------------------------------------- |
| |
| from typing import Union, Optional, Iterable, Dict, Sequence, Tuple, TYPE_CHECKING |
| from multiprocessing import Process |
| |
| import numpy as np |
| from py4j.java_gateway import JVMView, JavaObject |
| |
| from systemds.utils.consts import VALID_INPUT_TYPES, BINARY_OPERATIONS, VALID_ARITHMETIC_TYPES |
| from systemds.utils.helpers import create_params_string |
| from systemds.utils.converters import matrix_block_to_numpy |
| from systemds.script_building.script import DMLScript |
| from systemds.script_building.dag import OutputType, DAGNode |
| |
| |
| if TYPE_CHECKING: |
| # to avoid cyclic dependencies during runtime |
| from systemds.context import SystemDSContext |
| |
| |
| class OperationNode(DAGNode): |
| """A Node representing an operation in SystemDS""" |
| _result_var: Optional[Union[float, np.array]] |
| _lineage_trace: Optional[str] |
| _script: Optional[DMLScript] |
| _output_types: Optional[Iterable[VALID_INPUT_TYPES]] |
| |
| def __init__(self, sds_context: 'SystemDSContext', operation: str, |
| unnamed_input_nodes: Iterable[VALID_INPUT_TYPES] = None, |
| named_input_nodes: Dict[str, VALID_INPUT_TYPES] = None, |
| output_type: OutputType = OutputType.MATRIX, |
| is_python_local_data: bool = False, |
| number_of_outputs=1, |
| output_types: Iterable[OutputType] = None): |
| """ |
| Create general `OperationNode` |
| |
| :param sds_context: The SystemDS context for performing the operations |
| :param operation: The name of the DML function to execute |
| :param unnamed_input_nodes: inputs identified by their position, not name |
| :param named_input_nodes: inputs with their respective parameter name |
| :param output_type: type of the output in DML (double, matrix etc.) |
| :param is_python_local_data: if the data is local in python e.g. Numpy arrays |
| :param number_of_outputs: If set to other value than 1 then it is expected |
| that this operation node returns multiple values. If set remember to set the output_types value as well. |
| :param output_types: The types of output in a multi output scenario. |
| Default is None, and means every multi output is a matrix. |
| """ |
| self.sds_context = sds_context |
| if unnamed_input_nodes is None: |
| unnamed_input_nodes = [] |
| if named_input_nodes is None: |
| named_input_nodes = {} |
| self.operation = operation |
| self._unnamed_input_nodes = unnamed_input_nodes |
| self._named_input_nodes = named_input_nodes |
| self.output_type = output_type |
| self._is_python_local_data = is_python_local_data |
| self._result_var = None |
| self._lineage_trace = None |
| self._script = None |
| self._number_of_outputs = number_of_outputs |
| self._output_types = output_types |
| |
| def compute(self, verbose: bool = False, lineage: bool = False) -> \ |
| Union[float, np.array, Tuple[Union[float, np.array], str]]: |
| |
| if self._result_var is None or self._lineage_trace is None: |
| self._script = DMLScript(self.sds_context) |
| self._script.build_code(self) |
| if verbose: |
| print("SCRIPT:") |
| print(self._script.dml_script) |
| |
| if lineage: |
| result_variables, self._lineage_trace = self._script.execute( |
| lineage) |
| else: |
| result_variables = self._script.execute(lineage) |
| |
| if self.output_type == OutputType.DOUBLE: |
| self._result_var = result_variables.getDouble( |
| self._script.out_var_name[0]) |
| elif self.output_type == OutputType.MATRIX: |
| self._result_var = matrix_block_to_numpy(self.sds_context.java_gateway.jvm, |
| result_variables.getMatrixBlock(self._script.out_var_name[0])) |
| elif self.output_type == OutputType.LIST: |
| self._result_var = [] |
| for idx, v in enumerate(self._script.out_var_name): |
| if(self._output_types == None): |
| self._result_var.append(matrix_block_to_numpy(self.sds_context.java_gateway.jvm, |
| result_variables.getMatrixBlock(v))) |
| elif(self._output_types[idx] == OutputType.MATRIX): |
| self._result_var.append(matrix_block_to_numpy(self.sds_context.java_gateway.jvm, |
| result_variables.getMatrixBlock(v))) |
| else: |
| self._result_var.append(result_variables.getDouble( |
| self._script.out_var_name[idx])) |
| if verbose: |
| for x in self.sds_context.get_stdout(): |
| print(x) |
| for y in self.sds_context.get_stderr(): |
| print(y) |
| |
| if lineage: |
| return self._result_var, self._lineage_trace |
| else: |
| return self._result_var |
| |
| def get_lineage_trace(self) -> str: |
| """Get the lineage trace for this node. |
| |
| :return: Lineage trace |
| """ |
| if self._lineage_trace is None: |
| self._script = DMLScript(self.sds_context) |
| self._script.build_code(self) |
| self._lineage_trace = self._script.get_lineage() |
| |
| return self._lineage_trace |
| |
| def code_line(self, var_name: str, unnamed_input_vars: Sequence[str], |
| named_input_vars: Dict[str, str]) -> str: |
| if self.operation in BINARY_OPERATIONS: |
| assert len( |
| named_input_vars) == 0, 'Named parameters can not be used with binary operations' |
| assert len( |
| unnamed_input_vars) == 2, 'Binary Operations need exactly two input variables' |
| return f'{var_name}={unnamed_input_vars[0]}{self.operation}{unnamed_input_vars[1]}' |
| |
| inputs_comma_sep = create_params_string( |
| unnamed_input_vars, named_input_vars) |
| |
| if self.output_type == OutputType.LIST: |
| output = "[" |
| for idx in range(self._number_of_outputs): |
| output += f'{var_name}_{idx},' |
| output = output[:-1] + "]" |
| return f'{output}={self.operation}({inputs_comma_sep});' |
| else: |
| return f'{var_name}={self.operation}({inputs_comma_sep});' |
| |
| def pass_python_data_to_prepared_script(self, jvm: JVMView, var_name: str, prepared_script: JavaObject) -> None: |
| raise NotImplementedError( |
| 'Operation node has no python local data. Missing implementation in derived class?') |
| |
| def _check_matrix_op(self): |
| """Perform checks to assure operation is allowed to be performed on data type of this `OperationNode` |
| |
| :raise: AssertionError |
| """ |
| assert self.output_type == OutputType.MATRIX, f'{self.operation} only supported for matrices' |
| |
| def __add__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '+', [self, other]) |
| |
| def __sub__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '-', [self, other]) |
| |
| def __mul__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '*', [self, other]) |
| |
| def __truediv__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '/', [self, other]) |
| |
| def __floordiv__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '//', [self, other]) |
| |
| def __lt__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '<', [self, other]) |
| |
| def __le__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '<=', [self, other]) |
| |
| def __gt__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '>', [self, other]) |
| |
| def __ge__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '>=', [self, other]) |
| |
| def __eq__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '==', [self, other]) |
| |
| def __ne__(self, other) -> 'OperationNode': |
| return OperationNode(self.sds_context, '!=', [self, other]) |
| |
| def __matmul__(self, other: VALID_ARITHMETIC_TYPES) -> 'OperationNode': |
| return OperationNode(self.sds_context, '%*%', [self, other]) |
| |
| def sum(self, axis: int = None) -> 'OperationNode': |
| """Calculate sum of matrix. |
| |
| :param axis: can be 0 or 1 to do either row or column sums |
| :return: `OperationNode` representing operation |
| """ |
| self._check_matrix_op() |
| if axis == 0: |
| return OperationNode(self.sds_context, 'colSums', [self]) |
| elif axis == 1: |
| return OperationNode(self.sds_context, 'rowSums', [self]) |
| elif axis is None: |
| return OperationNode(self.sds_context, 'sum', [self], output_type=OutputType.DOUBLE) |
| raise ValueError( |
| f"Axis has to be either 0, 1 or None, for column, row or complete {self.operation}") |
| |
| def mean(self, axis: int = None) -> 'OperationNode': |
| """Calculate mean of matrix. |
| |
| :param axis: can be 0 or 1 to do either row or column means |
| :return: `OperationNode` representing operation |
| """ |
| self._check_matrix_op() |
| if axis == 0: |
| return OperationNode(self.sds_context, 'colMeans', [self]) |
| elif axis == 1: |
| return OperationNode(self.sds_context, 'rowMeans', [self]) |
| elif axis is None: |
| return OperationNode(self.sds_context, 'mean', [self], output_type=OutputType.DOUBLE) |
| raise ValueError( |
| f"Axis has to be either 0, 1 or None, for column, row or complete {self.operation}") |
| |
| def var(self, axis: int = None) -> 'OperationNode': |
| """Calculate variance of matrix. |
| |
| :param axis: can be 0 or 1 to do either row or column vars |
| :return: `OperationNode` representing operation |
| """ |
| self._check_matrix_op() |
| if axis == 0: |
| return OperationNode(self.sds_context, 'colVars', [self]) |
| elif axis == 1: |
| return OperationNode(self.sds_context, 'rowVars', [self]) |
| elif axis is None: |
| return OperationNode(self.sds_context, 'var', [self], output_type=OutputType.DOUBLE) |
| raise ValueError( |
| f"Axis has to be either 0, 1 or None, for column, row or complete {self.operation}") |
| |
| def abs(self) -> 'OperationNode': |
| """Calculate absolute. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'abs', [self]) |
| |
| def sin(self) -> 'OperationNode': |
| """Calculate sin. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'sin', [self]) |
| |
| def cos(self) -> 'OperationNode': |
| """Calculate cos. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'cos', [self]) |
| |
| def tan(self) -> 'OperationNode': |
| """Calculate tan. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'tan', [self]) |
| |
| def asin(self) -> 'OperationNode': |
| """Calculate arcsin. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'asin', [self]) |
| |
| def acos(self) -> 'OperationNode': |
| """Calculate arccos. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'acos', [self]) |
| |
| def atan(self) -> 'OperationNode': |
| """Calculate arctan. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'atan', [self]) |
| |
| def sinh(self) -> 'OperationNode': |
| """Calculate sin. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'sinh', [self]) |
| |
| def cosh(self) -> 'OperationNode': |
| """Calculate cos. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'cosh', [self]) |
| |
| def tanh(self) -> 'OperationNode': |
| """Calculate tan. |
| |
| :return: `OperationNode` representing operation |
| """ |
| return OperationNode(self.sds_context, 'tanh', [self]) |
| |
| def moment(self, moment, weights: DAGNode = None) -> 'OperationNode': |
| # TODO write tests |
| self._check_matrix_op() |
| unnamed_inputs = [self] |
| if weights is not None: |
| unnamed_inputs.append(weights) |
| unnamed_inputs.append(moment) |
| return OperationNode(self.sds_context, 'moment', unnamed_inputs, output_type=OutputType.DOUBLE) |
