text:
# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License").
# You may not use this file except in compliance with the License.
# A copy of the License is located at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# or in the "license" file accompanying this file. This file 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.

# Standard library imports
import importlib
import itertools
import json
import math
import pickle
import re
import textwrap
from functools import singledispatch
from pathlib import Path
from pydoc import locate
from typing import Any, Optional

# Third-party imports
import mxnet as mx
import numpy as np
import pandas as pd
from pydantic import BaseModel

# Relative imports
from gluonts.core import fqname_for

bad_type_msg = textwrap.dedent(
    """
    Cannot serialize type {}. See the documentation of the `encode` and
    `validate` functions at

        http://gluon-ts.mxnet.io/api/gluonts/gluonts.html

    and the Python documentation of the `__getnewargs_ex__` magic method at

        https://docs.python.org/3/library/pickle.html#object.__getnewargs_ex__

    for more information how to make this type serializable.
    """
).lstrip()

# Binary Serialization/Deserialization
# ------------------------------------

def dump_binary(o: Any) -> bytes:
    """
    Serializes an object ``o`` to binary format.

    Parameters
    ----------
    o
        The object to serialize.

    Returns
    -------
    bytes
        A sequence of bytes representing the serialized object.

    See Also
    --------
    load_binary
        Inverse function.
    """
    return pickle.dumps(o)

def load_binary(b: bytes) -> Any:
    """
    Deserializes an object from binary format.

    Parameters
    ----------
    b
        A sequence of bytes representing the serialized object.

    Returns
    -------
    Any
        The deserialized object.

    See Also
    --------
    dump_binary
        Inverse function.
    """
    return pickle.loads(b)

# JSON Serialization/Deserialization
# ----------------------------------

# The canonical way to do this is to define and `default` and `object_hook`
# parameters to the json.dumps and json.loads methods. Unfortunately, due
# to https://bugs.python.org/issue12657 this is not possible at the moment,
# as support for custom NamedTuple serialization is broken.
#
# To circumvent the issue, we pass the input value through custom encode
# and decode functions that map nested object terms to JSON-serializable
# data structures with explicit recursion.

def dump_json(o: Any, indent: Optional[int] = None) -> str:
    """
    Serializes an object to a JSON string.

    Parameters
    ----------
    o
        The object to serialize.
    indent
        An optional number of spaced to use as an indent.

    Returns
    -------
    str
        A string representing the object in JSON format.

    See Also
    --------
    load_json
        Inverse function.
    """
    return json.dumps(encode(o), indent=indent, sort_keys=True)

def load_json(s: str) -> Any:
    """
    Deserializes an object from a JSON string.

    Parameters
    ----------
    s
        A string representing the object in JSON format.

    Returns
    -------
    Any
        The deserialized object.

    See Also
    --------
    dump_json
        Inverse function.
    """
    return decode(json.loads(s))

# Code Serialization/Deserialization
# ----------------------------------

def dump_code(o: Any) -> str:
    """
    Serializes an object to a Python code string.

    Parameters
    ----------
    o
        The object to serialize.

    Returns
    -------
    str
        A string representing the object as Python code.

    See Also
    --------
    load_code
        Inverse function.
    """

    def _dump_code(x: Any) -> str:
        # r = { 'class': ..., 'args': ... }
        # r = { 'class': ..., 'kwargs': ... }
        if type(x) == dict and x.get("__kind__") == kind_inst:
            args = x["args"] if "args" in x else []
            kwargs = x["kwargs"] if "kwargs" in x else {}
            return "{fqname}({bindings})".format(
                fqname=x["class"],
                bindings=", ".join(
                    itertools.chain(
                        [_dump_code(v) for v in args],
                        [f"{k}={_dump_code(v)}" for k, v in kwargs.items()],
                    )
                ),
            )
        if type(x) == dict and x.get("__kind__") == kind_type:
            return x["class"]
        if isinstance(x, dict):
            elems = [f"{_dump_code(k)}: {_dump_code(v)}" for k, v in x.items()]
            return "{" + ", ".join(elems) + "}"
        elif isinstance(x, list):
            elems = [dump_code(v) for v in x]
            return "[" + ", ".join(elems) + "]"
        elif isinstance(x, tuple):
            elems = [dump_code(v) for v in x]
            return "(" + ", ".join(elems) + ",)"
        elif isinstance(x, str):
            return '"' + x + '"'  # TODO: escape comp characters
        elif isinstance(x, float) or np.issubdtype(type(x), np.inexact):
            return str(x) if math.isfinite(x) else 'float("' + str(x) + '")'
        elif (
            isinstance(x, int)
            or np.issubdtype(type(x), np.integer)
            or x is None
        ):
            return str(x)
        else:
            x = fqname_for(x.__class__)
            raise RuntimeError(f"Unexpected element type {x}")

    return _dump_code(encode(o))

def load_code(c: str) -> Any:
    """
    Deserializes an object from a Python code string.

    Parameters
    ----------
    c
        A string representing the object as Python code.

    Returns
    -------
    Any
        The deserialized object.

    See Also
    --------
    dump_code
        Inverse function.
    """

    def _load_code(code: str, modules=None):
        if modules is None:
            modules = {}
        try:
            return eval(code, modules)
        except NameError as e:
            m = re.match(r"name '(?P<module>.+)' is not defined", str(e))
            if m:
                name = m["module"]
                return _load_code(
                    code,
                    {**(modules or {}), name: importlib.import_module(name)},
                )
            else:
                raise e
        except AttributeError as e:
            m = re.match(
                r"module '(?P<module>.+)' has no attribute '(?P<package>.+)'",
                str(e),
            )
            if m:
                name = m["module"] + "." + m["package"]
                return _load_code(
                    code,
                    {**(modules or {}), name: importlib.import_module(name)},
                )
            else:
                raise e
        except Exception as e:
            raise e

    return _load_code(c)

# Structural encoding/decoding
# ----------------------------

kind_type = "type"
kind_inst = "instance"

@singledispatch
def encode(v: Any) -> Any:
    """
    Transforms a value `v` as a serializable intermediate representation (for
    example, named tuples are encoded as dictionaries). The intermediate
    representation is then recursively traversed and serialized either as
    Python code or as JSON string.

    This function is decorated with :func:`~functools.singledispatch` and can
    be specialized by clients for families of types that are not supported by
    the basic implementation (explained below).

    Examples
    --------

    The conversion logic implemented by the basic implementation is used
    as a fallback and is best explained by a series of examples.

    Lists (as lists).

    >>> encode([1, 2.0, '3'])
    [1, 2.0, '3']

    Tuples (as lists).

    >>> encode((1, 2.0, '3'))
    [1, 2.0, '3']

    Dictionaries (as dictionaries).

    >>> encode({'a': 1, 'b': 2.0, 'c': '3'})
    {'a': 1, 'b': 2.0, 'c': '3'}

    Named tuples (as dictionaries with a ``'__kind__': 'instance'`` member).

    >>> from pprint import pprint
    >>> from typing import NamedTuple
    >>> class ComplexNumber(NamedTuple):
    ...     x: float = 0.0
    ...     y: float = 0.0
    >>> pprint(encode(ComplexNumber(4.0, 2.0)))
    {'__kind__': 'instance',
     'class': 'gluonts.core.serde.ComplexNumber',
     'kwargs': {'x': 4.0, 'y': 2.0}}

    Classes with a :func:`~gluonts.core.component.validated` initializer (as
    dictionaries with a ``'__kind__': 'instance'`` member).

    >>> from gluonts.core.component import validated
    >>> class ComplexNumber:
    ...     @validated()
    ...     def __init__(self, x: float = 0.0, y: float = 0.0) -> None:
    ...         self.x = x
    ...         self.y = y
    >>> pprint(encode(ComplexNumber(4.0, 2.0)))
    {'__kind__': 'instance',
     'args': [],
     'class': 'gluonts.core.serde.ComplexNumber',
     'kwargs': {'x': 4.0, 'y': 2.0}}

    Classes with a ``__getnewargs_ex__`` magic method (as dictionaries with a
    ``'__kind__': 'instance'`` member).

    >>> from gluonts.core.component import validated
    >>> class ComplexNumber:
    ...     def __init__(self, x: float = 0.0, y: float = 0.0) -> None:
    ...         self.x = x
    ...         self.y = y
    ...     def __getnewargs_ex__(self):
    ...         return [], {'x': self.x, 'y': self.y}
    >>> pprint(encode(ComplexNumber(4.0, 2.0)))
    {'__kind__': 'instance',
     'args': [],
     'class': 'gluonts.core.serde.ComplexNumber',
     'kwargs': {'x': 4.0, 'y': 2.0}}

    Types (as dictionaries with a ``'__kind__': 'type' member``).

    >>> encode(ComplexNumber)
    {'__kind__': 'type', 'class': 'gluonts.core.serde.ComplexNumber'}

    Parameters
    ----------
    v
        The value to be encoded.

    Returns
    -------
    Any
        An encoding of ``v`` that can be serialized to Python code or
        JSON string.

    See Also
    --------
    decode
        Inverse function.
    dump_json
        Serializes an object to a JSON string.
    dump_code
        Serializes an object to a Python code string.
    """
    if isinstance(v, type(None)):
        return None
    elif isinstance(v, (float, int, str)):
        return v
    elif np.issubdtype(type(v), np.inexact):
        return float(v)
    elif np.issubdtype(type(v), np.integer):
        return int(v)
    elif isinstance(v, (list, set)) or type(v) == tuple:
        return [encode(v) for v in v]
    elif isinstance(v, tuple) and not hasattr(v, "_asdict"):
        return tuple([encode(v) for v in v])
    elif isinstance(v, dict):
        return {k: encode(v) for k, v in v.items()}
    elif isinstance(v, type):
        return {"__kind__": kind_type, "class": fqname_for(v)}
    elif isinstance(v, tuple) and hasattr(v, "_asdict"):
        return {
            "__kind__": kind_inst,
            "class": fqname_for(v.__class__),
            "kwargs": encode(getattr(v, "_asdict")()),
        }
    elif hasattr(v, "__getnewargs_ex__"):
        args, kwargs = getattr(v, "__getnewargs_ex__")()
        return {
            "__kind__": kind_inst,
            "class": fqname_for(v.__class__),
            "args": encode(args),
            "kwargs": encode(kwargs),
        }
    else:
        raise RuntimeError(bad_type_msg.format(fqname_for(v.__class__)))

@encode.register(Path)
def encode_path(v: Path) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~Path` class.
    """
    return {
        "__kind__": kind_inst,
        "class": fqname_for(v.__class__),
        "args": encode([str(v)]),
    }

@encode.register(BaseModel)
def encode_pydantic_model(v: BaseModel) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~BaseModel` class.
    """
    return {
        "__kind__": kind_inst,
        "class": fqname_for(v.__class__),
        "kwargs": encode(v.__values__),
    }

@encode.register(mx.Context)
def encode_mx_context(v: mx.Context) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~mxnet.Context` class.
    """
    return {
        "__kind__": kind_inst,
        "class": fqname_for(v.__class__),
        "args": encode([v.device_type, v.device_id]),
    }

@encode.register(np.ndarray)
def encode_np_ndarray(v: np.ndarray) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~mxnet.Context` class.
    """
    return {
        "__kind__": kind_inst,
        "class": "numpy.array",  # use "array" ctor instead of "nparray" class
        "args": encode([v.tolist(), v.dtype]),
    }

@encode.register(pd.Timestamp)
def encode_pd_timestamp(v: pd.Timestamp) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~pandas.Timestamp` class.
    """
    return {
        "__kind__": kind_inst,
        "class": "pandas.Timestamp",
        "args": encode([str(v)]),
        "kwargs": {"freq": v.freqstr if v.freq else None},
    }

@encode.register(np.dtype)
def encode_np_dtype(v: np.dtype) -> Any:
    """
    Specializes :func:`encode` for invocations where ``v`` is an instance of
    the :class:`~mxnet.Context` class.
    """
    return {
        "__kind__": kind_inst,
        "class": fqname_for(v.__class__),
        "args": encode([v.name]),
    }

@encode.register(mx.nd.NDArray)
def encode_mx_ndarray(v: mx.nd.NDArray) -> Any:
    return {
        "__kind__": kind_inst,
        "class": "mxnet.nd.array",
        "args": encode([v.asnumpy().tolist()]),
        "kwargs": {"dtype": encode(v.dtype)},
    }

def decode(r: Any) -> Any:
    """
    Decodes a value from an intermediate representation `r`.

    Parameters
    ----------
    r
        An intermediate representation to be decoded.

    Returns
    -------
    Any
        A Python data structure corresponding to the decoded version of ``r``.

    See Also
    --------
    encode
        Inverse function.
    """

    # structural recursion over the possible shapes of r
    # r = { 'class': ..., 'args': ... }
    # r = { 'class': ..., 'kwargs': ... }
    if type(r) == dict and r.get("__kind__") == kind_inst:
        cls = locate(r["class"])
        args = decode(r["args"]) if "args" in r else []
        kwargs = decode(r["kwargs"]) if "kwargs" in r else {}
        return cls(*args, **kwargs)
    # r = { 'class': ..., 'args': ... }
    # r = { 'class': ..., 'kwargs': ... }
    if type(r) == dict and r.get("__kind__") == kind_type:
        return locate(r["class"])
    # r = { k1: v1, ..., kn: vn }
    elif type(r) == dict:
        return {k: decode(v) for k, v in r.items()}
    # r = ( y1, ..., yn )
    elif type(r) == tuple:
        return tuple([decode(y) for y in r])
    # r = [ y1, ..., yn ]
    elif type(r) == list:
        return [decode(y) for y in r]
    # r = { y1, ..., yn }
    elif type(r) == set:
        return {decode(y) for y in r}
    # r = a
    else:
        return r

max_stars_repo_name: FadhelA/gluon-ts


text:
# Copyright 2020 Huawei Technologies Co., Ltd.
#
# Licensed 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.
"""
Testing FiveCrop in DE
"""
import pytest
import numpy as np

import mindspore.dataset as ds
import mindspore.dataset.transforms.vision.py_transforms as vision
from mindspore import log as logger
from util import visualize_list, save_and_check_md5

DATA_DIR = ["../data/dataset/test_tf_file_3_images/train-0000-of-0001.data"]
SCHEMA_DIR = "../data/dataset/test_tf_file_3_images/datasetSchema.json"

GENERATE_GOLDEN = False

def test_five_crop_op(plot=False):
    """
    Test FiveCrop
    """
    logger.info("test_five_crop")

    # First dataset
    data1 = ds.TFRecordDataset(DATA_DIR, SCHEMA_DIR, columns_list=["image"], shuffle=False)
    transforms_1 = [
        vision.Decode(),
        vision.ToTensor(),
    ]
    transform_1 = vision.ComposeOp(transforms_1)
    data1 = data1.map(input_columns=["image"], operations=transform_1())

    # Second dataset
    data2 = ds.TFRecordDataset(DATA_DIR, SCHEMA_DIR, columns_list=["image"], shuffle=False)
    transforms_2 = [
        vision.Decode(),
        vision.FiveCrop(200),
        lambda images: np.stack([vision.ToTensor()(image) for image in images])  # 4D stack of 5 images
    ]
    transform_2 = vision.ComposeOp(transforms_2)
    data2 = data2.map(input_columns=["image"], operations=transform_2())

    num_iter = 0
    for item1, item2 in zip(data1.create_dict_iterator(), data2.create_dict_iterator()):
        num_iter += 1
        image_1 = (item1["image"].transpose(1, 2, 0) * 255).astype(np.uint8)
        image_2 = item2["image"]

        logger.info("shape of image_1: {}".format(image_1.shape))
        logger.info("shape of image_2: {}".format(image_2.shape))

        logger.info("dtype of image_1: {}".format(image_1.dtype))
        logger.info("dtype of image_2: {}".format(image_2.dtype))
        if plot:
            visualize_list(np.array([image_1]*5), (image_2 * 255).astype(np.uint8).transpose(0, 2, 3, 1))

        # The output data should be of a 4D tensor shape, a stack of 5 images.
        assert len(image_2.shape) == 4
        assert image_2.shape[0] == 5

def test_five_crop_error_msg():
    """
    Test FiveCrop error message.
    """
    logger.info("test_five_crop_error_msg")

    data = ds.TFRecordDataset(DATA_DIR, SCHEMA_DIR, columns_list=["image"], shuffle=False)
    transforms = [
        vision.Decode(),
        vision.FiveCrop(200),
        vision.ToTensor()
    ]
    transform = vision.ComposeOp(transforms)
    data = data.map(input_columns=["image"], operations=transform())

    with pytest.raises(RuntimeError) as info:
        data.create_tuple_iterator().__next__()
    error_msg = "TypeError: img should be PIL Image or Numpy array. Got <class 'tuple'>"

    # error msg comes from ToTensor()
    assert error_msg in str(info.value)

def test_five_crop_md5():
    """
    Test FiveCrop with md5 check
    """
    logger.info("test_five_crop_md5")

    # First dataset
    data = ds.TFRecordDataset(DATA_DIR, SCHEMA_DIR, columns_list=["image"], shuffle=False)
    transforms = [
        vision.Decode(),
        vision.FiveCrop(100),
        lambda images: np.stack([vision.ToTensor()(image) for image in images])  # 4D stack of 5 images
    ]
    transform = vision.ComposeOp(transforms)
    data = data.map(input_columns=["image"], operations=transform())
    # Compare with expected md5 from images
    filename = "five_crop_01_result.npz"
    save_and_check_md5(data, filename, generate_golden=GENERATE_GOLDEN)

if __name__ == "__main__":
    test_five_crop_op(plot=True)
    test_five_crop_error_msg()
    test_five_crop_md5()

ext: py

max_stars_repo_name: kanchenhao/mindspore

max_issues_repo_name: dilingsong/mindspore

max_forks_repo_name: dilingsong/mindspore

reason: import numpy


text:
import argparse
import torch
import time
import numpy as np
import os
import pickle

from models import STPM
from data_loader import get_test_dataloader
from evalute import OneClassEvaluate
from utils import reshape_embedding
from sampling_methods.KNN import KNN
from sklearn.neighbors import NearestNeighbors

THRESHOLD = 1.0

class OneClassInference():
    def __init__(self, args):
        super().__init__()
        self.test_loader = get_test_dataloader(args.dataset_path, args.load_size, args.input_size)
        self.one_class_evaluate = OneClassEvaluate()
        self.n_neighbors = args.n_neighbors
        self.model = STPM()
        self.model.eval()
        self.method = args.method

    def process(self):
        for index, batch_data in enumerate(self.test_loader):
            start_time = time.time()
            prediction = self.infer(batch_data)
            class_index, class_confidence = self.post_process(prediction)
            print('Batch %d... Done. (%.3fs)' % (index, time.time() - start_time))

            self.one_class_evaluate.gt_list_img_lvl.append(batch_data[2].cpu().numpy()[0])
            self.one_class_evaluate.pred_list_img_lvl.append(class_confidence)
        self.one_class_evaluate.eval()
        self.one_class_evaluate.get_score()

    def infer(self, batch_data):
        x, gt, label, file_name, x_type = batch_data
        # extract embedding
        # print("x: {}".format(x.sum()))
        prediction = self.model(x, 1)
        # print("output: {}".format(prediction))
        return prediction

    def post_process(self, prediction):
        embedding_coreset = pickle.load(open(os.path.join("./embeddings", 'embedding.pickle'), 'rb'))
        embedding_test = np.array(reshape_embedding(np.array(prediction)))

        if self.method == "KNN":
            knn = KNN(torch.from_numpy(embedding_coreset).cuda(), k=self.n_neighbors)
            score_patches = knn(torch.from_numpy(embedding_test).cuda())[0].cpu().detach().numpy()
        else:
            nbrs = NearestNeighbors(n_neighbors=args.n_neighbors, algorithm='ball_tree', metric='minkowski', p=2).fit(
                embedding_coreset)
            score_patches, _ = nbrs.kneighbors(embedding_test)

        N_b = score_patches[np.argmax(score_patches[:, 0])]
        w = (1 - (np.max(np.exp(N_b)) / np.sum(np.exp(N_b))))
        score = w * max(score_patches[:, 0])  # Image-level score
        if score > THRESHOLD:
            class_index = 0
        else:
            class_index = 1

        return class_index, score

def get_args():
    parser = argparse.ArgumentParser(description='ANOMALYDETECTION')
    parser.add_argument('--dataset_path',
                        default=r'/home/changwoo/hdd/datasets/mvtec_anomaly_detection')  # 'D:\Dataset\mvtec_anomaly_detection')#
    parser.add_argument('--load_size', default=256)  # 256
    parser.add_argument('--input_size', default=224)
    parser.add_argument('--project_root_path',
                        default=r'/home/changwoo/hdd/project_results/patchcore/test')
    parser.add_argument('--save_anomaly_map', default=True)
    parser.add_argument('--method', default=r'NN')
    parser.add_argument('--n_neighbors', type=int, default=9)
    args = parser.parse_args()
    return args

if __name__ == '__main__':
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    args = get_args()

    one_class_test = OneClassInference(args)
    one_class_test.process()


text:
from typing import Union
import numpy as np
from numpy import log, sqrt
from scipy.optimize import brentq

FLOAT_OR_ARRAY = Union[float, np.array]

def lake_problem(X: FLOAT_OR_ARRAY, a: FLOAT_OR_ARRAY = 0.1,
                 q: FLOAT_OR_ARRAY = 2.0,
                 b: FLOAT_OR_ARRAY = 0.42,
                 eps: FLOAT_OR_ARRAY = 0.02) -> float:
    """Lake Problem as given in Hadka et al., (2015) and Kwakkel (2017)
    modified for use as a test function.

    The `mean` and `stdev` parameters control the log normal distribution of
    natural inflows (`epsilon` in [1] and [2]).

    References
    ----------
    .. [1] Hadka, D., Herman, J., Reed, P., Keller, K., (2015).
           "An open source framework for many-objective robust decision
                making."
           Environmental Modelling & Software 74, 114–129.
           doi:10.1016/j.envsoft.2015.07.014

    .. [2] Kwakkel, J.H, (2017). "The Exploratory Modeling Workbench: An open
           source toolkit for exploratory modeling, scenario discovery, and
           (multi-objective) robust decision making."
           Environmental Modelling & Software 96, 239–250.
           doi:10.1016/j.envsoft.2017.06.054

    .. [3] Singh, R., Reed, P., Keller, K., (2015). "Many-objective robust
           decision making for managing an ecosystem with a deeply uncertain
           threshold response."
           Ecology and Society 20.
           doi:10.5751/ES-07687-200312

    Parameters
    ----------
    X : float or np.ndarray
        normalized concentration of Phosphorus at point in time
    a : float or np.ndarray
        rate of anthropogenic pollution (0.0 to 0.1)
    q : float or np.ndarray
        exponent controlling recycling rate (2.0 to 4.5).
    b : float or np.ndarray
        decay rate for phosphorus
        (0.1 to 0.45, where default 0.42 is irreversible, as described in [1])
    eps : float or np.ndarray
          natural inflows of phosphorus (pollution), see [3]

    Returns
    -------
    * float, phosphorus pollution for a point in time
    """
    Xq = X**q
    X_t1 = X + a + (Xq / (1.0 + Xq)) - (b*X) + eps

    return X_t1

def evaluate_lake(values: np.ndarray, seed=101) -> np.ndarray:
    """Evaluate the Lake Problem with an array of parameter values.

    References
    ----------
    .. [1] Hadka, D., Herman, J., Reed, P., Keller, K., (2015).
           "An open source framework for many-objective robust decision
                making."
           Environmental Modelling & Software 74, 114–129.
           doi:10.1016/j.envsoft.2015.07.014

    .. [2] Singh, R., Reed, P., Keller, K., (2015). "Many-objective robust
           decision making for managing an ecosystem with a deeply uncertain
           threshold response."
           Ecology and Society 20.
           doi:10.5751/ES-07687-200312

    Parameters
    ----------
    values : np.ndarray,
             model inputs in the (column) order of
             a, q, b, mean, stdev

             where
             * `a` is rate of anthropogenic pollution
             * `q` is exponent controlling recycling rate
             * `b` is decay rate for phosphorus
             * `mean` and
             * `stdev` set the log normal distribution of `eps`, see [2]

    Returns
    -------
    np.ndarray, of Phosphorus pollution over time `t`
    """
    rng = np.random.default_rng(seed)

    nvars = values.shape[0]

    a, q, b, mean, stdev = values.T

    sq_mean = mean**2
    sq_std = stdev**2
    eps = rng.lognormal(log(sq_mean / sqrt(sq_std + sq_mean)),
                        sqrt(log(1.0 + sq_std / sq_mean)),
                        size=nvars)

    Y = np.zeros((nvars, nvars))
    for t in range(nvars):
        # First X value will be last Y value (should be 0 as we are filling
        # an array of zeros)
        Y[t] = lake_problem(Y[t-1], a, q, b, eps)

    return Y

def evaluate(values: np.ndarray, nvars: int = 100, seed=101):
    """Evaluate the Lake Problem with an array of parameter values.

    Parameters
    ----------
    values : np.ndarray,
             model inputs in the (column) order of
             a, q, b, mean, stdev, delta, alpha

    nvars : int,
            number of decision variables to simulate (default: 100)

    Returns
    -------
    np.ndarray : max_P, utility, inertia, reliability
    """
    a, q, b, _, __, delta, alpha = values.T

    nsamples = len(a)
    Y = np.empty((nsamples, 4))
    for i in range(nsamples):
        tmp = evaluate_lake(values[i, :5], seed=seed)

        a_i = a[i]
        q_i = q[i]

        Pcrit = brentq(lambda x: x**q_i/(1.0+x**q_i) - b[i]*x, 0.01, 1.5)

        reliability = len(tmp[tmp < Pcrit]) / nvars

        max_P = np.max(tmp)
        utility = np.sum(alpha[i]*a_i*np.power(delta[i], np.arange(nvars)))

        # In practice, `a` will be set by a separate decision model
        # See [2] in docstring for `lake_problem()`
        # Here, it is a constant for a given scenario.
        # The value for `tau` (0.02) is taken from [2].
        inertia = len(a_i[a_i < 0.02]) / nvars

        Y[i, :] = max_P, utility, inertia, reliability

    return Y

if __name__ == '__main__':
    from SALib.sample import latin
    from SALib.analyze import delta

    SEED_VAL = 101

    LAKE_SPEC = {
        'num_vars': 7,
        'names': ['a', 'q', 'b', 'mean', 'stdev', 'delta', 'alpha'],
        'bounds': [[0.0, 0.1], [2.0, 4.5], [0.1, 0.45], [0.01, 0.05],
                   [0.001, 0.005], [0.93, 0.99], [0.2, 0.5]],
        'outputs': ['max_P', 'Utility', 'Inertia', 'Reliability']
    }

    latin_samples = latin.sample(LAKE_SPEC, 1000, seed=SEED_VAL)
    Y = evaluate(latin_samples)

    for i, name in enumerate(LAKE_SPEC['outputs']):
        Si = delta.analyze(LAKE_SPEC, latin_samples, Y[:, i])
        print(name)
        print(Si.to_df())

max_stars_repo_name: zjzh/SALib

max_issues_repo_name: zjzh/SALib

max_forks_repo_name: zjzh/SALib

reason: import numpy,from numpy,from scipy


text:
import numpy as np
from annoy import AnnoyIndex

class PreTrainedEmbeddings(object):
    def __init__(self, word_to_index, word_vectors):
        self.word_to_index = word_to_index
        self.word_vectors = word_vectors
        self.index_to_word = {v: k for k, v in self.word_to_index.items()}
        self.index = AnnoyIndex(len(word_vectors[0]), metric="euclidean")
        for _, i in self.word_to_index.items():
            self.index.add_item(i, self.word_vectors[i])
        self.index.build(50)

    @classmethod
    def from_embeddings_file(cls, embedding_file):
        word_to_index = {}
        word_vectors = []
        with open(embedding_file) as fp:
            for line in fp.readlines():
                line = line.split(" ")
                word = line[0]
                vec = np.array([float(x) for x in line[1:]])

                word_to_index[word] = len(word_to_index)
                word_vectors.append(vec)
        return cls(word_to_index, word_vectors)

    def get_embedding(self, word):
        return self.word_vectors[self.word_to_index[word]]

    def get_closest_to_vector(self, vector, n=1):
        nn_indices = self.index.get_nns_by_vector(vector, n)
        return [self.index_to_word[neighbor] for neighbor in nn_indices]

    def compute_and_print_analogy(self, word1, word2, word3):
        vec1 = self.get_embedding(word1)
        vec2 = self.get_embedding(word2)
        vec3 = self.get_embedding(word3)

        spatial_relationship = vec2 - vec1
        vec4 = vec3 + spatial_relationship

        closest_words = self.get_closest_to_vector(vec4, n=4)
        existing_words = set([word1, word2, word3])
        closest_words = [word for word in closest_words if word not in existing_words]

        return closest_words

max_stars_repo_name: dbradf/nlp-pytorch

max_issues_repo_name: dbradf/nlp-pytorch

max_forks_repo_name: dbradf/nlp-pytorch


text:
"""
Filename: figures.py

Authors:
    Jason Youn - jyoun@ucdavis.edu

Description:
    Draw figures.

To-do:
"""
# standard imports
import argparse
import os
import sys

# third party imports
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

# global variables
DEFAULT_OUTDIR_STR = '../output'
DEFAULT_CONFIDENCE_FILE_STR = 'hypotheses_confidence.txt'

def parse_argument():
    """
    Parse input arguments.

    Returns:
        - parsed arguments
    """
    parser = argparse.ArgumentParser(description='Analyze hypotheses.')

    parser.add_argument(
        '--outdir',
        default=DEFAULT_OUTDIR_STR,
        help='Output directory.')

    return parser.parse_args()

def bin_hypotheses(pd_data, intervals=[0.0, 0.25, 0.5, 0.75, 1.0]):
    intervals_sorted = sorted(intervals)
    dict_bin_count = {}

    for i in range(len(intervals_sorted)-1):
        floor = intervals_sorted[i]
        ceil = intervals_sorted[i+1]
        bin_name = '[{}, {}{}'.format(str(floor), str(ceil), ']' if ceil == 1 else ')')

        index = pd_data['Confidence'] >= floor
        if ceil != 1:
            index &= pd_data['Confidence'] < ceil
        else:
            index &= pd_data['Confidence'] <= ceil

        dict_bin_count[bin_name] = pd_data[index].shape[0]

    print(dict_bin_count)

def which_to_test(pd_data):
    floor = 0.1
    ceil = 1.0

    index = pd_data['Confidence'] >= floor
    if ceil != 1.0:
        index &= pd_data['Confidence'] < ceil
    else:
        index &= pd_data['Confidence'] <= ceil

    pd_filtered = pd_data[index]

    pd_group = pd_filtered.groupby('Object').size()
    pd_group = pd_group.sort_values(ascending=False).to_frame(name='count')
    pd_group['cum_sum'] = pd_group['count'].cumsum()
    pd_group['cum_percentage'] = (pd_group['cum_sum'] / pd_group['count'].sum()) * 100

    print(pd_group)

    pd_group.reset_index().to_csv('~/Jason/UbuntuShare/to_test.txt', sep='\t')

def main():
    """
    Main function.
    """
    args = parse_argument()

    filepath = os.path.join(args.outdir, DEFAULT_CONFIDENCE_FILE_STR)
    col_names = ['Subject', 'Predicate', 'Object', 'Label', 'Confidence']
    pd_data = pd.read_csv(filepath, sep='\t', names=col_names)

    bin_hypotheses(pd_data.copy())
    which_to_test(pd_data.copy())

    # plt.show()

if __name__ == '__main__':
    main()


text:
from . import ipm
from . import info
from . import util
import numpy
# Solve the problem
#
#  min  1/2 Q x^2 + c x
#   x
#
# s.t.  0 <= x <= XU

class Params:
    def __init__(self, quadratic=1.0, linear=-0.9, upper_bound=1.0):
        self.quadratic = quadratic
        self.linear = linear
        self.upper_bound = upper_bound

    def get_optimal_solution(self):
        return numpy.clip(-self.linear / self.quadratic, 0.0, self.upper_bound)

    def get_s(self, x):
        return self.upper_bound - x

    def get_mult_s(self, x, mult_x):
        return mult_x - (self.quadratic * x + self.linear)

# Primal-Dual feasible iterate
class FeasibleIterate:
    def __init__(self, init_x, init_mult_x, params):
        self.params = params
        if init_x <= 0.0 or init_x >= params.upper_bound:
            raise ValueError("Invalid value for x!")
        if init_mult_x <= 0.0:
            raise ValueError("Invalid value for lower multiplier!")
        self.x = init_x
        self.mult_x = init_mult_x
        if self.s <= 0:
            raise ValueError("Value for x too close to upper bound, causing "
                             "slack to be 0.0!")
        if self.mult_s <= 0:
            raise ValueError("Value for lower multiplier forces upper "
                             "multiplier to be nonpositive!")

    @property
    def s(self):
        return self.params.get_s(self.x)

    @property
    def mult_s(self):
        return self.params.get_mult_s(self.x, self.mult_x)

    @property
    def mult_equality(self):
        return self.mult_s

    def get_compl_products(self):
        return [self.x * self.mult_x, self.s * self.mult_s]

    def avg_compl(self):
        products = self.get_compl_products()
        return sum(products) / len(products)

    def get_affine_compl_products(self, step, stepsize):
        return [(self.x + stepsize * step.x) *
                (self.mult_x + stepsize * step.mult_x),
                (self.s + stepsize * step.s) *
                (self.mult_s + stepsize * step.mult_s)]

    def affine_avg_compl(self, step, stepsize):
        products = self.get_affine_compl_products(step, stepsize)
        return (max(0.0, products[0]) + max(0.0, products[1])) / 2

    def get_mixed_products(self, step):
        return [self.x * step.mult_x + self.mult_x * step.x,
                self.s * step.mult_s + self.mult_s * step.s]

    def get_residual(self):
        return [0.0, 0.0, 0.0]

    def update(self, step, stepsize):
        self.x += stepsize * step.x
        self.mult_x += stepsize * step.mult_x
        assert self.x > 0.0
        assert self.s > 0.0
        assert self.mult_x > 0.0
        assert self.mult_s > 0.0

    def get_max_stepsize(self, step, stepsize_limiter):
        assert stepsize_limiter.is_fulfilled(self)
        return stepsize_limiter.get_max_stepsize(self, step)

    def isclose(self, other, rtol=1e-10, atol=0.0):
        return numpy.allclose([self.x, self.s, self.mult_x, self.mult_s],
                              [other.x, other.s, other.mult_x, other.mult_s],
                              rtol=rtol, atol=atol)

class Step:
    def __init__(self, raw_step):
        dx = raw_step[0]
        ds = raw_step[1]
        dmult_x = raw_step[2]
        dmult_s = raw_step[3]
        dmult_equality = raw_step[4]
        assert numpy.isclose(dx, -ds, rtol=0, atol=1e-14)
        assert numpy.isclose(dmult_s, dmult_equality, rtol=0, atol=1e-14)
        self.x = (dx - ds) / 2.0
        self.mult_x = dmult_x
        self.mult_s = (dmult_s + dmult_equality) / 2.0

    @property
    def s(self):
        return -self.x

    def get_compl_products(self):
        return [self.x * self.mult_x, self.s * self.mult_s]

    def __add__(self, other):
        cls = self.__class__
        result = cls.__new__(cls)
        result.x = self.x + other.x
        result.mult_x = self.mult_x + other.mult_x
        result.mult_s = self.mult_s + other.mult_s
        return result

    def __mul__(self, other):
        cls = self.__class__
        result = cls.__new__(cls)
        result.x = self.x * other
        result.mult_x = self.mult_x * other
        result.mult_s = self.mult_s * other
        return result

    def __rmul__(self, other):
        return self * other


text:
# Copyright (c) 2014 Evalf
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

'''
The points module defines the :class:`Points` base class, which bundles point
coordinates, point weights, a local triangulation and a hull triangulation. The
module provides several different implementations such as :class:`TensorPoints`
and :class:`SimplexGaussPoints` that reflect the variety of elements in the
:mod:`nutils.element` module.
'''

from . import types, transform, numeric, util
import numpy, functools, itertools, warnings
_ = numpy.newaxis

class Points(types.Singleton):
  '''Collection of points on an element.

  The :class:`Points` base class bundles point coordinates, point weights,
  a local triangulation and hull triangulation. Of these only the coordinates
  are mandatory, and should be provided by the derived class in the form of the
  ``coords`` attribute. Of the remaining properties only :func:`hull` has a
  functional base implementation that relies on the availability of ``tri``.

  .. attribute:: coords

    Coordinates of the points as a :class:`float` array.

  .. attribute:: weights

    Weights of the points as a :class:`float` array.

  Args
  ----
  npoints : :class:`int`
    Number of discrete points.
  ndims : :class:`int`
    Number of spatial dimensions.
  '''

  __cache__ = 'hull', 'onhull'

  @types.apply_annotations
  def __init__(self, npoints:types.strictint, ndims:types.strictint):
    self.npoints = npoints
    self.ndims = ndims

  @property
  def tri(self):
    '''Triangulation of interior.

    A two-dimensional integer array with ``ndims+1`` columns, of which every
    row defines a simplex by mapping vertices into the list of points.
    '''

    if self.ndims == 0 and self.npoints == 1:
      return types.frozenarray([[0]])
    raise Exception('tri not defined for {}'.format(self))

  @property
  def hull(self):
    '''Triangulation of the exterior hull.

    A two-dimensional integer array with ``ndims`` columns, of which every row
    defines a simplex by mapping vertices into the list of points.
    '''

    edges = set()
    iedges = numpy.array(list(itertools.combinations(range(self.ndims+1), self.ndims)))
    for tri in self.tri:
      edges.symmetric_difference_update(map(tuple, numpy.sort(tri[iedges], axis=1)))
    return numpy.array(sorted(edges))

  @property
  def onhull(self):
    '''Boolean mask marking boundary points.

    The array of length ``npoints`` is ``True`` where the corresponding point
    is part of the :attr:`hull`, and ``False`` where it is not.
    '''

    onhull = numpy.zeros(self.npoints, dtype=bool)
    onhull[numpy.ravel(self.hull)] = True # not clear why ravel is necessary but setitem seems to require it
    return types.frozenarray(onhull, copy=False)

strictpoints = types.strict[Points]

class CoordsPoints(Points):
  '''Manually supplied points.'''

  @types.apply_annotations
  def __init__(self, coords:types.frozenarray[float]):
    self.coords = coords
    super().__init__(*coords.shape)

class CoordsWeightsPoints(CoordsPoints):
  '''Manually supplied points and weights.'''

  @types.apply_annotations
  def __init__(self, coords:types.frozenarray[float], weights:types.frozenarray[float]):
    self.weights = weights
    super().__init__(coords)

class CoordsUniformPoints(CoordsPoints):
  '''Manually supplied points with uniform weights.'''

  @types.apply_annotations
  def __init__(self, coords:types.frozenarray[float], volume:float):
    self.weights = types.frozenarray.full([len(coords)], volume/len(coords))
    super().__init__(coords)

class TensorPoints(Points):
  '''Tensor product of two Points instances.'''

  __cache__ = 'coords', 'weights', 'tri', 'hull'

  @types.apply_annotations
  def __init__(self, points1:strictpoints, points2:strictpoints):
    self.points1 = points1
    self.points2 = points2
    super().__init__(points1.npoints * points2.npoints, points1.ndims + points2.ndims)

  @property
  def coords(self):
    coords = numpy.empty((self.points1.npoints, self.points2.npoints, self.ndims))
    coords[:,:,:self.points1.ndims] = self.points1.coords[:,_,:]
    coords[:,:,self.points1.ndims:] = self.points2.coords[_,:,:]
    return types.frozenarray(coords.reshape(self.npoints, self.ndims), copy=False)

  @property
  def weights(self):
    return types.frozenarray((self.points1.weights[:,_] * self.points2.weights[_,:]).ravel(), copy=False)

  @property
  def tri(self):
    if self.points1.ndims == 1:
      # For an n-dimensional simplex with vertices a0,a1,..,an, the extruded
      # element has vertices a0,a1,..,an,b0,b1,..,bn. These can be divided in
      # simplices by selecting a0,a1,..,an,b0; a1,..,an,b0,n1; and so on until
      # an,b0,b1,..,bn; resulting in n+1 n+1-dimensional simplices. In the
      # algorithm below this is achieved by first taking the tensorial product
      # of triangulations and raveling, effectively achieving vectorized
      # concatenation. The overlapping vertex subsets then follow directly from
      # numeric.overlapping.
      tri12 = self.points1.tri[:,_,:,_] * self.points2.npoints + self.points2.tri[_,:,_,:] # ntri1 x ntri2 x 2 x ndims
      return types.frozenarray(numeric.overlapping(tri12.reshape(-1, 2*self.ndims), n=self.ndims+1).reshape(-1, self.ndims+1), copy=False)
    return super().tri

  @property
  def hull(self):
    if self.points1.ndims == 1:
      hull1 = self.points1.hull[:,_,:,_] * self.points2.npoints + self.points2.tri[_,:,_,:] # 2 x ntri2 x 1 x ndims
      hull2 = self.points1.tri[:,_,:,_] * self.points2.npoints + self.points2.hull[_,:,_,:] # ntri1 x nhull2 x 2 x ndims-1
      # The subdivision of hull2 into simplices follows identical logic to that
      # used in the construction of self.tri.
      hull = numpy.concatenate([hull1.reshape(-1, self.ndims), numeric.overlapping(hull2.reshape(-1, 2*(self.ndims-1)), n=self.ndims).reshape(-1, self.ndims)])
      return types.frozenarray(hull, copy=False)
    return super().hull

class SimplexGaussPoints(CoordsWeightsPoints):
  '''Gauss quadrature points on a simplex.'''

  @types.apply_annotations
  def __init__(self, ndims:types.strictint, degree:types.strictint):
    super().__init__(*gaussn[ndims](degree))

class SimplexBezierPoints(CoordsPoints):
  '''Bezier points on a simplex.'''

  __cache__ = 'tri', 'hull'

  @types.apply_annotations
  def __init__(self, ndims:types.strictint, n:types.strictint):
    self.n = n
    linspace = numpy.linspace(0, 1, n)
    super().__init__([linspace[list(index)[::-1]] for index in numpy.ndindex(*[n] * ndims) if sum(index) < n])

  @property
  def tri(self):
    if self.ndims == 1:
      return types.frozenarray(numeric.overlapping(numpy.arange(self.n)), copy=False)
    if self.ndims == 2:
      n = self.n
      vert1 = [((2*n-i+1)*i)//2 + numpy.array([j,j+1,j+n-i]) for i in range(n-1) for j in range(n-i-1)]
      vert2 = [((2*n-i+1)*i)//2 + numpy.array([j+1,j+n-i+1,j+n-i]) for i in range(n-1) for j in range(n-i-2)]
      return types.frozenarray(vert1 + vert2, copy=False)
    return super().tri

  @property
  def hull(self):
    if self.ndims == 2:
      n = self.n
      hull = numpy.concatenate([numpy.arange(n), numpy.arange(n-1,0,-1).cumsum()+n-1, numpy.arange(n+1,2,-1).cumsum()[::-1]-n-1])
      return types.frozenarray(numeric.overlapping(hull), copy=False)
    return super().hull

class TransformPoints(Points):
  '''Affinely transformed Points.'''

  __cache__ = 'coords', 'weights'

  @types.apply_annotations
  def __init__(self, points:strictpoints, trans:transform.stricttransformitem):
    self.points = points
    self.trans = trans
    super().__init__(points.npoints, points.ndims)

  @property
  def coords(self):
    return self.trans.apply(self.points.coords)

  @property
  def weights(self):
    return self.points.weights * abs(float(self.trans.det))

  @property
  def tri(self):
    return self.points.tri

  @property
  def hull(self):
    return self.points.hull

class ConcatPoints(Points):
  '''Concatenation of several Points objects.

  An optional ``duplicates`` argument lists all points that are equal,
  triggering deduplication and resulting in a smaller total point count.
  '''

  __cache__ = 'coords', 'weights', 'tri', 'masks'

  @types.apply_annotations
  def __init__(self, allpoints:types.tuple[strictpoints], duplicates:frozenset=frozenset()):
    self.allpoints = allpoints
    self.duplicates = duplicates
    super().__init__(sum(points.npoints for points in allpoints) - sum(len(d)-1 for d in duplicates), allpoints[0].ndims)

  @property
  def masks(self):
    masks = [numpy.ones(points.npoints, dtype=bool) for points in self.allpoints]
    for pairs in self.duplicates:
      for i, j in pairs[1:]:
        masks[i][j] = False
    return tuple(masks)

  @property
  def coords(self):
    return types.frozenarray(numpy.concatenate([points.coords[mask] for mask, points in zip(self.masks, self.allpoints)] if self.duplicates else [points.coords for points in self.allpoints]), copy=False)

  @property
  def weights(self):
    if not self.duplicates:
      return types.frozenarray(numpy.concatenate([points.weights for points in self.allpoints]), copy=False)
    weights = [points.weights[mask] for mask, points in zip(self.masks, self.allpoints)]
    for pairs in self.duplicates:
      I, J = pairs[0]
      weights[I][self.masks[I][:J].sum()] += sum(self.allpoints[i].weights[j] for i, j in pairs[1:])
    return types.frozenarray(numpy.concatenate(weights), copy=False)

  @property
  def tri(self):
    if not self.duplicates:
      offsets = util.cumsum(points.npoints for points in self.allpoints)
      return types.frozenarray(numpy.concatenate([points.tri + offset for offset, points in zip(offsets, self.allpoints)]), copy=False)
    renumber = []
    n = 0
    for mask in self.masks:
      cumsum = mask.cumsum()
      renumber.append(cumsum+(n-1))
      n += cumsum[-1]
    assert n == self.npoints
    for pairs in self.duplicates:
      I, J = pairs[0]
      for i, j in pairs[1:]:
        renumber[i][j] = renumber[I][J]
    return types.frozenarray(numpy.concatenate([renum.take(points.tri) for renum, points in zip(renumber, self.allpoints)]), copy=False)

class ConePoints(Points):
  '''Affinely transformed lower-dimensional points plus tip.

  The point count is incremented by one regardless of the nature of the point
  set; no effort is made to introduce extra points between base plane and tip.
  Likewise, the simplex count stays equal, with all simplices obtaining an
  extra vertex in tip.
  '''

  __cache__ = 'coords', 'tri'

  @types.apply_annotations
  def __init__(self, edgepoints:strictpoints, edgeref:transform.stricttransformitem, tip:types.frozenarray):
    self.edgepoints = edgepoints
    self.edgeref = edgeref
    self.tip = tip
    super().__init__(edgepoints.npoints+1, edgepoints.ndims+1)

  @property
  def coords(self):
    return types.frozenarray(numpy.concatenate([self.edgeref.apply(self.edgepoints.coords), self.tip[_,:]]), copy=False)

  @property
  def tri(self):
    tri = numpy.concatenate([self.edgepoints.tri, [[self.edgepoints.npoints]]*len(self.edgepoints.tri)], axis=1)
    return types.frozenarray(tri, copy=False)

## UTILITY FUNCTIONS

@functools.lru_cache(8)
def gauss(n):
  k = numpy.arange(n) + 1
  d = k / numpy.sqrt(4*k**2-1)
  x, w = numpy.linalg.eigh(numpy.diagflat(d,-1)) # eigh operates (by default) on lower triangle
  return types.frozenarray((x+1) * .5, copy=False), types.frozenarray(w[0]**2, copy=False)

def gauss1(degree):
  '''Gauss quadrature for line.'''

  x, w = gauss(degree//2)
  return x[:,_], w

@functools.lru_cache(8)
def gauss2(degree):
  '''Gauss quadrature for triangle.

  Reference: http://www.cs.rpi.edu/~flaherje/pdf/fea6.pdf'''

  assert isinstance(degree, int) and degree >= 0

  I = [0,0],
  J = [1,1],[0,1],[1,0]
  K = [1,2],[2,0],[0,1],[2,1],[1,0],[0,2]

  icw = [
    (I, [1/3], 1)
  ] if degree <= 1 else [
    (J, [2/3,1/6], 1/3)
  ] if degree == 2 else [
    (I, [1/3], -9/16),
    (J, [3/5,1/5], 25/48),
  ] if degree == 3 else [
    (J, [0.816847572980458,0.091576213509771], 0.109951743655322),
    (J, [0.108103018168070,0.445948490915965], 0.223381589678011),
  ] if degree == 4 else [
    (I, [1/3], 0.225),
    (J, [0.797426985353088,0.101286507323456], 0.125939180544827),
    (J, [0.059715871789770,0.470142064105115], 0.132394152788506),
  ] if degree == 5 else [
    (J, [0.873821971016996,0.063089014491502], 0.050844906370207),
    (J, [0.501426509658180,0.249286745170910], 0.116786275726379),
    (K, [0.636502499121399,0.310352451033785,0.053145049844816], 0.082851075618374),
  ] if degree == 6 else [
    (I, [1/3.], -0.149570044467671),
    (J, [0.479308067841924,0.260345966079038], 0.175615257433204),
    (J, [0.869739794195568,0.065130102902216], 0.053347235608839),
    (K, [0.638444188569809,0.312865496004875,0.048690315425316], 0.077113760890257),
  ]

  if degree > 7:
    warnings.warn('inexact integration for polynomial of degree {}'.format(degree))

  return types.frozenarray(numpy.concatenate([numpy.take(c,i) for i, c, w in icw]), copy=False), \
         types.frozenarray(numpy.concatenate([[w/2] * len(i) for i, c, w in icw]), copy=False)

@functools.lru_cache(8)
def gauss3(degree):
  '''Gauss quadrature for tetrahedron.

  Reference http://www.cs.rpi.edu/~flaherje/pdf/fea6.pdf'''

  assert isinstance(degree, int) and degree >= 0

  I = [0,0,0],
  J = [1,1,1],[0,1,1],[1,1,0],[1,0,1]
  K = [0,1,1],[1,0,1],[1,1,0],[1,0,0],[0,1,0],[0,0,1]
  L = [0,1,1],[1,0,1],[1,1,0],[2,1,1],[1,2,1],[1,1,2],[1,0,2],[0,2,1],[2,1,0],[1,2,0],[0,1,2],[2,0,1]

  icw = [
    (I, [1/4], 1),
  ] if degree == 1 else [
    (J, [0.5854101966249685,0.1381966011250105], 1/4),
  ] if degree == 2 else [
    (I, [.25], -.8),
    (J, [.5,1/6], .45),
  ] if degree == 3 else [
    (I, [.25], -.2368/3),
    (J, [0.7857142857142857,0.0714285714285714], .1372/3),
    (K, [0.1005964238332008,0.3994035761667992], .448/3),
  ] if degree == 4 else [
    (I, [.25], 0.1817020685825351),
    (J, [0,1/3.], 0.0361607142857143),
    (J, [8/11.,1/11.], 0.0698714945161738),
    (K, [0.4334498464263357,0.0665501535736643], 0.0656948493683187),
  ] if degree == 5 else [
    (J, [0.3561913862225449,0.2146028712591517], 0.0399227502581679),
    (J, [0.8779781243961660,0.0406739585346113], 0.0100772110553207),
    (J, [0.0329863295731731,0.3223378901422757], 0.0553571815436544),
    (L, [0.2696723314583159,0.0636610018750175,0.6030056647916491], 0.0482142857142857),
  ] if degree == 6 else [
    (I, [.25], 0.1095853407966528),
    (J, [0.7653604230090441,0.0782131923303186],  0.0635996491464850),
    (J, [0.6344703500082868,0.1218432166639044], -0.3751064406859797),
    (J, [0.0023825066607383,0.3325391644464206],  0.0293485515784412),
    (K, [0,.5], 0.0058201058201058),
    (L, [.2,.1,.6], 0.1653439153439105)
  ] if degree == 7 else [
    (I, [.25], -0.2359620398477557),
    (J, [0.6175871903000830,0.1274709365666390], 0.0244878963560562),
    (J, [0.9037635088221031,0.0320788303926323], 0.0039485206398261),
    (K, [0.4502229043567190,0.0497770956432810], 0.0263055529507371),
    (K, [0.3162695526014501,0.1837304473985499], 0.0829803830550589),
    (L, [0.0229177878448171,0.2319010893971509,0.5132800333608811], 0.0254426245481023),
    (L, [0.7303134278075384,0.0379700484718286,0.1937464752488044], 0.0134324384376852),
  ]

  if degree > 8:
    warnings.warn('inexact integration for polynomial of degree {}'.format(degree))

  return types.frozenarray(numpy.concatenate([numpy.take(c,i) for i, c, w in icw]), copy=False), \
         types.frozenarray(numpy.concatenate([[w/6] * len(i) for i, c, w in icw]), copy=False)

gaussn = None, gauss1, gauss2, gauss3

def find_duplicates(allpoints):
  coords = {}
  for i, points in enumerate(allpoints):
    for j in points.onhull.nonzero()[0]:
      coords.setdefault(tuple(points.coords[j]), []).append((i, j))
  return [tuple(pairs) for pairs in coords.values() if len(pairs) > 1]

# vim:sw=2:sts=2:et


text:
import os
from pathlib import Path
import pandas as pd
import numpy as np
from nbformat import read, NO_CONVERT
import re
import glob
from github import Github
import git
import warnings

import webbrowser

def _find_assignment(directory=None):
    """Find an Assignment Dyamically

    A helper function to validate and locate the lab file based on its
    extension being Rmd or ipynb. The utility will search recursively up the
    directory. If multiple candidate files are found, the user will be
    prompted to select which file they wish.

    Args:
        directory ([type], optional): A directory path to be searched
            recursively. If None given, will use the working directory.

    Returns:
        [str]: A path to the file selected.
    """
    if directory is None:
        directory = os.getcwd()

    types = ["*.ipynb", "*.Rmd"]
    files = []
    for type in types:
        path = str(Path(directory).expanduser())
        # path = str(directory)
        pathname = path + "/**/*" + type
        type_files = glob.glob(pathname, recursive=True)
        files += type_files

    names = [
        str(n + 1) + "." + os.path.basename(file)
        for n, file in enumerate(files)
    ]
    print("The existing files are:")
    for item in names:
        print(item)
    notebook = input("Enter your file number from the above list:")
    notebook = files[int(notebook) - 1]
    return notebook

def create_github_token(host="https://github.ubc.ca"):
    """Open A Browser to Generate a New Github Enterprise Token

    Args:
        host (str):
            The URL to the upstream host. Defaults to UBC Github Enterprise.
    Returns:
        None

    Examples:
        >>> from labzen import labzen as lz
        >>> # Open a web browser
        >>> lz.create_github_token()
    """
    opts = "scopes=repo,user,gist,workflow&description=LABZEN"
    url = f"{host}/settings/tokens/new?{opts}"
    webbrowser.open(url, new=2)

def parse_lab(notebook=None):
    """Parse MDS lab files to return the markdown content

    Args:
        notebook (str):
            A path or list of paths to MDS lab files (either .ipynb
            or .Rmd). If left blank, the function will recursively
            search for all labs in the working directory based on the file
            extension.

    Returns:
        list: Each element of list is a content of one markdown cell.

    Example:
        >>> # Download the demo files into the working directory
        >>> import urllib.request
        >>> from labzen import labzen as lz
        >>>
        >>> baseurl = (
        >>>     "https://raw.githubusercontent.com"
        >>>     + "/UBC-MDS/labzen/main/data-raw"
        >>> )
        >>> labs = {
        >>>     "dummylab.Rmd": f"{baseurl}/dummylab.Rmd",
        >>>     "dummylab.ipynb": f"{baseurl}/dummylab.ipynb",
        >>> }
        >>>
        >>> for name, url in labs.items():
        >>>     urllib.request.urlretrieve(url, name)
        >>>
        >>> # parse the labs
        >>> lz.parse_lab("dummylab.ipynb")
        >>> lz.parse_lab("dummylab.Rmd")
        >>>
        >>> # Alternatively, navigate to a student assignment repo and
        >>> # run the following code.
        >>> lz.parse_lab()
    """
    # If the user did not define the specific file, recursively
    # search for rmd and ipynb files in the working directory
    if notebook is None:
        notebook = _find_assignment()
    path = Path(notebook)
    _, extension = os.path.splitext(notebook)

    # defensive tests
    if extension != ".Rmd" and extension != ".ipynb":
        raise Exception(
            "Sorry, you have not provided Rmarkdown or jupyter notebook file"
        )

    if not isinstance(notebook, str):
        raise Exception("The file path should be string")

    # Parse the markdown contents of rmd or ipynb file
    source = []
    if extension == ".Rmd":
        text_and_code = path.expanduser().read_text()
        text_and_code = text_and_code.split("```")

        code_blocks = []
        for string in text_and_code:
            if string.startswith("{r"):
                code_blocks.append(string)
            elif string.startswith("{python"):
                code_blocks.append(string)
            else:
                source.append(string)
    else:
        with open(notebook, encoding="utf8") as file:
            notebook = read(file, NO_CONVERT)
            cells = notebook["cells"]
            code_cells = [c for c in cells if c["cell_type"] == "markdown"]
            for cell in code_cells:
                source.append(cell["source"])

    return source

def count_points(file_name: str = None, margins: bool = True):
    """Tally Available Points in Lab

    Args:
        file_name (str): A path or list of paths to MDS lab files (either
            .ipynb or .Rmd). If left blank, the function will recursively
            search for all labs in the working directory based on the file
            extension.
        margins (bool): A boolean indicating whether to add a row for the
            total number of points (optional + required). Defaults to True.

    Returns:
        (pandas.core.frame.DataFrame, pandas.core.frame.DataFrame):
            A tuple of DataFrames. The first is a section-by-section overview
            of points available. The second is a cross table summarising the
            number of optional, required, and total points per lab.

    Example:
        >>> from labzen import labzen as lz
        >>> import urllib.request
        >>>
        >>> # Download the demo files into the working directory
        >>> baseurl = (
        >>>     "https://raw.githubusercontent.com"
        >>>     + "/UBC-MDS/labzen/main/data-raw"
        >>> )
        >>> labs = {
        >>>     "dummylab.Rmd": f"{baseurl}/dummylab.Rmd",
        >>>     "dummylab.ipynb": f"{baseurl}/dummylab.ipynb",
        >>> }
        >>> for name, url in labs.items():
        >>>     urllib.request.urlretrieve(url, name)
        >>>
        >>> # for Jupyter notebooks:
        >>> df, tab = lz.count_points("dummylab.ipynb")
        >>> print(df[["rubric", "points", "type"]])
                        rubric  points          type
        0            [mechanics]     [5]  Non-Optional
        1            [reasoning]     [4]  Non-Optional
        2  [accuracy, reasoning]  [3, 2]  Non-Optional
        3  [accuracy, reasoning]  [6, 4]      Optional
        4  [accuracy, reasoning]  [7, 3]      Optional
        5                  [viz]     [5]  Non-Optional
        >>> print(tab)
                   type  total  prop
        0  Non-Optional     19  0.95
        1      Optional     20  1.00
        2           All     39  1.95
        >>>
        >>> # for Rmarkdown:
        >>> df, tab = lz.count_points("dummylab.Rmd")
        >>> print(tab)
                type  total      prop
        0  Non-Optional     42  0.950000
        1      Optional     12  0.271429
        2           All     54  1.221429
        >>>
        >>> # Alternatively, navigate to a student assignment repo and run the
        >>> # following code.
        >>> df, tab = lz.count_points()
    """
    # Parse a lab file into its markdown blocks
    res = parse_lab(file_name)
    df = pd.DataFrame({"block": np.arange(1, len(res) + 1), "txt": res})

    # Tidy breaks, new lines, extra spaces, and make each line a row
    df["txt"] = df["txt"].str.replace("
", "\n")
    df["txt"] = df["txt"].str.split("\n")
    df = df.explode("txt")
    df["txt"] = df["txt"].replace(["", "
"], np.nan)
    df = df.dropna()
    df["txt"] = df["txt"].str.strip()

    # Add variable transformations
    df["header"] = df["txt"].shift(1)
    df["rubric"] = df["txt"].str.contains(r"^rubric\=\{")
    df["below_header"] = df["header"].str.contains(r"^[#]{1,6}\s")
    df["optional"] = df["header"].str.contains("optional|bonus", case=False)

    # Subset to lines containing rubrics only
    df = df.dropna().query("rubric")

    # Extract and sum points
    df["points"] = df["txt"].str.findall(r"(\d+)")
    df2 = df.explode("points")
    df2["points"] = df2["points"].astype(int)
    df["points"] = df2["points"].groupby(df2.index).apply(list)
    df["total"] = df["points"].apply(sum)

    # defensive check
    if not all(df["below_header"]):
        raise Exception(
            "There is a problem parsing this lab. Expecting a rubric tag to "
            + "below a markdown header."
        )

    # Tidy and make the result more human-readable
    booldict = {True: "Optional", False: "Non-Optional"}
    df["type"] = df["optional"].replace(booldict)
    df = df.drop(columns=["rubric", "below_header", "optional"])
    df = df.reset_index(drop=True)
    df["header"] = df["header"].str.replace(r"(^[#]+\s+)", "", regex=True)

    # Generate crosstab
    tab = df.pivot_table("total", "type", aggfunc=sum, margins=margins)
    tab = tab.reset_index()
    one_pt_worth = 0.95 / tab.loc[tab["type"] == "Non-Optional", "total"]
    tab["prop"] = tab["total"] * one_pt_worth[0]

    # add percent to full table
    df["prop"] = df["total"] * one_pt_worth[0]

    # simplify rubric names
    df["rubric"] = df["txt"].str.findall(r"([a-z]+)(?=\:\d)")

    # re-order columns
    df = df[["block", "header", "rubric", "points", "total", "prop", "type"]]

    return df, tab

def _check_repo_link(file_name: str = None):
    """Check whether the user has included the github repo link in his/her
        repository

    Args:
        file_name (str):
            A path or list of paths to MDS lab files (either
            .ipynb or .Rmd). If left blank, the function will recursively
            search for all labs in the working directory based on the file
            extension.

    Returns:
        bool: a boolean output

    Example:
        >>> # Navigate to the root of labzen repo and run the following code
        >>> # using the dummy files:
        >>>
        >>> # for jupyter notebook:
        >>> _check_repo_link("data-raw/dummylab.ipynb")
        >>>
        >>> # for Rmarkdown:
        >>> _check_repo_link("data-raw/dummylab.Rmd")
        >>>
        >>> # Alternatively, navigate to a student assignment repo and run the
        >>> # following code.
        >>> _check_repo_link()
    """

    # Parse a lab file into its markdown blocks
    res = parse_lab(file_name)

    df = pd.DataFrame({"block": np.arange(1, len(res) + 1), "txt": res})

    # finding out if there is any link
    rex = re.compile(
        r"((https://)?(www.)?github\.ubc\.ca"  # base url for GH Enterprise
        r"\/MDS-\d{4}-\d{2}"  # organization
        r"\/DSCI_\d{3}_lab\d_[a-z]+)"  # lab repo with CWL username
    )
    warnings.filterwarnings("ignore", "This pattern has match groups")
    df["link"] = df["txt"].str.contains(rex, regex=True)

    # displaying the result in boolean
    repo_link = df["link"].any()
    if repo_link:
        print("Check 3: Repository link is included in the file")
        print(f"Check 3: {repo_link}")
    else:
        print("Check 3: Repository link is not included in the file")
        print("Check 3:", False)

    return repo_link

def _check_lat_version(path: str, token: str):
    """Check whether the user has pushed the latest version in his/her
        repository

    Args:
        path (str): A local file path to either a lab directory or to a
            lab file inside a local git directory.
        token (str): A token for https://github.ubc.ca

    Returns:
        bool: a boolean output

    Example:
        >>> from labzen import labzen as lz
        >>>
        >>> # navigate to a student repo and run:
        >>> token = "544c96ce0d3dc9b66ac8d70b32c07bd0c46129db"
        >>> lz._check_lat_version(token=token)
    """
    # locate the repo root
    local_repo = git.Repo(path, search_parent_directories=True)

    # locate the Github Enterprise repo name
    ghe = Github(token, base_url="https://github.ubc.ca/api/v3")
    ghe_repo = ghe.get_repo(__find_ghe_repo(local_repo))

    # find latest commit on GHE
    ghe_commit = ghe_repo.get_commits()[0].sha

    # find the latest local commit
    local_commit = str(local_repo.head.commit)

    # comparing the both SHAs
    if ghe_commit == local_commit:
        print("Check 2: Remote has the latest version of the repository")
        print("Check 2:", True)
    else:
        print(
            "Check 2: Remote does not have the latest version of the ",
            "repository",
        )
        print("Check 2:", False)
    return ghe_commit == local_commit

def __find_ghe_repo(local_repo, org="MDS-2020-21"):
    """Find a Github Repo Path

    Args:
        local_repo ([git.repo.base.Repo]): A local Github repository
        org (str): The name of the organization on Github Enterprise to search.
            Defaults to 'MDS-2020-21'.

    Returns:
        [type]: [github.Repository.Repository] An UBC GHE remote (student
            project repo).
    """
    # find the name of the repo on Github Enterprise
    remote_urls = [list(x.urls)[0] for x in local_repo.remotes]
    rex = "([A-Za-z0-9_-]+)(?=\\.git$)"
    ghe_name = list(set([re.search(rex, x).group(1) for x in remote_urls]))[0]

    return f"{org}/{ghe_name}"

def _check_commits(path: str, token: str, verbose=False):
    """Check whether the user has at least three commits

    Args:
        path (str): A local repo path or local file path to a lab.git
        token (str): A token for Github Enterprise.
        verbose (bool): Whether to print commit details to screen

    Returns:
        bool: a boolean output

    Example:
        >>> from labzen import labzen as lz
        >>>
        >>> path = "/Users/jene/MDS/Block5/lab/DSCI_599_lab1_jene3456"
        >>> token = "544c96ce0d3dc9b66ac8d70b32c07bd0c46129db"
        >>> lz._check_commits(path, token)
    """

    # locate the repo root
    local_repo = git.Repo(path, search_parent_directories=True)

    # locate the Github Enterprise repo name
    ghe = Github(token, base_url="https://github.ubc.ca/api/v3")

    try:
        # org/repo_name
        ghe_name = __find_ghe_repo(local_repo)
        ghe_repo = ghe.get_repo(ghe_name)
    except Exception:
        raise Exception(f"{ghe_name} not found on github.ubc.ca")

    # count the total number of commits on the remote
    ghe_commits = ghe_repo.get_commits()
    ghe_commit_n = ghe_commits.totalCount

    if ghe_commit_n > 3:
        student_name = ghe.get_user().name
        student_commits_n = 0
        for commit in ghe_commits:
            valid_names = set()
            if hasattr(commit.author, "name"):
                valid_names.add(commit.author.name)
            if hasattr(commit.committer, "name"):
                valid_names.add(commit.committer.name)

            if student_name in valid_names:
                student_commits_n += 1

                if verbose:
                    print(
                        commit,
                        commit.commit.committer,
                        commit.commit.author.email,
                        commit.commit.author.date,
                        commit.author.name,
                    )
        check_result = student_commits_n >= 3
    else:
        check_result = False

    # print check result to screen
    if check_result:
        print(
            "Check 1: Repository has at least 3 commits with the ",
            f"student username {student_name}",
        )
        print("Check 1:", check_result)
    elif student_commits_n < 3:
        print(
            "Check 1: Repository does not have 3 commits with the "
            f"student username {student_name}"
        )
        print("Check 1:", check_result)
    else:
        print(
            f"Check 1: Repo {ghe_repo.name} has fewer than 3 commits ",
            f"with the student username {student_name}",
        )
        print("Check 1:", check_result)

    return check_result

def check_mechanics(path: str = None, token=None):
    """Performs Mechanics Checks on a MDS Lab

       This function checks that you...
        1. ... have a Github repo link;
        2. ... have pushed your latest commit; and
        3. ... have at least three commit messages authored by you in
            your history.

    Args:
        path (str): A local path to a Github directory or an MDS lab file
            (.ipynb or .Rmd) within such a directory.

        token (str) : A personal access token for https://github.ubc.ca. See
            ``create_github_token()`` for details.

    Returns:
        bool : A boolean whether all checks passed. The function also prints
            informative messages as a side effect.

    Example:
        >>> from labzen import labzen as lz
        >>>
        >>> # Step 1: get a token
        >>> lz.create_github_token()
        >>>
        >>> # Step 2: check mechanics
        >>> file = "~/MDS/Block5/lab1/DSCI_599_lab1_jene3456"
        >>> token = "544c96ce0d3dc9b66ac8d70b32c07bd0c46129db"
        >>> lz.check_mechanics(file, token)
        Check 1: Repository has at least 3 commits with the student
        username JENE SMITH
        Check 1: True
        Check 2: Remote has the latest version of the repository
        Check 2: True
        Check 3: Repository link is included in the file
        Check 3: True
        >>>
        >>> # Alternatively, just run the following from an MDS lab directory:
        >>> lz.check_mechanics(token = token)

    """
    # use the current working directory if no path given
    if path is None:
        path = os.getcwd()

    # local the repo root
    repo_path = git.Repo(path, search_parent_directories=True).git_dir

    # local lab file
    _, extension = os.path.splitext(path)
    if extension == ".Rmd" or extension == ".ipynb":
        lab_path = path
    else:
        lab_path = _find_assignment(path)

    result = [
        _check_commits(repo_path, token=token),
        _check_lat_version(repo_path, token=token),
        _check_repo_link(lab_path),
    ]

    return all(result)

max_stars_repo_name: UBC-MDS/labzen

max_issues_repo_name: UBC-MDS/labzen

max_forks_repo_name: UBC-MDS/labzen


text:
import copy
import math
import copy

from coopihc.base.Space import Space
from coopihc.base.State import State
from coopihc.helpers import sort_two_lists
from coopihc.policy.BasePolicy import BasePolicy

import numpy

# ============= Discrete Policies =================

# ----------- Bayesian Information Gain Policy ---------

class BIGDiscretePolicy(BasePolicy):
    """BIGDiscretePolicy [summary]

    Bayesian Information Gain Policy, adapted from [1]_.

    The main ideas/assumptions are:

        * A user wants the task to go to some goal state :math:`\\Theta`
        * The assistant can put the task in a number of states (X)
        * The user can perform a given set of action Y
        * A model :math:`p(Y=y|X=X, \\Theta = \\theta)` exists for user behavior

    Make sure to call:

        * attach_set_theta, to specify the potential goal states
        * attach_transition_function, to specify how the task state evolves after an assistant action

    .. [1] Liu, Wanyu, et al. "Bignav: Bayesian information gain for guiding multiscale navigation." Proceedings of the 2017 CHI Conference on Human Factors in Computing Systems. 2017.

    :param assistant_action_state: action state of the assistant
    :type assistant_action_state: `State<coopihc.base.State.State>`
    :param user_policy_model: user policy model. This may be the real policy of the user, but realistically has to be a model of the user policy. This policy must currently be an `ELLDiscretePolicy<coopihc.policy.ELLDiscretePolicy.ELLDiscretePolicy>`.
    :type user_policy_model: ELLDiscretePolicy<coopihc.policy.ELLDiscretePolicy.ELLDiscretePolicy>`
    """

    def __init__(
        self, assistant_action_state, user_policy_model, *args, threshold=0.8, **kwargs
    ):
        self.threshold = threshold
        super().__init__(*args, action_state=assistant_action_state, **kwargs)

        self.assistant_action_set = Space.cartesian_product(
            self.action_state["action"].space
        )[0]

        self.user_policy_model = user_policy_model
        self.user_action_set = Space.cartesian_product(
            user_policy_model.action_state["action"].space
        )[0]

        self.user_policy_likelihood_function = user_policy_model.compute_likelihood

    def attach_set_theta(self, set_theta):
        self.set_theta = set_theta

    def attach_transition_function(self, trans_func):
        self.transition_function = trans_func

    def PYy_Xx(self, user_action, assistant_action, potential_states, beliefs):
        """:math:`P(Y=y|X=x)`

        Computes the conditional probability :math:`P(Y=y|X=x)`, where X is the assistant outcome and Y the user's response.

        :param user_action: user action y for which the condition is computed
        :type user_action: `StateElement<coopihc.base.StateElement.StateElement>`
        :param assistant_action: assistant action to be evaluated
        :type assistant_action: `StateElement<coopihc.base.StateElement.StateElement>`
        :param potential_states: collection of potential goal states
        :type potential_states: iterable
        :param beliefs: (list) beliefs for each target
        :type beliefs: (list) beliefs for each target

        :return: the conditional :math:`P(Y=y|X=x)`
        :rtype: list
        """
        pYy__Xx = 0
        for potential_state, belief in zip(potential_states, beliefs):
            pYy__Xx += (
                self.user_policy_likelihood_function(user_action, potential_state)
                * belief
            )
        return pYy__Xx

    def HY__Xx(self, potential_states, assistant_action, beliefs):
        """:math:`H(Y |X=x)`

        Computes the conditional entropy :math:`H(Y |X=x) = -\mathbb{E}[\log(p(Y|X=x))]`.

        :param assistant_action: assistant action to be evaluated
        :type assistant_action: `StateElement<coopihc.base.StateElement.StateElement>`
        :param potential_states: collection of potential goal states
        :type potential_states: iterable
        :param beliefs: (list) beliefs for each target
        :type beliefs: (list) beliefs for each target
        :return: :math:`H(Y |X=x)`
        :rtype: float
        """
        H = 0
        for user_action in self.user_action_set:
            pYy_Xx = self.PYy_Xx(
                user_action, assistant_action, potential_states, beliefs
            )
            if pYy_Xx != 0:
                H += -pYy_Xx * math.log(pYy_Xx, 2)
        return H

    def HY__OoXx(self, potential_states, beliefs):
        """:math:`H(Y |\Theta = \theta, X=x)`

        Computes the conditional entropy :math:`H(Y |\Theta = \theta, X=x) = -\mathbb{E}[\log(p(Y|\Theta = \theta, X=x))]`.

        :param potential_states: collection of potential goal states
        :type potential_states: iterable
        :param beliefs: (list) beliefs for each target
        :type beliefs: (list) beliefs for each target
        :return: :math:`H(Y |\Theta = \theta, X=x)`
        :rtype: float
        """
        H = 0
        for user_action in self.user_action_set:
            for potential_state, belief in zip(potential_states, beliefs):

                pYy__OoXx = self.user_policy_likelihood_function(
                    user_action, potential_state
                )

                if pYy__OoXx != 0:  # convention: 0 log 0 = 0
                    H += -belief * pYy__OoXx * math.log(pYy__OoXx, 2)

        return H

    def IG(self, assistant_action, observation, beliefs):
        """Information Gain :math:`\mathrm{IG}(X=x)`

        Computes the expected information gain :math:`\mathrm{IG}(X=x) = H(Y |X=x) - H(Y |\Theta = \theta, X=x)` for a potential assistant action x.

        :param assistant_action: assistant action to be evaluated
        :type assistant_action: `StateElement<coopihc.base.StateElement.StateElement>`
        :param observation: current assistant observation
        :type observation: `State<coopihc.base.State.State>`
        :param beliefs: (list) beliefs for each target
        :type beliefs: (list) beliefs for each target
        :return: [description]
        :rtype: [type]
        """

        observation = self.transition_function(assistant_action, observation)
        potential_states = []
        for nt, t in enumerate(self.set_theta):
            # Deepcopy would be safer, but copy should do. Deepcopy is much more expensive to produce.
            # potential_state = copy.deepcopy(observation)
            potential_state = copy.copy(observation)
            for key, value in t.items():
                try:
                    potential_state[key[0]][key[1]] = value
                except KeyError:  # key[0] is not in observation
                    _state = State()
                    _state[key[1]] = value
                    potential_state[key[0]] = _state
            potential_states.append(potential_state)

        return self.HY__Xx(potential_states, assistant_action, beliefs) - self.HY__OoXx(
            potential_states, beliefs
        )

    def find_best_action(self):
        """find_best_action

        Finds expected information gain associated with each possible future cursor position and ranks them in order from the most to less informative.

        :return: (assistant actions, associated information gain)
        :rtype: tuple(list, list)
        """

        beliefs = self.host.state["beliefs"]
        index = numpy.argmax(beliefs)
        hp = beliefs[index]
        if hp > self.threshold:
            targets = self.observation["task_state"]["targets"]
            hp_target = targets[index]
            return [hp_target], [None]
        else:
            observation = self.observation

        IG_storage = [
            self.IG(action, observation, beliefs.squeeze().tolist())
            for action in self.assistant_action_set
        ]

        _IG, action = sort_two_lists(
            IG_storage, self.assistant_action_set, lambda pair: pair[0]
        )
        action.reverse(), _IG.reverse()
        return action, _IG

    @BasePolicy.default_value
    def sample(self, agent_observation=None, agent_state=None):
        """sample

        Choose action (select the action with highest expected information gain)

        :return: (assistant action, associated reward)
        :rtype: tuple(`StateElement<coopihc.base.StateElement.StateElement>`, float)
        """
        self._actions, self._IG = self.find_best_action()

        return self._actions[0], 0

max_stars_repo_name: jgori-ouistiti/CoopIHC

max_issues_repo_name: jgori-ouistiti/CoopIHC

max_forks_repo_name: jgori-ouistiti/CoopIHC