11447759208

Committed 21 Oct 2024 08:06PM UTC coverage: 83.676% (-1.3%) from 84.952%

Build # 11447759208

Build Type

Pull #3071

github

Committed by

web-flow

Commit Message

tests: Use MappingProxyType to pass dict test parameters (#3076)

Prevent accidental modifications of global dictionary instances.

Pull Request Pull Request #3071: Devel

Run Details

9267 of 12302 branches covered (75.33%)

Branch coverage included in aggregate %.

2134 of 2524 new or added lines in 66 files covered. (84.55%)

123 existing lines in 26 files now uncovered.

32176 of 37226 relevant lines covered (86.43%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.08

/psyneulink/core/globals/mdf.py

"""

Contents
--------

  * `MDF_Overview`
  * `MDF_Examples`
  * `MDF_Model_Specification`

.. _MDF_Overview:


Overview
--------

The developers of PsyNeuLink are collaborating with the scientific community, as part of the `OpenNeuro effort
<https://openneuro.org>`_, to create a standard, serialzied format for the description and exchange of computational
models of brain and psychological function across different simulation environments. As part of this effort,
PsyNeuLink supports the `ModECI Model Description Format <https://github.com/ModECI/MDF/includes>`_ (MDF) by
including the ability to produce an MDF-compatible model from a PsyNeuLink model and to construct valid Python
scripts that express a PsyNeuLink model from an MDF model.

Any PsyNeuLink `Composition` or `Component` can be exported to MDF format using its `as_mdf_model` method or
to serialized format using its `json_summary` or `yaml_summary` methods. These methods generate strings that, passed into the
`generate_script_from_mdf` function, produce a valid Python script replicating the original PsyNeuLink model.
`write_mdf_file` can be used to write the serialization for one or more Compositions into a specified file (though
see `note <MDF_Write_Multiple_Compositions_Note>`). `generate_script_from_mdf` can accept either the string returned
by `get_mdf_serialized` or the name of a file containing one.
Calling ``exec(generate_script_from_mdf(<input>))`` will load into the current namespace all of the PsyNeuLink
objects specified in the ``input``; and `get_compositions` can be used to retrieve a list of all of the Compositions
in that namespace, including any generated by execution of `generate_script_from_mdf`. `generate_script_from_mdf`
may similarly be used to create a PsyNeuLink Python script from a ModECI MDF Model object, such as that created
by `as_mdf_model <Composition.as_mdf_model>`.

.. _MDF_Security_Warning:

.. warning::
   Use of `generate_script_from_json` or `generate_script_from_mdf` to generate a Python script from a file without taking proper precautions can
   introduce a security risk to the system on which the Python interpreter is running.  This is because it calls
   exec, which has the potential to execute non-PsyNeuLink-related code embedded in the file.  Therefore,
   `generate_script_from_json` or `generate_script_from_mdf` should be used to read only files of known and secure origin.

.. _MDF_Examples:

Model Examples
--------------

Below is an example of a script that implements a PsyNeuLink model of the Stroop model with conflict monitoring,
and its output in JSON. Running `generate_script_from_json` on the output will produce another PsyNeuLink script
that will give the same results when run on the same input as the original.

:download:`Download stroop_conflict_monitoring.py
<../../tests/mdf/stroop_conflict_monitoring.py>`

:download:`Download stroop_conflict_monitoring.json
<../../docs/source/_static/stroop_conflict_monitoring.json>`

.. _MDF_Model_Specification:

MDF Model Specification
------------------------

.. note::
    The format is in development, and is subject to change.

See https://github.com/ModECI/MDF/blob/main/docs/README.md#model


.. _MDF_Simple_Edge_Format:

MDF Simple Edge Format
----------------------

Models may be output as they are in PsyNeuLink or in "simple edge"
format. In simple edge format, PsyNeuLink Projections are written as a
combination of two Edges and an intermediate Node, because the generic
MDF execution engine does not support using Functions on Edges.
PsyNeuLink is capable of re-importing models exported by PsyNeuLink in
either form.
"""

import ast
import base64
import binascii
import dill
import enum
import graph_scheduler
import inspect
import json
import math
import numbers
import numpy
import os
import pickle
import pint
import psyneulink
import re
import tempfile
import types
import time
import warnings

from psyneulink.core.globals.keywords import \
    MODEL_SPEC_ID_GENERIC, MODEL_SPEC_ID_PARAMETER_SOURCE, \
    MODEL_SPEC_ID_PARAMETER_INITIAL_VALUE, MODEL_SPEC_ID_PARAMETER_VALUE, MODEL_SPEC_ID_PSYNEULINK, \
    MODEL_SPEC_ID_TYPE, MODEL_SPEC_ID_MDF_VARIABLE, MODEL_SPEC_ID_SHAPE, MODEL_SPEC_ID_METADATA, MODEL_SPEC_ID_INPUT_PORT_COMBINATION_FUNCTION
from psyneulink.core.globals.parameters import ParameterAlias
from psyneulink.core.globals.sampleiterator import SampleIterator
from psyneulink.core.globals.utilities import convert_to_list, gen_friendly_comma_str, get_all_explicit_arguments, \
    parse_string_to_psyneulink_object_string, parse_valid_identifier, safe_equals, convert_to_np_array

__all__ = [
    'MDFError', 'MDFSerializable', 'PNLJSONEncoder',
    'generate_json', 'generate_script_from_json', 'generate_script_from_mdf',
    'write_json_file', 'get_mdf_model', 'get_mdf_serialized', 'write_mdf_file'
]


# file extension to mdf common name
supported_formats = {
    'json': 'json',
    'yml': 'yaml',
    'yaml': 'yaml',
}


class MDFError(Exception):
    pass


class MDFSerializable:
    @property
    def json_summary(self):
        return self.as_mdf_model().to_json()

    @property
    def yaml_summary(self):
        return self.as_mdf_model().to_yaml()


# leaving this due to instructions in test_documentation_models
# (useful for exporting Composition results to JSON)
class PNLJSONEncoder(json.JSONEncoder):
    """
        A `JSONEncoder
        <https://docs.python.org/3/library/json.html#json.JSONEncoder>`_
        that parses `_dict_summary <Component._dict_summary>` output
        into a more JSON-friendly format.
    """
    def default(self, o):
        import modeci_mdf.mdf as mdf
        from psyneulink.core.components.component import Component, ComponentsMeta

        if isinstance(o, ComponentsMeta):
            return o.__name__
        elif isinstance(o, (type, types.BuiltinFunctionType)):
            if o.__module__ == 'builtins':
                # just give standard type, like float or int
                return f'{o.__name__}'
            elif o is numpy.ndarray:
                return f'{o.__module__}.array'
            else:
                # some builtin modules are internally "_module"
                # but are imported with "module"
                return f"{o.__module__.lstrip('_')}.{o.__name__}"
        elif isinstance(o, (enum.Enum, types.FunctionType, types.SimpleNamespace)):
            return str(o)
        elif isinstance(o, types.MethodType):
            return o.__qualname__
        elif o is NotImplemented:
            return None
        elif isinstance(o, Component):
            return o.name
        elif isinstance(o, SampleIterator):
            return f'{o.__class__.__name__}({repr(o.specification)})'
        elif isinstance(o, numpy.ndarray):
            try:
                return list(o)
            except TypeError:
                return o.item()
        elif isinstance(o, numpy.random.RandomState):
            return f'numpy.random.RandomState({o.seed})'
        elif isinstance(o, numpy.number):
            return o.item()
        elif isinstance(o, mdf.BaseWithId):
            return json.loads(o.to_json())

        try:
            return super().default(o)
        except TypeError:
            return str(o)


def _get_variable_parameter_name(obj):
    try:
        if obj.parameters.variable.mdf_name is not None:
            return obj.parameters.variable.mdf_name
    except AttributeError:
        pass

    return MODEL_SPEC_ID_MDF_VARIABLE


def _mdf_obj_from_dict(d):
    import modeci_mdf.mdf as mdf

    def _get_mdf_object(obj, cls_):
        try:
            model_id = obj['id']
        except KeyError:
            try:
                model_id = obj['metadata']['name']
            except KeyError:
                model_id = f'{cls_.__name__}_{time.perf_counter_ns()}'

        return cls_.from_dict({model_id: obj})

    for cls_name in mdf.__all__:
        cls_ = getattr(mdf, cls_name)
        if all([attr.name in d or attr.name in {'id', 'parameters'} for attr in cls_.__attrs_attrs__]):
            return _get_mdf_object(d, cls_)

    if 'function' in d and 'args' in d:
        return _get_mdf_object(d, mdf.Function)

    # nothing else seems to fit, try Function (unreliable)
    if 'value' in d:
        return _get_mdf_object(d, mdf.Function)

    return None


def _get_parameters_from_mdf_base_object(model, pnl_type):
    model_params = getattr(model, pnl_type._model_spec_id_parameters)

    if isinstance(model_params, list):
        parameters = {p.id: p.value for p in model_params}
    elif isinstance(model_params, dict):
        parameters = dict(model_params)
    else:
        parameters = {}

    return parameters


def _parse_component_type(model_obj):
    def get_pnl_component_type(s):
        from psyneulink.core.components.component import ComponentsMeta

        try:
            return getattr(psyneulink, s)
        except AttributeError:
            for o in dir(psyneulink):
                if s.lower() == o.lower():
                    o = getattr(psyneulink, o)
                    if isinstance(o, ComponentsMeta):
                        return o
            # if matching component not found, raise original exception
            raise

    type_str = None
    try:
        try:
            type_dict = model_obj.metadata[MODEL_SPEC_ID_TYPE]
        except AttributeError:
            # could be a dict specification
            type_str = model_obj[MODEL_SPEC_ID_METADATA][MODEL_SPEC_ID_TYPE]
    except (KeyError, TypeError):
        # specifically for functions the keyword is not 'type'
        type_str = model_obj.function

    if type_str is None:
        try:
            type_str = type_dict[MODEL_SPEC_ID_PSYNEULINK]
        except KeyError:
            # catch error outside of this function if necessary
            type_str = type_dict[MODEL_SPEC_ID_GENERIC]
        except TypeError:
            # actually a str
            type_str = type_dict
    elif isinstance(type_str, dict):
        if len(type_str) != 1:
            raise MDFError
        else:
            elem = list(type_str.keys())[0]
            # not a function_type: args dict
            if MODEL_SPEC_ID_METADATA in type_str[elem]:
                raise MDFError
            else:
                type_str = elem

    try:
        # gets the actual psyneulink type (Component, etc..) from the module
        return get_pnl_component_type(type_str)
    except (AttributeError, TypeError):
        pass

    try:
        from modeci_mdf.functions.standard import mdf_functions
        mdf_functions[type_str]['function']
    # remove import/module errors when modeci_mdf is a package
    except (ImportError, KeyError, ModuleNotFoundError):
        pass
    else:
        return f"modeci_mdf.functions.standard.mdf_functions['{type_str}']['function']"

    try:
        getattr(math, type_str)
    except (AttributeError, TypeError):
        pass
    else:
        return f'math.{type_str}'

    try:
        eval(type_str)
    except (TypeError, SyntaxError):
        pass
    except NameError:
        return type_str
    else:
        return type_str

    raise MDFError(f'Invalid type specified for MDF object: {model_obj}')


def _parse_parameter_value(value, component_identifiers=None, name=None, parent_parameters=None):
    import modeci_mdf.mdf as mdf

    if component_identifiers is None:
        component_identifiers = {}

    exec('import numpy')
    try:
        pnl_type = _parse_component_type(value)
    except (AttributeError, TypeError, MDFError):
        # ignore parameters that aren't components
        pnl_type = None

    if isinstance(value, list):
        new_val = [_parse_parameter_value(x, component_identifiers, name, parent_parameters) for x in value]

        # check for ParameterPort spec
        if (
            len(value) == 2
            and isinstance(value[0], (numbers.Number, numpy.ndarray))
            and isinstance(value[1], dict)
        ):
            # make tuple instead of list
            value = f"({', '.join([str(x) for x in new_val])})"
        else:
            value = f"[{', '.join([str(x) for x in new_val])}]"
    elif isinstance(value, dict):
        if (
            MODEL_SPEC_ID_PARAMETER_SOURCE in value
            and MODEL_SPEC_ID_PARAMETER_VALUE in value
        ):
            # handle ParameterPort spec
            try:
                value_type = eval(value[MODEL_SPEC_ID_TYPE])
            except Exception as e:
                raise MDFError(
                    'Invalid python type specified in MDF object: {0}'.format(
                        value[MODEL_SPEC_ID_TYPE]
                    )
                ) from e

            value = _parse_parameter_value(
                value[MODEL_SPEC_ID_PARAMETER_VALUE],
                component_identifiers,
                name,
                parent_parameters,
            )

            # handle tuples and numpy arrays, which both are dumped
            # as lists in JSON form
            if value_type is tuple:
                # convert list brackets to tuple brackets
                assert value[0] == '[' and value[-1] == ']'
                value = f'({value[1:-1]})'
            elif value_type is numpy.ndarray:
                value = f'{value[MODEL_SPEC_ID_TYPE]}({value})'
        elif MODEL_SPEC_ID_PARAMETER_INITIAL_VALUE in value:
            # is a stateful parameter with initial value
            value = _parse_parameter_value(
                value[MODEL_SPEC_ID_PARAMETER_INITIAL_VALUE],
                component_identifiers,
                name,
                parent_parameters
            )
        elif MODEL_SPEC_ID_PARAMETER_VALUE in value and pnl_type is None:
            # is a standard mdf Parameter class with value
            value = _parse_parameter_value(
                value[MODEL_SPEC_ID_PARAMETER_VALUE],
                component_identifiers,
                name,
                parent_parameters
            )
        else:
            if len(value) == 1:
                try:
                    identifier = list(value.keys())[0]
                except KeyError:
                    identifier = name

                mdf_object = value[identifier]
            else:
                try:
                    identifier = value['id']
                except KeyError:
                    identifier = name

                mdf_object = value

            # it is either a Component spec or just a plain dict
            if (
                identifier in component_identifiers
                and component_identifiers[identifier]
            ):
                # if this spec is already created as a node elsewhere,
                # then just use a reference
                value = identifier
            else:
                if not isinstance(mdf_object, mdf.Base):
                    mdf_object = _mdf_obj_from_dict(mdf_object)

                try:
                    value = _generate_component_string(
                        mdf_object,
                        component_identifiers,
                        component_name=identifier,
                        parent_parameters=parent_parameters
                    )
                except (AttributeError, MDFError, KeyError, TypeError):
                    # standard dict handling
                    value = '{{{0}}}'.format(
                        ', '.join([
                            '{0}: {1}'.format(
                                str(_parse_parameter_value(k, component_identifiers, name)),
                                str(_parse_parameter_value(v, component_identifiers, name))
                            )
                            for k, v in value.items()
                        ])
                    )

    elif isinstance(value, str):
        # handle pointer to parent's parameter value
        try:
            return _parse_parameter_value(parent_parameters[value])
        except (KeyError, TypeError):
            pass

        # handle reference to psyneulink object
        obj_string = parse_string_to_psyneulink_object_string(value)
        if obj_string is not None:
            return f'psyneulink.{obj_string}'

        # handle dill string
        try:
            dill_str = base64.decodebytes(bytes(value, 'utf-8'))
            dill.loads(dill_str)
            return f'dill.loads({dill_str})'
        except (binascii.Error, pickle.UnpicklingError, EOFError):
            pass

        # handle IO port specification
        match = re.match(r'(.+)\.(.+)_ports\.(.+)', value)
        if match is not None:
            comp_name, port_type, name = match.groups()
            comp_identifer = parse_valid_identifier(comp_name)

            if comp_identifer in component_identifiers:
                name_as_kw = parse_string_to_psyneulink_object_string(name)
                if name_as_kw is not None:
                    name = f'psyneulink.{name_as_kw}'
                else:
                    name = f"'{name}'"

                return f'{comp_identifer}.{port_type}_ports[{name}]'

        # if value is just a non-fixed component name, use the fixed name
        identifier = parse_valid_identifier(value)
        if identifier in component_identifiers:
            value = identifier

        try:
            psyneulink._unit_registry.Unit(value)
        except (AttributeError, TypeError, ValueError, pint.errors.DefinitionSyntaxError):
            pass
        else:
            value = f"'{value}'"

        try:
            psyneulink._unit_registry.Quantity(value)
        except (AttributeError, TypeError, ValueError, pint.errors.DefinitionSyntaxError):
            pass
        else:
            value = f"'{value}'"

        evaluates = False
        try:
            eval(value)
            evaluates = True
        except (TypeError, NameError, SyntaxError):
            pass

        # handle generic string
        if (
            value not in component_identifiers
            # assume a string that contains a dot is a command, not a raw
            # string, this is definitely imperfect and can't handle the
            # legitimate case, but don't know how to distinguish..
            and '.' not in value
            and not evaluates
        ):
            value = f"'{value}'"

    elif isinstance(value, mdf.Base):
        value = _generate_component_string(
            value,
            component_identifiers,
            component_name=value.id,
            parent_parameters=parent_parameters
        )

    return value


def _generate_component_string(
    component_model,
    component_identifiers,
    component_name=None,
    parent_parameters=None,
    assignment=False,
    default_type=None   # used if no PNL or generic types are specified
):
    from psyneulink.core.components.functions.function import Function_Base
    from psyneulink.core.components.functions.userdefinedfunction import UserDefinedFunction

    try:
        component_type = _parse_component_type(component_model)
    except AttributeError as e:
        # acceptable to exclude type currently
        if default_type is not None:
            component_type = default_type
        else:
            raise type(e)(
                f'{component_model} has no PNL or generic type and no '
                'default_type is specified'
            ) from e

    if component_name is None:
        name = component_model.id
    else:
        name = component_name
        try:
            assert component_name == component_model.id
        except KeyError:
            pass

    is_user_defined_function = False
    try:
        parameters = _get_parameters_from_mdf_base_object(component_model, component_type)
    except AttributeError:
        is_user_defined_function = True

    if is_user_defined_function or component_type is UserDefinedFunction:
        custom_func = component_type
        component_type = UserDefinedFunction
        parameters = _get_parameters_from_mdf_base_object(component_model, component_type)
        parameters['custom_function'] = f'{custom_func}'
        try:
            del component_model.metadata['custom_function']
        except KeyError:
            pass

    try:
        # args in function dict
        parameters.update(component_model.function[list(component_model.function.keys())[0]])
    except (AttributeError, KeyError):
        pass

    parameter_names = {}

    # TODO: remove this?
    # If there is a parameter that is the psyneulink identifier string
    # (as of this comment, 'pnl'), then expand these parameters as
    # normal ones. We don't check and expand for other
    # special strings here, because we assume that these are specific
    # to other modeling platforms.
    try:
        parameters.update(parameters[MODEL_SPEC_ID_PSYNEULINK])
        del parameters[MODEL_SPEC_ID_PSYNEULINK]
    except KeyError:
        pass

    try:
        functions = component_model.functions
    except AttributeError:
        try:
            functions = [_mdf_obj_from_dict(v) for k, v in component_model.metadata['functions'].items()]
        except KeyError:
            functions = None
        except AttributeError:
            functions = component_model.metadata['functions']

    # pnl objects only have one function unless specified in another way
    # than just "function"

    if functions is not None:
        dup_function_names = set([f.id for f in functions if f.id in component_identifiers])
        if len(dup_function_names) > 0:
            warnings.warn(
                f'Functions ({gen_friendly_comma_str(dup_function_names)}) of'
                f' {name} share names of mechanisms or compositions in this'
                ' model. This is likely to cause incorrect script reproduction.'
            )

        function_determined_by_output_port = False

        try:
            output_ports = component_model.output_ports
        except AttributeError:
            pass
        else:
            if len(output_ports) == 1 or isinstance(output_ports, list):
                try:
                    primary_output_port = output_ports[0]
                except KeyError:
                    primary_output_port = output_ports[list(output_ports)[0]]
                function_determined_by_output_port = True
            else:
                try:
                    # 'out_port' appears to be the general primary output_port term
                    # should ideally have a marker in mdf to define it as primary
                    primary_output_port = output_ports['out_port']
                except KeyError:
                    pass
                else:
                    function_determined_by_output_port = True

        # neuroml-style mdf has MODEL_SPEC_ID_PARAMETER_VALUE in output port definitions
        if function_determined_by_output_port and hasattr(primary_output_port, MODEL_SPEC_ID_PARAMETER_VALUE):
            parameter_names['function'] = re.sub(r'(.*)\[\d+\]', '\\1', getattr(primary_output_port, MODEL_SPEC_ID_PARAMETER_VALUE))
        else:
            parameter_names['function'] = [
                f.id for f in functions
                if not f.id.endswith(MODEL_SPEC_ID_INPUT_PORT_COMBINATION_FUNCTION)
            ][0]

        parameters['function'] = [f for f in functions if f.id == parameter_names['function']][0]

    assignment_str = f'{parse_valid_identifier(name)} = ' if assignment else ''

    additional_arguments = []
    # get the nonvariable arg and keyword arguments for the component's
    # constructor
    constructor_arguments = get_all_explicit_arguments(
        component_type,
        '__init__'
    )

    # put name as first argument
    if 'name' in constructor_arguments:
        additional_arguments.append(f"name='{name}'")

    if parent_parameters is None:
        parent_parameters = parameters

    parameters = {
        **{k: v for k, v in parent_parameters.items() if isinstance(v, dict) and MODEL_SPEC_ID_PARAMETER_INITIAL_VALUE in v},
        **parameters,
        **(component_model.metadata if component_model.metadata is not None else {})
    }

    # MDF input ports do not have functions, so their shape is
    # equivalent to ours after the InputPort function is run (this
    # function may change the shape of the default variable), so ignore
    # the input port shape if input_ports parameter is specified
    if 'variable' not in parameters and 'input_ports' not in parameters:
        try:
            ip = getattr(parameters['function'], Function_Base._model_spec_id_parameters)[MODEL_SPEC_ID_MDF_VARIABLE]
            var = convert_to_np_array(
                numpy.zeros(ast.literal_eval(component_model.input_ports[ip][MODEL_SPEC_ID_SHAPE])),
                dimension=2
            ).tolist()
            parameters['variable'] = var
        except KeyError:
            pass

    def parameter_value_matches_default(component_type, param, value):
        default_val = getattr(component_type.defaults, param)
        evaled_val = NotImplemented

        # see if val is a psyneulink class instantiation
        # if so, do not instantiate it (avoid offsetting rng for
        # testing - see if you can bypass another way?)
        try:
            eval(re.match(r'(psyneulink\.\w+)\(', value).group(1))
            is_pnl_instance = True
        except (AttributeError, TypeError, NameError, ValueError):
            is_pnl_instance = False

        if not is_pnl_instance:
            # val may be a string that evaluates to the default value
            # also skip listing in constructor in this case
            try:
                evaled_val = eval(value)
            except (TypeError, NameError, ValueError):
                pass
            except Exception:
                # Assume this occurred in creation of a Component
                # that probably needs some hidden/automatic modification.
                # Special handling here?
                # still relevant after testing for instance above?
                pass

        # skip specifying parameters that match the class defaults
        if (
            not safe_equals(value, default_val)
            and (
                evaled_val is NotImplemented
                or not safe_equals(evaled_val, default_val)
            )
        ):
            # test for dill use/equivalence
            try:
                is_dill_str = value[:5] == 'dill.'
            except TypeError:
                is_dill_str = False

            if (
                not is_dill_str
                or dill.dumps(eval(value)) != dill.dumps(default_val)
            ):
                return False

        return True

    mdf_names_to_pnl = {
        p.mdf_name: p.name for p in component_type.parameters
        if p.mdf_name is not None and not isinstance(p, ParameterAlias)
    }

    # sort on arg name
    for arg, val in sorted(parameters.items(), key=lambda p: p[0]):
        try:
            arg = mdf_names_to_pnl[arg]
        except KeyError:
            pass

        try:
            constructor_parameter_name = getattr(component_type.parameters, arg).constructor_argument
            # Some Parameters may be stored just to be replicated here, and
            # they may have different names than are used in the
            # constructor of the object.
            # Examples:
            #   Component.variable / default_variable
            #   ControlMechanism.output_ports / control
            if constructor_parameter_name is not None:
                constructor_arg = constructor_parameter_name
            else:
                constructor_arg = arg
        except AttributeError:
            constructor_arg = arg

        if constructor_arg in constructor_arguments:
            try:
                val = _parse_parameter_value(
                    val, component_identifiers,
                    name=parameter_names[arg],
                    parent_parameters=parent_parameters,
                )
            except KeyError:
                val = _parse_parameter_value(val, component_identifiers, parent_parameters=parent_parameters)

            if (
                (arg in component_type.parameters or constructor_arg in component_type.parameters)
                and not parameter_value_matches_default(component_type, arg, val)
            ):
                additional_arguments.append(f'{constructor_arg}={val}')
        elif component_type is UserDefinedFunction:
            try:
                val[MODEL_SPEC_ID_PARAMETER_INITIAL_VALUE]
            except (KeyError, TypeError):
                pass
            else:
                # is a stateful parameter corresponding to this function
                if val[MODEL_SPEC_ID_PARAMETER_VALUE] == name:
                    additional_arguments.append(f"stateful_parameter='{arg}'")

            if arg != MODEL_SPEC_ID_MDF_VARIABLE:
                val = _parse_parameter_value(
                    val, component_identifiers, parent_parameters=parent_parameters
                )

                try:
                    matches = parameter_value_matches_default(component_type, arg, val)
                except AttributeError:
                    matches = False

                if not matches:
                    additional_arguments.append(f'{constructor_arg}={val}')

    output = '{0}psyneulink.{1}{2}{3}{4}'.format(
        assignment_str,
        component_type.__name__,
        '(' if len(additional_arguments) > 0 else '',
        ', '.join(additional_arguments),
        ')' if len(additional_arguments) > 0 else '',
    )

    return output


def _generate_scheduler_string(
    scheduler_id,
    scheduler_model,
    component_identifiers,
    blacklist=[]
):
    output = []

    for node, condition in scheduler_model.node_specific.items():
        if node not in blacklist:
            output.append(
                '{0}.add_condition({1}, {2})'.format(
                    scheduler_id,
                    parse_valid_identifier(node),
                    _generate_condition_string(
                        condition,
                        component_identifiers
                    )
                )
            )

    termination_str = []
    for scale, cond in scheduler_model.termination.items():
        termination_str.insert(
            1,
            'psyneulink.{0}: {1}'.format(
                f'TimeScale.{str.upper(scale)}',
                _generate_condition_string(cond, component_identifiers)
            )
        )

    output.append(
        '{0}.termination_conds = {{{1}}}'.format(
            scheduler_id,
            ', '.join(termination_str)
        )
    )

    return '\n'.join(output)


def _generate_condition_string(condition_model, component_identifiers):
    def _parse_condition_arg_value(value):
        try:
            identifier = parse_valid_identifier(value)
        except TypeError:
            pass
        else:
            if identifier in component_identifiers:
                return str(identifier)

        try:
            getattr(psyneulink.core.scheduling.condition, value.type)
        except (AttributeError, KeyError, TypeError):
            pass
        else:
            return _generate_condition_string(value, component_identifiers)

        # handle value/outputport fix for threshold
        try:
            if re.match(r'\w+_OutputPort_0', value):
                return '"value"'
        except TypeError:
            pass

        return str(_parse_parameter_value(value, component_identifiers))

    def _parse_graph_scheduler_type(typ):
        for ts, pnl_ts in graph_scheduler.time._time_scale_aliases.items():
            ts_class_name = graph_scheduler.time._time_scale_to_class_str(ts)
            pnl_ts_class_name = graph_scheduler.time._time_scale_to_class_str(pnl_ts)

            if ts_class_name in typ:
                return typ.replace(ts_class_name, pnl_ts_class_name)

        return typ

    args_str = ''
    cond_type = _parse_graph_scheduler_type(condition_model.type)
    sig = inspect.signature(getattr(psyneulink, cond_type).__init__)

    var_positional_arg_name = None

    for name, param in sig.parameters.items():
        if param.kind is inspect.Parameter.VAR_POSITIONAL:
            var_positional_arg_name = name
            break

    args_dict = condition_model.kwargs

    try:
        pos_args = args_dict[var_positional_arg_name]
    except KeyError:
        pass
    else:
        if len(pos_args) > 0:
            arg_str_list = []
            for arg in pos_args:
                # handle nested Conditions
                try:
                    arg = _generate_condition_string(arg, component_identifiers)
                except TypeError:
                    pass

                arg_str_list.append(_parse_condition_arg_value(arg))
            args_str = f", {', '.join(arg_str_list)}"

    kwargs_str = ''
    kwargs = {k: v for k, v in args_dict.items() if k not in {'function', var_positional_arg_name}}
    if len(kwargs) > 0:
        kwarg_str_list = []
        for key, val in kwargs.items():
            kwarg_str_list.append(f'{key}={_parse_condition_arg_value(val)}')
        kwargs_str = f", {', '.join(kwarg_str_list)}"

    if 'function' in args_dict and args_dict['function'] is not None:
        func_str = args_dict['function']
    else:
        func_str = ''

    arguments_str = '{0}{1}{2}'.format(
        func_str,
        args_str,
        kwargs_str
    )
    if len(arguments_str) > 0 and arguments_str[0] == ',':
        arguments_str = arguments_str[2:]

    return f'psyneulink.{cond_type}({arguments_str})'


def _generate_composition_string(graph, component_identifiers):
    import modeci_mdf.mdf as mdf

    # used if no generic types are specified
    default_composition_type = psyneulink.Composition
    default_node_type = psyneulink.ProcessingMechanism
    default_edge_type = psyneulink.MappingProjection

    control_mechanism_types = (psyneulink.ControlMechanism, )
    # these are not actively added to a Composition
    implicit_types = (
        psyneulink.ObjectiveMechanism,
        psyneulink.ControlProjection,
        psyneulink.AutoAssociativeProjection,
        psyneulink.LearningMechanism,
        psyneulink.LearningProjection,
    )
    implicit_roles = (
        psyneulink.NodeRole.LEARNING,
    )

    output = []

    comp_identifer = parse_valid_identifier(graph.id)

    def alphabetical_order(items):
        alphabetical = enumerate(
            sorted(items)
        )
        return {
            parse_valid_identifier(item[1]): item[0]
            for item in alphabetical
        }

    # get order in which nodes were added
    # may be node names or dictionaries
    try:
        node_order = graph.metadata['node_ordering']
        node_order = {
            parse_valid_identifier(list(node.keys())[0]) if isinstance(node, dict)
            else parse_valid_identifier(node): node_order.index(node)
            for node in node_order
        }

        unspecified_node_order = {
            node: position + len(node_order)
            for node, position in alphabetical_order([
                parse_valid_identifier(n.id) for n in graph.nodes if n.id not in node_order
            ]).items()
        }

        node_order.update(unspecified_node_order)

        assert all([
            (parse_valid_identifier(node.id) in node_order)
            for node in graph.nodes
        ])
    except (KeyError, TypeError, AssertionError):
        # if no node_ordering attribute exists, fall back to
        # alphabetical order
        node_order = alphabetical_order([parse_valid_identifier(n.id) for n in graph.nodes])

    keys_to_delete = []

    for node in graph.nodes:
        try:
            component_type = _parse_component_type(node)
        except (AttributeError, KeyError):
            # will use a default type
            pass
        else:
            # projection was written out as a node for simple_edge_format
            if issubclass(component_type, psyneulink.Projection_Base):
                assert len(node.input_ports) == 1
                assert len(node.output_ports) == 1

                extra_projs_to_delete = set()

                sender = None
                sender_port = None
                receiver = None
                receiver_port = None

                for proj in graph.edges:
                    if proj.receiver == node.id:
                        assert 'dummy' in proj.id
                        sender = proj.sender
                        sender_port = proj.sender_port
                        extra_projs_to_delete.add(proj.id)

                    if proj.sender == node.id:
                        assert 'dummy' in proj.id
                        receiver = proj.receiver
                        receiver_port = proj.receiver_port
                        # if for some reason the projection has node as both sender and receiver
                        # this is a bug, let the deletion fail
                        extra_projs_to_delete.add(proj.id)

                if sender is None:
                    raise MDFError(f'Dummy node {node.id} for projection has no sender in projections list')

                if receiver is None:
                    raise MDFError(f'Dummy node {node.id} for projection has no receiver in projections list')

                main_proj = mdf.Edge(
                    id=node.id.rstrip('_dummy_node'),
                    sender=sender,
                    receiver=receiver,
                    sender_port=sender_port,
                    receiver_port=receiver_port,
                    metadata={
                        # variable isn't specified for projections
                        **{k: v for k, v in node.metadata.items() if k != 'variable'},
                        'functions': node.functions
                    }
                )
                proj.parameters = {p.id: p for p in node.parameters}
                graph.edges.append(main_proj)

                keys_to_delete.append(node.id)
                for p in extra_projs_to_delete:
                    del graph.edges[graph.edges.index([e for e in graph.edges if e.id == p][0])]

                for nr_item in ['required_node_roles', 'excluded_node_roles']:
                    nr_removal_indices = []

                    for i, (nr_name, nr_role) in enumerate(
                        graph.metadata[nr_item]
                    ):
                        if nr_name == node.id:
                            nr_removal_indices.append(i)

                    for i in nr_removal_indices:
                        del graph.metadata[nr_item][i]

    for name_to_delete in keys_to_delete:
        del graph.nodes[graph.nodes.index([n for n in graph.nodes if n.id == name_to_delete][0])]

    # generate string for Composition itself
    output.append(
        "{0} = {1}\n".format(
            comp_identifer,
            _generate_component_string(
                graph,
                component_identifiers,
                component_name=graph.id,
                default_type=default_composition_type
            )
        )
    )
    component_identifiers[comp_identifer] = True

    mechanisms = []
    compositions = []
    control_mechanisms = []
    implicit_mechanisms = []

    try:
        node_roles = {
            parse_valid_identifier(node): role for (node, role) in
            graph.metadata['required_node_roles']
        }
    except KeyError:
        node_roles = []

    try:
        excluded_node_roles = {
            parse_valid_identifier(node): role for (node, role) in
            graph.metadata['excluded_node_roles']
        }
    except KeyError:
        excluded_node_roles = []

    # add nested compositions and mechanisms in order they were added
    # to this composition
    for node in sorted(
        graph.nodes,
        key=lambda item: node_order[parse_valid_identifier(item.id)]
    ):
        if isinstance(node, mdf.Graph):
            compositions.append(node)
        else:
            try:
                component_type = _parse_component_type(node)
            except (AttributeError, KeyError):
                component_type = default_node_type
            identifier = parse_valid_identifier(node.id)

            try:
                node_role = eval(_parse_parameter_value(node_roles[identifier]))
            except (KeyError, TypeError):
                node_role = None

            if issubclass(component_type, control_mechanism_types):
                control_mechanisms.append(node)
                component_identifiers[identifier] = True
            elif (
                issubclass(component_type, implicit_types)
                or node_role in implicit_roles
            ):
                implicit_mechanisms.append(node)
            else:
                mechanisms.append(node)
                component_identifiers[identifier] = True

    implicit_names = [node.id for node in implicit_mechanisms + control_mechanisms]

    for mech in mechanisms:
        try:
            mech_type = _parse_component_type(mech)
        except (AttributeError, KeyError):
            mech_type = None

        if (
            isinstance(mech_type, type)
            and issubclass(mech_type, psyneulink.Function)
        ):
            # removed branch converting functions defined as nodes
            # should no longer happen with recent MDF versions
            assert False

        output.append(
            _generate_component_string(
                mech,
                component_identifiers,
                component_name=parse_valid_identifier(mech.id),
                assignment=True,
                default_type=default_node_type
            )
        )
    if len(mechanisms) > 0:
        output.append('')

    for mech in control_mechanisms:
        output.append(
            _generate_component_string(
                mech,
                component_identifiers,
                component_name=parse_valid_identifier(mech.id),
                assignment=True,
                default_type=default_node_type
            )
        )

    if len(control_mechanisms) > 0:
        output.append('')

    # recursively generate string for inner Compositions
    for comp in compositions:
        output.append(
            _generate_composition_string(
                comp,
                component_identifiers
            )
        )
    if len(compositions) > 0:
        output.append('')

    # do not add the controller as a normal node
    try:
        controller_name = graph.metadata['controller']['id']
    except (AttributeError, KeyError, TypeError):
        controller_name = None

    for node in sorted(
        graph.nodes,
        key=lambda item: node_order[parse_valid_identifier(item.id)]
    ):
        name = node.id
        if (
            name not in implicit_names
            and name != controller_name
        ):
            name = parse_valid_identifier(name)

            output.append(
                '{0}.add_node({1}{2})'.format(
                    comp_identifer,
                    name,
                    ', {0}'.format(
                        _parse_parameter_value(
                            node_roles[name],
                            component_identifiers
                        )
                    ) if name in node_roles else ''
                )
            )
    if len(graph.nodes) > 0:
        output.append('')

    if len(excluded_node_roles) > 0:
        for node, roles in excluded_node_roles.items():
            if name not in implicit_names and name != controller_name:
                output.append(
                    f'{comp_identifer}.exclude_node_roles({node}, {_parse_parameter_value(roles, component_identifiers)})'
                )
        output.append('')

    # generate string to add the projections
    for proj in graph.edges:
        try:
            projection_type = _parse_component_type(proj)
        except (AttributeError, KeyError):
            projection_type = default_edge_type

        if (
            not issubclass(projection_type, implicit_types)
            and proj.sender not in implicit_names
            and proj.receiver not in implicit_names
        ):
            output.append(
                '{0}.add_projection(projection={1}, sender={2}, receiver={3})'.format(
                    comp_identifer,
                    _generate_component_string(
                        proj,
                        component_identifiers,
                        default_type=default_edge_type
                    ),
                    parse_valid_identifier(proj.sender),
                    parse_valid_identifier(proj.receiver),
                )
            )

    # add controller if it exists (must happen after projections)
    if controller_name is not None:
        output.append(
            '{0}.add_controller({1})'.format(
                comp_identifer,
                parse_valid_identifier(controller_name)
            )
        )

    # add schedulers
    # blacklist automatically generated nodes because they will
    # not exist in the script namespace
    output.append('')
    output.append(
        _generate_scheduler_string(
            f'{comp_identifer}.scheduler',
            graph.conditions,
            component_identifiers,
            blacklist=implicit_names
        )
    )

    return output


def generate_script_from_json(model_input, outfile=None):
    """
        Generate a Python script from JSON **model_input** in the
        `general JSON format <JSON_Model_Specification>`

        .. warning::
           Use of `generate_script_from_json` to generate a Python script from a file without taking proper precautions
           can introduce a security risk to the system on which the Python interpreter is running.  This is because it
           calls exec, which has the potential to execute non-PsyNeuLink-related code embedded in the file.  Therefore,
           `generate_script_from_json` should be used to read only files of known and secure origin.

        Arguments
        ---------

            model_input : str
                a JSON string in the proper format, or a filename
                containing such

        Returns
        -------

            Text of Python script : str



    """
    warnings.warn(
        'generate_script_from_json is replaced by generate_script_from_mdf and will be removed in a future version',
        FutureWarning
    )
    return generate_script_from_mdf(model_input, outfile)


def generate_script_from_mdf(model_input, outfile=None):
    """
        Generate a Python script from MDF model **model_input**

        .. warning::
           Use of `generate_script_from_mdf` to generate a Python script from a model without taking proper precautions
           can introduce a security risk to the system on which the Python interpreter is running.  This is because it
           calls exec, which has the potential to execute non-PsyNeuLink-related code embedded in the file.  Therefore,
           `generate_script_from_mdf` should be used to read only model of known and secure origin.

        Arguments
        ---------

            model_input : modeci_mdf.Model

        Returns
        -------

            Text of Python script : str
    """
    import modeci_mdf.mdf as mdf
    from modeci_mdf.utils import load_mdf

    def get_declared_identifiers(model):
        names = set()

        for graph in model.graphs:
            names.add(parse_valid_identifier(graph.id))
            for node in graph.nodes:
                if isinstance(node, mdf.Graph):
                    names.update(get_declared_identifiers(graph))

                names.add(parse_valid_identifier(node.id))

        return names

    # accept either json string or filename
    try:
        model = load_mdf(model_input)
    except (FileNotFoundError, OSError, ValueError):
        try:
            model = mdf.Model.from_json(model_input)
        except json.decoder.JSONDecodeError:
            # assume yaml
            # delete=False because of problems with reading file on windows
            with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
                f.write(model_input)
                model = load_mdf(f.name)

    imports_str = ''
    comp_strs = []
    # maps declared names to whether they are accessible in the script
    # locals. that is, each of these will be names specified in the
    # composition and subcomposition nodes, and their value in this dict
    # will correspond to True if they can be referenced by this name in the
    # script
    component_identifiers = {
        i: False
        for i in get_declared_identifiers(model)
    }

    for graph in model.graphs:
        comp_strs.append(_generate_composition_string(graph, component_identifiers))

    module_friendly_name_mapping = {
        'psyneulink': 'pnl',
        'dill': 'dill',
        'numpy': 'np'
    }

    potential_module_names = set()
    module_names = set()
    model_output = []

    for i in range(len(comp_strs)):
        # greedy and non-greedy
        for cs in comp_strs[i]:
            cs_potential_names = set([
                *re.findall(r'([A-Za-z_\.]+)\.', cs),
                *re.findall(r'([A-Za-z_\.]+?)\.', cs)
            ])
            potential_module_names.update(cs_potential_names)

        for module in potential_module_names:
            if module not in component_identifiers:
                try:
                    exec(f'import {module}')
                    module_names.add(module)
                except (ImportError, ModuleNotFoundError, SyntaxError):
                    pass

        for j in range(len(comp_strs[i])):
            for module in module_names.copy():
                try:
                    friendly_name = module_friendly_name_mapping[module]
                    comp_strs[i][j] = re.sub(f'{module}\\.', f'{friendly_name}.', comp_strs[i][j])
                except KeyError:
                    pass

        for m in module_names.copy():
            for n in module_names.copy():
                # remove potential modules that are substrings of another
                if m is not n and m in n:
                    module_names.remove(m)

        for module in sorted(module_names):
            try:
                friendly_name = module_friendly_name_mapping[module]
            except KeyError:
                friendly_name = module

            imports_str += 'import {0}{1}\n'.format(
                module,
                f' as {friendly_name}' if friendly_name != module else ''
            )

        comp_strs[i] = '\n'.join(comp_strs[i])

    model_output = '{0}{1}{2}'.format(
        imports_str,
        '\n' if len(imports_str) > 0 else '',
        '\n'.join(comp_strs)
    )

    if outfile is not None:
        # pass through any file exceptions
        with open(outfile, 'w') as outfile:
            outfile.write(model_output)
            print(f'Wrote script to {outfile.name}')
    else:
        return model_output


def generate_json(*compositions, simple_edge_format=True):
    """
        Generate the `general JSON format <JSON_Model_Specification>`
        for one or more `Compositions <Composition>` and associated
        objects.
        .. _MDF_Write_Multiple_Compositions_Note:

        .. note::
           At present, if more than one Composition is specified, all
           must be fully disjoint;  that is, they must not share any
           `Components <Component>` (e.g., `Mechanism`, `Projections`
           etc.). This limitation will be addressed in a future update.

        Arguments:
            *compositions : Composition
                specifies `Composition` or iterable of ones to be output
                in JSON
    """
    warnings.warn(
        'generate_json is replaced by get_mdf_serialized and will be removed in a future version',
        FutureWarning
    )
    return get_mdf_serialized(*compositions, fmt='json', simple_edge_format=simple_edge_format)


def get_mdf_serialized(*compositions, fmt='json', simple_edge_format=True):
    """
        Generate the `general MDF serialized format <JSON_Model_Specification>`
        for one or more `Compositions <Composition>` and associated
        objects.

        .. note::
           At present, if more than one Composition is specified, all
           must be fully disjoint;  that is, they must not share any
           `Components <Component>` (e.g., `Mechanism`, `Projections`
           etc.). This limitation will be addressed in a future update.

        Arguments:
            *compositions : Composition
                specifies `Composition` or iterable of ones to be output
                in **fmt**

            fmt : str
                specifies file format of output. Current options ('json', 'yml'/'yaml')

            simple_edge_format : bool
                specifies use of
                `simple edge format <MDF_Simple_Edge_Format>` or not
    """
    model = get_mdf_model(*compositions, simple_edge_format=simple_edge_format)

    try:
        return getattr(model, f'to_{supported_formats[fmt]}')()
    except AttributeError as e:
        raise ValueError(
            f'Unsupported MDF output format "{fmt}". Supported formats: {gen_friendly_comma_str(supported_formats.keys())}'
        ) from e


def write_json_file(compositions, filename:str, path:str=None, simple_edge_format=True):
    """
        Write one or more `Compositions <Composition>` and associated objects to file in the `general JSON format
        <JSON_Model_Specification>`

        .. _MDF_Write_Multiple_Compositions_Note:

        .. note::
           At present, if more than one Composition is specified, all must be fully disjoint;  that is, they must not
           share  any `Components <Component>` (e.g., `Mechanism`, `Projections` etc.).  This limitation will be
           addressed in a future update.

        Arguments
        ---------
        compositions : Composition or list˚
             specifies `Composition` or list of ones to be written to **filename**

        filename : str
             specifies name of file in which to write JSON specification of `Composition(s) <Composition>`
             and associated objects.

        path : str : default None
             specifies path of file for JSON specification;  if it is not specified then the current directory is used.

    """
    warnings.warn(
        'write_json_file is replaced by write_mdf_file and will be removed in a future version',
        FutureWarning
    )
    write_mdf_file(compositions, filename, path, 'json', simple_edge_format)


def write_mdf_file(compositions, filename: str, path: str = None, fmt: str = None, simple_edge_format: bool = True):
    """
        Write the `general MDF serialized format <MDF_Model_Specification>`
        for one or more `Compositions <Composition>` and associated
        objects to file.

        .. note::
           At present, if more than one Composition is specified, all
           must be fully disjoint;  that is, they must not share any
           `Components <Component>` (e.g., `Mechanism`, `Projections`
           etc.). This limitation will be addressed in a future update.

        Arguments:
            compositions : Composition or list
                specifies `Composition` or list of ones to be written to
                **filename**

            filename : str
                specifies name of file in which to write MDF
                specification of `Composition(s) <Composition>` and
                associated objects.

            path : str : default None
                specifies path of file for MDF specification; if it is
                not specified then the current directory is used.

            fmt : str
                specifies file format of output. Auto-detect based on
                **filename** extension if None.
                Current options: 'json', 'yml'/'yaml'

            simple_edge_format : bool
                specifies use of
                `simple edge format <MDF_Simple_Edge_Format>` or not
    """
    compositions = convert_to_list(compositions)
    model = get_mdf_model(*compositions, simple_edge_format=simple_edge_format)

    if fmt is None:
        try:
            fmt = re.match(r'(.*)\.(.*)$', filename).groups()[1]
        except (AttributeError, IndexError):
            fmt = 'json'

    if path is not None:
        filename = os.path.join(path, filename)

    try:
        return getattr(model, f'to_{supported_formats[fmt]}_file')(filename)
    except AttributeError as e:
        raise ValueError(
            f'Unsupported MDF output format "{fmt}". Supported formats: {gen_friendly_comma_str(supported_formats.keys())}'
        ) from e


def get_mdf_model(*compositions, simple_edge_format=True):
    """
        Generate the MDF Model object for one or more
        `Compositions <Composition>` and associated objects.

        .. note::
           At present, if more than one Composition is specified, all
           must be fully disjoint;  that is, they must not share any
           `Components <Component>` (e.g., `Mechanism`, `Projections`
           etc.). This limitation will be addressed in a future update.

        Arguments:
            *compositions : Composition
                specifies `Composition` or iterable of ones to be output
                in the Model

            simple_edge_format : bool
                specifies use of
                `simple edge format <MDF_Simple_Edge_Format>` or not
    """
    import modeci_mdf
    import modeci_mdf.mdf as mdf
    from psyneulink.core.compositions.composition import Composition

    model_name = "_".join([c.name for c in compositions])

    model = mdf.Model(
        id=model_name,
        format=f'ModECI MDF v{modeci_mdf.__version__}',
        generating_application=f'PsyNeuLink v{psyneulink.__version__}',
    )

    for c in compositions:
        if not isinstance(c, Composition):
            raise MDFError(
                f'Item in compositions arg of {__name__}() is not a Composition: {c}.'
            )
        model.graphs.append(c.as_mdf_model(simple_edge_format=simple_edge_format))

    return model

PrincetonUniversity / PsyNeuLink / 11447759208

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous