text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# This file is part of the bapsflib package, a Python toolkit for the
# BaPSF group at UCLA.
#
# http://plasma.physics.ucla.edu/
#
# Copyright 2017-2018 Erik T. Everson and contributors
#
# License: Standard 3-clause BSD; see "LICENSES/LICENSE.txt" for full
#   license terms and contributor agreement.
#
import numpy as np
import unittest as ut

from numpy.lib import recfunctions as rfn
from unittest import mock

from bapsflib.utils.exceptions import HDFMappingError

from ..sis3301 import HDFMapDigiSIS3301
from .common import DigitizerTestCase

class TestSIS3301(DigitizerTestCase):
    """Test class for HDFMapDigiSIS3301"""

    DEVICE_NAME = "SIS 3301"
    DEVICE_PATH = f"/Raw data + config/{DEVICE_NAME}"
    MAP_CLASS = HDFMapDigiSIS3301

    def setUp(self):
        super().setUp()

    def tearDown(self):
        super().tearDown()

    def test_construct_dataset_name(self):
        """Test functionality of method `construct_dataset_name`"""
        # setup
        config_name = "config01"
        adc = "SIS 3301"
        # config_path = 'Configuration: {}'.format(config_name)
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.n_configs = 2
        self.mod.knobs.active_brdch = bc_arr

        # -- Handling of kwarg `config_name`                        ----
        # not specified, and only ONE active config
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        with self.assertWarns(UserWarning):
            self.assertEqual(self.map.construct_dataset_name(brd, ch), dset_name)

        # not specified, and MULTIPLE active configs
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        self.mod.knobs.active_config = (config_name, "config02")
        self.assertRaises(ValueError, self.map.construct_dataset_name, brd, ch)
        self.mod.knobs.active_config = config_name

        # not specified, and NO active configs
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        _map = self.map
        with mock.patch.object(
            HDFMapDigiSIS3301, "active_configs", new_callable=mock.PropertyMock
        ) as mock_aconfig:
            mock_aconfig.return_value = []
            self.assertRaises(ValueError, _map.construct_dataset_name, brd, ch)

        # `config_name` not in configs
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        self.assertRaises(
            ValueError,
            self.map.construct_dataset_name,
            brd,
            ch,
            config_name="not a config",
        )

        # `config_name` in configs but not active
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        self.assertRaises(
            ValueError, self.map.construct_dataset_name, brd, ch, config_name="config02"
        )

        # -- Handling of kwarg `adc`                                ----
        # `adc` not 'SIS 3301'
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        self.assertRaises(
            ValueError, self.map.construct_dataset_name, brd, ch, adc="not SIS 3301"
        )

        # -- `board` and `channel` combo not in configs             ----
        brd = 1
        ch = 1
        for conn in my_bcs:
            if brd == conn[0] and ch in conn[1]:
                self.fail(
                    "test setup is incorrect, brd and ch should not be in connections"
                )
        self.assertRaises(ValueError, self.map.construct_dataset_name, brd, ch)

        # -- return when `return_info=True`                         ----
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        _map = self.map
        d_info = {}
        for conn in _map.configs[config_name][adc]:
            if brd == conn[0] and ch in conn[1]:
                d_info = conn[2]
                break

        # get dset_name
        val = _map.construct_dataset_name(brd, ch, return_info=True)

        self.assertIsInstance(val, tuple)
        self.assertEqual(len(val), 2)

        # first element is dataset name
        self.assertEqual(val[0], dset_name)

        # second element is dataset info
        self.assertIsInstance(val[1], dict)
        keys = (
            "adc",
            "bit",
            "clock rate",
            "configuration name",
            "digitizer",
            "nshotnum",
            "nt",
            "sample average (hardware)",
            "shot average (software)",
        )
        for key in keys:
            self.assertIn(key, val[1])

            if key == "adc":
                self.assertEqual(val[1][key], adc)
            elif key == "configuration name":
                self.assertEqual(val[1][key], config_name)
            elif key == "digitizer":
                self.assertEqual(val[1][key], _map.info["group name"])
            else:
                self.assertEqual(val[1][key], d_info[key])

    def test_construct_header_dataset_name(self):
        """
        Test functionality of method `construct_header_dataset_name`
        """
        # setup:
        config_name = "config01"
        # adc = 'SIS 3301'
        # config_path = 'Configuration: {}'.format(config_name)
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.n_configs = 2
        self.mod.knobs.active_brdch = bc_arr

        # `return_info` does NOT return extra info
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        hdset_name = f"{dset_name} headers"
        _map = self.map
        with mock.patch.object(
            HDFMapDigiSIS3301, "construct_dataset_name", wraps=_map.construct_dataset_name
        ) as mock_cdn:

            name = _map.construct_header_dataset_name(brd, ch, return_info=True)

            # check `construct_dataset_name` was called
            self.assertTrue(mock_cdn.called)

            # check equality
            self.assertEqual(name, hdset_name)

    def test_map_failures(self):
        """Test scenarios that should raise HDFMappingError"""
        # 1. board config group missing attribute 'Board'
        # 2. more than one board config group (for the same config_name)
        #    defines the same board number
        # 3. channel config group missing attribute 'Channel'
        # 4. there are no identifiable configuration groups
        # 5. none of the configurations are active
        # 6. adc connections for active config are NULL
        #
        # setup group
        config_name = "config01"
        # adc = 'SIS 3301'
        config_path = f"Configuration: {config_name}"
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.active_brdch = bc_arr

        # -- failures that occur in `_find_adc_connections`         ----
        # attribute 'Board' missing in board config group
        path = f"{config_path}/Boards[0]"
        brd_group = self.dgroup[path]
        brd = brd_group.attrs["Board"]
        del brd_group.attrs["Board"]
        with self.assertRaises(HDFMappingError):
            _map = self.map
        brd_group.attrs["Board"] = brd

        # the same board number is defined multiple times for an active
        # configuration
        path = f"{config_path}/Boards[0]"
        path2 = f"{config_path}/Boards[{len(my_bcs)}]"
        brd = self.dgroup[path].attrs["Board"]
        self.dgroup.create_group(path2)
        self.dgroup[path2].attrs["Board"] = brd
        with self.assertRaises(HDFMappingError):
            _map = self.map
        del self.dgroup[path2]

        # attribute 'Channel' missing in channel config group
        brd_path = f"{config_path}/Boards[0]"
        ch_path = f"{brd_path}/Channels[0]"
        ch_group = self.dgroup[ch_path]
        ch = ch_group.attrs["Channel"]
        del ch_group.attrs["Channel"]
        with self.assertRaises(HDFMappingError):
            _map = self.map
        ch_group.attrs["Channel"] = ch

        # -- failures that occur in `_build_configs`                ----
        # group had no identifiable configuration group
        self.dgroup.move(config_path, "wrong config name")
        with self.assertRaises(HDFMappingError):
            _map = self.map
        self.dgroup.move("wrong config name", config_path)

        # none of the configurations are active
        self.dgroup.move(config_path, "Configuration: Not used")
        with self.assertRaises(HDFMappingError):
            _map = self.map
        self.dgroup.move("Configuration: Not used", config_path)

        # adc connections for active config are NULL
        brd_config_names = list(self.dgroup[config_path])
        for name in brd_config_names:
            path = f"{config_path}/{name}"
            del self.dgroup[path]
        with self.assertRaises(HDFMappingError):
            _map = self.map

    def test_map_warnings(self):
        """Test scenarios that should cause a UserWarning."""
        # 1.  a configuration group sub-group does not match naming
        #     scheme for a board config group
        # 2.  'Board' attribute for a board config group is not an int
        #     or np.integer
        # 3.  'Board' attribute for a board config group is a negative
        #     integer
        # 4.  for a none active config, two board groups define the same
        #     board number
        # 5.  a board config sub-group does not match the naming scheme
        #     for a channel group
        # 6.  'Channel' attribute for a channel config group is not an
        #     int or np.integer
        # 7.  'Channel' attribute for a channel config group is a
        #     negative integer
        # 8.  two channel config groups define the same channel number
        # 9.  the list of discovered channel numbers is NULL
        # 10. config group attribute 'Samples to average' is not
        #     convertible to int
        # 11. an expected dataset is missing
        # 12. all expected datasets for a board are missing
        # 13. dataset has fields
        # 14. dataset is not a 2D array
        # 15. number of dataset time samples not consistent for all
        #     channels connected to a board
        # 16. number of dataset shot numbers not consistent for all
        #     channels connect to a board, but are still consistent
        #     with their associated header dataset
        # 17. header dataset missing expected shot number field
        # 18. shot number field in header dataset does not have
        #     expected shape and/or dtype
        # 19. dataset and associated header dataset do not have same
        #     number of shot numbers
        # 20. after all the above checks, ensure the connected channels
        #     are not NULL for the board
        #
        # setup group
        config_name = "config01"
        adc = "SIS 3301"
        config_path = f"Configuration: {config_name}"
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.n_configs = 2
        self.mod.knobs.active_brdch = bc_arr

        # -- warnings that occur in `_find_adc_connections`         ----
        # configuration group sub-group does not match board group   (1)
        # name
        brd_path = f"{config_path}/Boards[0]"
        new_path = f"{config_path}/Not a board"
        self.dgroup.move(brd_path, new_path)
        with self.assertWarns(UserWarning):
            _map = self.map
        self.dgroup.move(new_path, brd_path)

        # 'Board' attribute for a board config group is not an int   (2)
        # or np.integer
        brd_path = f"{config_path}/Boards[0]"
        brd_group = self.dgroup[brd_path]
        brd = brd_group.attrs["Board"]
        brd_group.attrs["Board"] = "five"
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn("five", [conn[0] for conn in _map.configs[config_name][adc]])

        # 'Board' attribute for a board config group is a negative   (3)
        # int
        brd_group.attrs["Board"] = -1
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(-1, [conn[0] for conn in _map.configs[config_name][adc]])
        brd_group.attrs["Board"] = brd

        # for none active config, two board config groups define     (4)
        # the same board number
        path = "Configuration: config02/Boards[0]"
        path2 = "Configuration: config02/Boards[1]"
        brd = self.dgroup[path2].attrs["Board"]
        self.dgroup[path2].attrs["Board"] = self.dgroup[path].attrs["Board"]
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(brd, [conn[0] for conn in _map.configs["config02"][adc]])
        self.dgroup[path2].attrs["Board"] = brd

        # a board config group sub-group does not match naming       (5)
        # scheme for a channel group
        brd_path = f"{config_path}/Boards[0]"
        ch_path = f"{brd_path}/Channels[0]"
        new_path = f"{brd_path}/Not a channel"
        self.dgroup.move(ch_path, new_path)
        with self.assertWarns(UserWarning):
            _map = self.map
        self.dgroup.move(new_path, ch_path)

        # 'Channel' attribute for a channel config group is not an   (6)
        # int or np.integer
        brd_path = f"{config_path}/Boards[0]"
        ch_path = f"{brd_path}/Channels[0]"
        ch_group = self.dgroup[ch_path]
        brd = self.dgroup[brd_path].attrs["Board"]
        ch = ch_group.attrs["Channel"]
        ch_group.attrs["Channel"] = "five"
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn("five", chs)

        # 'Channel' attribute for a channel config group is a        (7)
        # negative int
        ch_group.attrs["Channel"] = -1
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(-1, chs)
        ch_group.attrs["Channel"] = ch

        # two channel config groups define the same channel number   (8)
        path = f"{brd_path}/Channels[0]"
        path2 = f"{brd_path}/Channels[1]"
        brd = self.dgroup[brd_path].attrs["Board"]
        ch = self.dgroup[path2].attrs["Channel"]
        self.dgroup[path2].attrs["Channel"] = self.dgroup[path].attrs["Channel"]
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(brd, [conn[0] for conn in _map.configs[config_name][adc]])
        self.dgroup[path2].attrs["Channel"] = ch

        # the list of discovered channel numbers is NULL             (9)
        # - this could happen if there are no channel config groups
        brd_path = f"{config_path}/Boards[0]"
        brd = self.dgroup[brd_path].attrs["Board"]
        ch_group_names = list(self.dgroup[brd_path])
        for name in ch_group_names:
            old_path = f"{brd_path}/{name}"
            new_path = old_path + "Q"
            self.dgroup.move(old_path, new_path)
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(brd, [conn[0] for conn in _map.configs[config_name][adc]])
        for name in ch_group_names:
            old_path = f"{brd_path}/{name}"
            new_path = old_path + "Q"
            self.dgroup.move(new_path, old_path)

        # -- warnings that occur in `_adc_info_first_pass`          ----
        # config group attribute 'Samples to average' is not        (10)
        # convertible to int
        config_group = self.dgroup[config_path]
        s2a = config_group.attrs["Samples to average"]
        config_group.attrs["Samples to average"] = b"Average 9.0 Samples"
        with self.assertWarns(UserWarning):
            _map = self.map

            for conn in _map.configs[config_name][adc]:
                self.assertIsNone(conn[2]["sample average (hardware)"])
        config_group.attrs["Samples to average"] = s2a

        # -- warnings that occur in `_adc_info_second_pass`         ----
        # an expected dataset is missing                            (11)
        # i.e. the config groups define a board-channel combo that
        #      does not have an existing dataset
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        new_name = dset_name + "Q"
        self.dgroup.move(dset_name, new_name)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        self.dgroup.move(new_name, dset_name)

        # all expected datasets for a given board are missing       (12)
        brd = my_bcs[0][0]
        chs = my_bcs[0][1]
        for ch in chs:
            dset_name = f"{config_name} [{brd}:{ch}]"
            new_name = dset_name + "Q"
            self.dgroup.move(dset_name, new_name)
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(brd, [conn[0] for conn in _map.configs[config_name][adc]])
        for ch in chs:
            dset_name = f"{config_name} [{brd}:{ch}]"
            new_name = dset_name + "Q"
            self.dgroup.move(new_name, dset_name)

        # datasets has fields                                       (13)
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        new_name = dset_name + "Q"
        self.dgroup.move(dset_name, new_name)
        data = np.empty(3, dtype=[("f1", np.int16), ("f2", np.int16)])
        self.dgroup.create_dataset(dset_name, data=data)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[dset_name]
        self.dgroup.move(new_name, dset_name)

        # dataset is not a 2D array                                 (14)
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        new_name = dset_name + "Q"
        self.dgroup.move(dset_name, new_name)
        data = np.empty((3, 100, 3), dtype=np.int16)
        self.dgroup.create_dataset(dset_name, data=data)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[dset_name]
        self.dgroup.move(new_name, dset_name)

        # number of dataset time samples not consistent for         (15)
        # all channels connected to board
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        new_name = dset_name + "Q"
        self.dgroup.move(dset_name, new_name)
        dset = self.dgroup[new_name]
        data = np.empty((dset.shape[0], dset.shape[1] + 1), dtype=dset.dtype)
        self.dgroup.create_dataset(dset_name, data=data)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

                    # nt is set to -1
                    self.assertEqual(conn[2]["nt"], -1)

            # channel still in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertIn(ch, chs)
        del self.dgroup[dset_name]
        self.dgroup.move(new_name, dset_name)

        # number dataset shot numbers not consistent for all        (16)
        # channels connected to board, but are still consistent
        # with their associated header dataset
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        hdset_name = f"{dset_name} headers"
        data = self.dgroup[dset_name][...]
        hdata = self.dgroup[hdset_name][...]
        data2 = np.append(data, data[-2::, ...], axis=0)
        hdata2 = np.append(hdata, hdata[-2::, ...], axis=0)
        self.dgroup.move(dset_name, dset_name + "Q")
        self.dgroup.move(hdset_name, hdset_name + "Q")
        self.dgroup.create_dataset(dset_name, data=data2)
        self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

                    # nshotnum is set to -1
                    self.assertEqual(conn[2]["nshotnum"], -1)

            # channel still in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertIn(ch, chs)
        del self.dgroup[dset_name]
        del self.dgroup[hdset_name]
        self.dgroup.move(dset_name + "Q", dset_name)
        self.dgroup.move(hdset_name + "Q", hdset_name)

        # header dataset missing expected shot number field         (17)
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        hdset_name = f"{dset_name} headers"
        hdata = self.dgroup[hdset_name][...]
        names = list(hdata.dtype.names)
        names.remove("Shot")
        hdata2 = hdata[names]
        self.dgroup.move(hdset_name, hdset_name + "Q")
        self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            # channel not in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[hdset_name]
        self.dgroup.move(hdset_name + "Q", hdset_name)

        # shot number field in header dataset does not have         (18)
        # expected shape and/or dtype
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        hdset_name = f"{dset_name} headers"
        hdata = self.dgroup[hdset_name][...]
        self.dgroup.move(hdset_name, hdset_name + "Q")

        # wrong dtype
        hdata2 = np.empty(hdata.shape, dtype=[("Shot", np.float32)])
        self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            # channel not in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[hdset_name]

        # wrong shape
        hdata2 = np.empty(hdata.shape, dtype=[("Shot", np.uint32, 2)])
        self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            # channel not in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[hdset_name]
        self.dgroup.move(hdset_name + "Q", hdset_name)

        # dataset and associated header dataset do not have same    (19)
        # number of shot numbers
        brd = my_bcs[0][0]
        ch = my_bcs[0][1][0]
        dset_name = f"{config_name} [{brd}:{ch}]"
        hdset_name = f"{dset_name} headers"
        hdata = self.dgroup[hdset_name][...]
        hdata2 = np.append(hdata, hdata[-2::, ...], axis=0)
        self.dgroup.move(hdset_name, hdset_name + "Q")
        self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            chs = None
            for conn in _map.configs[config_name][adc]:
                if conn[0] == brd:
                    chs = conn[1]

            # channel not in mapping
            if chs is None:
                self.fail("board missing from connections")
            else:
                self.assertNotIn(ch, chs)
        del self.dgroup[hdset_name]
        self.dgroup.move(hdset_name + "Q", hdset_name)

        # after all the above checks, ensure the connected          (20)
        # channels are not NULL for the board
        # i.e. this could happen if all the header datasets for a
        #      given board are missing the shot number field
        brd = my_bcs[0][0]
        chs = my_bcs[0][1]
        for ch in chs:
            dset_name = f"{config_name} [{brd}:{ch}]"
            hdset_name = f"{dset_name} headers"

            hdata = self.dgroup[hdset_name][...]
            names = list(hdata.dtype.names)
            names.remove("Shot")
            hdata2 = hdata[names]

            self.dgroup.move(hdset_name, hdset_name + "Q")
            self.dgroup.create_dataset(hdset_name, data=hdata2)
        with self.assertWarns(UserWarning):
            _map = self.map

            self.assertNotIn(brd, [conn[0] for conn in _map.configs[config_name][adc]])
        for ch in chs:
            dset_name = f"{config_name} [{brd}:{ch}]"
            hdset_name = f"{dset_name} headers"
            del self.dgroup[hdset_name]
            self.dgroup.move(hdset_name + "Q", hdset_name)

    def test_mappings(self):
        """Test various digitizer group setups."""
        # -- One Config & One Active Config                         ----
        # setup faux group
        config_name = "config01"
        adc = "SIS 3301"
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.active_brdch = bc_arr

        # test
        _map = self.map
        self.assertDigitizerMapBasics(_map, self.dgroup)
        self.assertEqual(_map.active_configs, [config_name])
        self.assertEqual(list(_map.configs), [config_name])
        self.assertConnectionsEqual(_map, tuple(my_bcs), adc, config_name)

        # -- Multiple Configs & One Active Config                   ----
        # setup faux group
        config_name = "config02"
        adc = "SIS 3301"
        self.mod.knobs.n_configs = 3
        self.mod.knobs.active_config = config_name
        my_bcs = [(0, (1, 2, 3)), (3, (0, 1, 2, 3))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.active_brdch = bc_arr

        # test
        _map = self.map
        self.assertDigitizerMapBasics(_map, self.dgroup)
        self.assertEqual(_map.active_configs, [config_name])
        self.assertEqual(sorted(list(_map.configs)), sorted(self.mod.config_names))
        self.assertConnectionsEqual(_map, tuple(my_bcs), adc, config_name)

        # -- Multiple Configs & Two Active Config                   ----
        # setup faux group
        config_names = ("config02", "config03")
        adc = "SIS 3301"
        self.mod.knobs.n_configs = 3
        self.mod.knobs.active_config = config_names
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7)), (8, (1,))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.active_brdch = bc_arr

        # test
        _map = self.map
        self.assertDigitizerMapBasics(_map, self.dgroup)
        self.assertEqual(sorted(_map.active_configs), sorted(list(config_names)))
        self.assertEqual(sorted(list(_map.configs)), sorted(self.mod.config_names))
        for config_name in config_names:
            self.assertConnectionsEqual(_map, tuple(my_bcs), adc, config_name)

    def test_misc(self):
        """
        Test misc behavior the does not fit into other test methods.
        """
        # What's tested...
        #   1. Behavior of digitizer configuration group attribute
        #      `Shots to average`
        #   2. Behavior of digitizer configuration group attribute
        #      `Samples to average`
        #      - including when attribute is named 'Unnamed' instead
        #   4. Identifying header datasets with shot number field name
        #      of 'Shot number'
        #
        # setup
        config_name = "config01"
        adc = "SIS 3301"
        config_path = f"Configuration: {config_name}"
        my_bcs = [(0, (0, 3, 5)), (3, (0, 1, 2, 3)), (5, (5, 6, 7))]
        bc_arr = self.mod.knobs.active_brdch
        bc_arr[...] = False
        for brd, chns in my_bcs:
            bc_arr[brd, chns] = True
        self.mod.knobs.active_brdch = bc_arr

        # -- config group attribute `Shots to average`              ----
        sh2a = self.dgroup[config_path].attrs["Shots to average"]

        # `Shots to average' missing
        del self.dgroup[config_path].attrs["Shots to average"]
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["shot average (software)"])

        # `Shots to average' is 0 or 1
        self.dgroup[config_path].attrs["Shots to average"] = 1
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["shot average (software)"])

        # `Shots to average' is >1
        self.dgroup[config_path].attrs["Shots to average"] = 5
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertEqual(conn[2]["shot average (software)"], 5)

        self.dgroup[config_path].attrs["Shots to average"] = sh2a

        # -- config group attribute `Samples to average`            ----
        sp2a = self.dgroup[config_path].attrs["Samples to average"]

        # 'Samples to average' is missing
        del self.dgroup[config_path].attrs["Samples to average"]
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["sample average (hardware)"])

        # 'Samples to average' is 'No averaging'
        self.dgroup[config_path].attrs["Samples to average"] = np.bytes_("No averaging")
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["sample average (hardware)"])

        # 'Samples to average' does not match string
        # 'Average {} Samples'
        self.dgroup[config_path].attrs["Samples to average"] = np.bytes_("hello")
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["sample average (hardware)"])

        # 'Samples to average' does match string 'Average {} Samples',
        # but can not be converted to int
        self.dgroup[config_path].attrs["Samples to average"] = np.bytes_(
            "Average 5.0 Samples"
        )
        _map = None
        with self.assertWarns(UserWarning):
            _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["sample average (hardware)"])

        # 'Samples to average' does match string 'Average {} Samples',
        # and is converted to an int of 0 or 1
        for val in (0, 1):
            self.dgroup[config_path].attrs["Samples to average"] = np.bytes_(
                f"Average {val} Samples"
            )
            _map = self.map
            for conn in _map.configs[config_name][adc]:
                self.assertIsNone(conn[2]["sample average (hardware)"])

        # 'Samples to average' does match string 'Average {} Samples',
        # and is converted to an int >1
        self.dgroup[config_path].attrs["Samples to average"] = np.bytes_(
            "Average 5 Samples"
        )
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertEqual(conn[2]["sample average (hardware)"], 5)

        # 'Samples to average' is named 'Unnamed' instead, and valid
        # (as seen on some SmPD HDF5 files)
        del self.dgroup[config_path].attrs["Samples to average"]
        self.dgroup[config_path].attrs["Unnamed"] = np.bytes_("Average 2 Samples")
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertEqual(conn[2]["sample average (hardware)"], 2)

        # 'Samples to average' is named 'Unnamed' instead, and is NOT a
        # byte string (as seen on some SmPD HDF5 files)
        self.dgroup[config_path].attrs["Unnamed"] = 5
        _map = self.map
        for conn in _map.configs[config_name][adc]:
            self.assertIsNone(conn[2]["sample average (hardware)"])
        del self.dgroup[config_path].attrs["Unnamed"]

        # restore original value
        self.dgroup[config_path].attrs["Samples to average"] = sp2a

        # -- header dataset with 'Shot number' field                ----
        # rename field for all board-channel connections
        for conn in my_bcs:
            brd = conn[0]
            for ch in conn[1]:
                dset_name = f"{config_name} [{brd}:{ch}]"
                hdset_name = f"{dset_name} headers"
                hdset = self.dgroup[hdset_name][...]
                hdset = rfn.rename_fields(hdset, {"Shot": "Shot number"})
                del self.dgroup[hdset_name]
                self.dgroup.create_dataset(hdset_name, data=hdset)

        _map = self.map
        self.assertEqual(
            _map.configs[config_name]["shotnum"]["dset field"], ("Shot number",)
        )

        # Note: module has NOT been reset to defaults at this point

    def test_parse_config_name(self):
        """Test HDFMapDigiSIS3301 method `_parse_config_name`."""
        _map = self.map  # type: HDFMapDigiSIS3301
        self.assertTrue(hasattr(_map, "_parse_config_name"))
        self.assertEqual(
            _map._parse_config_name("Configuration: all-probes"), "all-probes"
        )
        self.assertIsNone(_map._parse_config_name("Not a config"))

if __name__ == "__main__":
    ut.main()

max_stars_repo_name: BaPSF/bapsflib

max_issues_repo_name: BaPSF/bapsflib


text:
"""Test high-level mutual information functions."""
import numpy as np

from frites.stats import (permute_mi_vector, permute_mi_trials,
                          bootstrap_partitions, dist_to_ci)

rnd = np.random.RandomState(0)

mi_types = ['cc', 'cd', 'ccd']
inferences = ['ffx', 'rfx']
n_perm = 15

n_times, n_epochs, n_suj, n_conds = 100, 100, 2, 3
y = rnd.rand(n_epochs)
suj = np.round(np.linspace(0, n_suj, n_epochs)).astype(int)

class TestNonParam(object):  # noqa

    def test_permute_mi_vector(self):
        """Test function permute_mi_vector."""
        for mi in mi_types:
            for inf in inferences:
                y_p = permute_mi_vector(y, suj, mi_type=mi, inference=inf,
                                        n_perm=n_perm)
                assert len(y_p) == n_perm

    def test_permute_mi_trials(self):
        """Test function permute_mi_trials."""
        suj = np.array([0, 0, 0, 1, 1, 1])
        for inf in inferences:
            y_p = permute_mi_trials(suj, inference=inf, n_perm=n_perm)
            assert len(y_p) == n_perm

            # rfx testing
            if inf == 'ffx':
                assert any([k[0:3].max() > 2 for k in y_p])
            elif inf == 'rfx':
                for k in y_p:
                    assert (k[0:3].min() == 0) and (k[0:3].max() == 2)
                    assert (k[3::].min() == 3) and (k[3::].max() == 5)

    def test_bootstrap_partitions(self):
        """Test function bootstrap_partitions."""
        # overall testing
        part = bootstrap_partitions(10, n_partitions=5)
        assert len(part) == 5
        for k in part:
            assert (0 <= k).all() and (k < 10).all()

        # group testing
        gp_1 = np.array([0, 0, 0, 1, 1, 1])
        part = bootstrap_partitions(6, gp_1, n_partitions=5)
        for k in part:
            assert (0 <= k[0:3].min()) and (k[0:3].max() <= 2)
            assert (3 <= k[3::].min()) and (k[3::].max() <= 5)

        gp_1 = np.array([0, 0, 1, 1, 2, 2])
        part = bootstrap_partitions(6, gp_1, n_partitions=5)
        for k in part:
            assert (0 <= k[0:2].min()) and (k[0:2].max() <= 1)
            assert (2 <= k[2:4].min()) and (k[2:4].max() <= 3)
            assert (4 <= k[4::].min()) and (k[4::].max() <= 5)

    def test_dist_to_ci(self):
        """Test function dist_to_ci."""
        # standard definition
        dist = np.random.rand(200, 1, 30)
        assert dist_to_ci(dist, cis=[95]).shape == (1, 2, 30)
        assert dist_to_ci(dist, cis=[95, 99, 99.9]).shape == (3, 2, 30)

        # mean effect size
        dist = np.random.rand(200, 10, 30)
        assert dist_to_ci(dist, cis=[95], inference='rfx').shape == (1, 2, 30)
        assert dist_to_ci(
            dist, cis=[95, 99, 99.9], inference='rfx').shape == (3, 2, 30)

if __name__ == '__main__':
    TestNonParam().test_dist_to_ci()

ext: py

reason: import numpy


text:
################################################################################
# Copyright (c) 2021 ContinualAI.                                              #
# Copyrights licensed under the MIT License.                                   #
# See the accompanying LICENSE file for terms.                                 #
#                                                                              #
# Date: 26-06-2020                                                             #
# Author(s): Gabriele Graffieti                                                #
# E-mail: contact@continualai.org                                              #
# Website: avalanche.continualai.org                                           #
################################################################################

from typing import Optional, Sequence, Union, Any
from os.path import expanduser
import torch
from PIL.Image import Image
from torch import Tensor
from torchvision.datasets import MNIST
from torchvision.transforms import ToTensor, ToPILImage, Compose, Normalize, \
    RandomRotation
import numpy as np

from avalanche.benchmarks import NCScenario, nc_scenario
from avalanche.benchmarks.utils import train_test_transformation_datasets

_default_mnist_train_transform = Compose([
    ToTensor(),
    Normalize((0.1307,), (0.3081,))
])

_default_mnist_test_transform = Compose([
    ToTensor(),
    Normalize((0.1307,), (0.3081,))
])

class PixelsPermutation(object):
    """
    Apply a fixed permutation to the pixels of the given image.

    Works with both Tensors and PIL images. Returns an object of the same type
    of the input element.
    """

    def __init__(self, index_permutation: Sequence[int]):
        self.permutation = index_permutation
        self._to_tensor = ToTensor()
        self._to_image = ToPILImage()

    def __call__(self, img: Union[Image, Tensor]):
        is_image = isinstance(img, Image)
        if (not is_image) and (not isinstance(img, Tensor)):
            raise ValueError('Invalid input: must be a PIL image or a Tensor')

        if is_image:
            img = self._to_tensor(img)

        img = img.view(-1)[self.permutation].view(*img.shape)

        if is_image:
            img = self._to_image(img)

        return img

def SplitMNIST(
        n_experiences: int,
        return_task_id=False,
        seed: Optional[int] = None,
        fixed_class_order: Optional[Sequence[int]] = None,
        train_transform=_default_mnist_train_transform,
        test_transform=_default_mnist_test_transform):
    """
    Creates a CL scenario using the MNIST dataset.
    This helper create the basic split MNIST scenario, where the 10 classes of
    the MNIST dataset are evenly splitted into the given number of tasks (or
    experiences, more in general).
    If the dataset is not present in the computer the method automatically
    download it and store the data in the data folder.

    :param n_experiences: The number of incremental experiences in the current
        scenario.
        The value of this parameter should be a divisor of 10.
    :param return_task_id: if True, for every experience the task id is
        returned and the Scenario is Multi Task. This means that the scenario
        returned will be of type ``NCMultiTaskScenario``. If false the task
        index is not returned (default to 0 for every batch) and the returned
        scenario is of type ``NCSingleTaskScenario``.
    :param seed: A valid int used to initialize the random number generator.
        Can be None.
    :param fixed_class_order: A list of class IDs used to define the class
        order. If None, value of ``seed`` will be used to define the class
        order. If non-None, ``seed`` parameter will be ignored.
        Defaults to None.
    :param train_transform: The transformation to apply to the training data,
        e.g. a random crop, a normalization or a concatenation of different
        transformations (see torchvision.transform documentation for a
        comprehensive list of possible transformations).
        If no transformation is passed, the default train transformation
        will be used.
    :param test_transform: The transformation to apply to the test data,
        e.g. a random crop, a normalization or a concatenation of different
        transformations (see torchvision.transform documentation for a
        comprehensive list of possible transformations).
        If no transformation is passed, the default test transformation
        will be used.

    :returns: A :class:`NCMultiTaskScenario` instance initialized for the the
        MT split MNIST scenario if the parameter ``return_task_id`` is True,
        a :class:`NCSingleTaskScenario` initialized for the SIT split MNIST
        scenario otherwise.
    """

    mnist_train, mnist_test = _get_mnist_dataset(
        train_transform, test_transform)

    if return_task_id:
        return nc_scenario(
            train_dataset=mnist_train,
            test_dataset=mnist_test,
            n_experiences=n_experiences,
            task_labels=True,
            seed=seed,
            fixed_class_order=fixed_class_order,
            class_ids_from_zero_in_each_exp=True)
    else:
        return nc_scenario(
            train_dataset=mnist_train,
            test_dataset=mnist_test,
            n_experiences=n_experiences,
            task_labels=False,
            seed=seed,
            fixed_class_order=fixed_class_order)

def PermutedMNIST(
        n_experiences: int,
        seed: Optional[int] = None,
        train_transform: Any = _default_mnist_train_transform,
        test_transform: Any = _default_mnist_test_transform) -> NCScenario:
    """
    This helper create a permuted MNIST scenario: where a given number of random
    pixel permutations is used to permute the MNIST images in
    ``n_experiences`` different manners, creating an equal number of tasks.
    Each task is composed of all the original MNIST 10 classes, but the pixel
    in the images are permuted in different ways in every task.
    If the dataset is not present in the computer the method automatically
    download it and store the data in the data folder.

    :param n_experiences: The number of experiences (tasks) in the current
        scenario. It indicates how many different permutations of the MNIST
        dataset have to be created.
        The value of this parameter should be a divisor of 10.
    :param seed: A valid int used to initialize the random number generator.
        Can be None.
    :param train_transform: The transformation to apply to the training data
        before the random permutation, e.g. a random crop, a normalization or a
        concatenation of different transformations (see torchvision.transform
        documentation for a comprehensive list of possible transformations).
        If no transformation is passed, the default train transformation
        will be used.
    :param test_transform: The transformation to apply to the test data
        before the random permutation, e.g. a random crop, a normalization or a
        concatenation of different transformations (see torchvision.transform
        documentation for a comprehensive list of possible transformations).
        If no transformation is passed, the default test transformation
        will be used.

    :returns: A :class:`NCMultiTaskScenario` instance initialized for the the
        MT permuted MNIST scenario.
    """

    list_train_dataset = []
    list_test_dataset = []
    rng_permute = np.random.RandomState(seed)

    # for every incremental experience
    for _ in range(n_experiences):
        # choose a random permutation of the pixels in the image
        idx_permute = torch.from_numpy(rng_permute.permutation(784)).type(
            torch.int64)

        permutation = PixelsPermutation(idx_permute)

        mnist_train, mnist_test = _get_mnist_dataset(permutation, permutation)

        # Freeze the permutation, then add the user defined transformations
        permuted_train = mnist_train \
            .freeze_transforms() \
            .add_transforms(train_transform)

        permuted_test = mnist_test \
            .freeze_transforms() \
            .add_transforms(test_transform)

        list_train_dataset.append(permuted_train)
        list_test_dataset.append(permuted_test)

    return nc_scenario(
        list_train_dataset,
        list_test_dataset,
        n_experiences=len(list_train_dataset),
        task_labels=True,
        shuffle=False,
        class_ids_from_zero_in_each_exp=True,
        one_dataset_per_exp=True)

def RotatedMNIST(
        n_experiences: int,
        seed: Optional[int] = None,
        rotations_list: Optional[Sequence[int]] = None,
        train_transform=_default_mnist_train_transform,
        test_transform=_default_mnist_test_transform) -> NCScenario:
    """
    This helper create a rotated MNIST scenario: where a given number of random
    rotations are used to rotate the MNIST images in
    ``n_experiences`` different manners, creating an equal number of tasks.
    Each task is composed of all the original MNIST 10 classes, but the images
    are rotated in different ways and using different values in every task.
    If the dataset is not present in the computer the method automatically
    download it and store the data in the data folder.

    :param n_experiences: The number of experiences (tasks) in the current
        scenario. It indicates how many different rotations of the MNIST
        dataset have to be created.
        The value of this parameter should be a divisor of 10.
    :param seed: A valid int used to initialize the random number generator.
        Can be None.
    :param rotations_list: A list of rotations values in degrees (from -180 to
        180) used to define the rotations. The rotation specified in position
        0 of the list will be applieed to the task 0, the rotation specified in
        position 1 will be applyed to task 1 and so on.
        If None, value of ``seed`` will be used to define the rotations.
        If non-None, ``seed`` parameter will be ignored.
        Defaults to None.
    :param train_transform: The transformation to apply to the training data
        after the random rotation, e.g. a random crop, a normalization or a
        concatenation of different transformations (see torchvision.transform
        documentation for a comprehensive list of possible transformations).
        If no transformation is passed, the default train transformation
        will be used.
    :param test_transform: The transformation to apply to the test data
        after the random rotation, e.g. a random crop, a normalization or a
        concatenation of different transformations (see torchvision.transform
        documentation for a comprehensive list of possible transformations).
        If no transformation is passed, the default test transformation
        will be used.

    :returns: A :class:`NCMultiTaskScenario` instance initialized for the the
        MT rotated MNIST scenario.
    """

    assert len(rotations_list) == n_experiences, "The number of rotations" \
                                                 " should match the number" \
                                                 " of incremental experiences."
    assert all(-180 <= rotations_list[i] <= 180
               for i in range(len(rotations_list))), "The value of a rotation" \
                                                     " should be between -180" \
                                                     " and 180 degrees."

    list_train_dataset = []
    list_test_dataset = []
    rng_rotate = np.random.RandomState(seed)

    # for every incremental experience
    for exp in range(n_experiences):
        if rotations_list is not None:
            rotation_angle = rotations_list[exp]
        else:
            # choose a random rotation of the pixels in the image
            rotation_angle = rng_rotate.randint(-180, 181)

        rotation = RandomRotation(degrees=(rotation_angle, rotation_angle))

        mnist_train, mnist_test = _get_mnist_dataset(rotation, rotation)

        # Freeze the rotation, then add the user defined transformations
        rotated_train = mnist_train \
            .freeze_transforms() \
            .add_transforms(train_transform)

        rotated_test = mnist_test \
            .freeze_transforms() \
            .add_transforms(test_transform)

        list_train_dataset.append(rotated_train)
        list_test_dataset.append(rotated_test)

    return nc_scenario(
        list_train_dataset,
        list_test_dataset,
        n_experiences=len(list_train_dataset),
        task_labels=True,
        shuffle=False,
        class_ids_from_zero_in_each_exp=True,
        one_dataset_per_exp=True)

def _get_mnist_dataset(train_transformation, test_transformation):
    train_set = MNIST(root=expanduser("~") + "/.avalanche/data/mnist/",
                      train=True, download=True)

    test_set = MNIST(root=expanduser("~") + "/.avalanche/data/mnist/",
                     train=False, download=True)

    return train_test_transformation_datasets(
        train_set, test_set, train_transformation, test_transformation)

__all__ = [
    'SplitMNIST',
    'PermutedMNIST',
    'RotatedMNIST'
]

if __name__ == "__main__":
    import matplotlib.pyplot as plt
    import numpy as np
    from PIL import Image

    scenario = PermutedMNIST(5, train_transform=None, test_transform=None)
    for i, (img, label) in enumerate(scenario.train_stream[0].dataset):
        plt.imshow(np.asarray(img))
        plt.show()
        if ((i+1) % 5) == 0:
            break

max_stars_repo_name: Mattdl/RehearsalRevealed

max_issues_repo_name: Mattdl/RehearsalRevealed

max_forks_repo_name: Mattdl/RehearsalRevealed


text:
# vim: tabstop=2 expandtab shiftwidth=2 softtabstop=8
if __name__ == '__main__':
  raise ValueError('Run utils/test_app_utils.py. This is a library file.')

import numpy as np
import itertools as it
import string

from core import config
from core.cs import CS
from utils.output_validation_utils import detect_discrepancies_in_test

# Get results given a matrix M and cycle times cts
def get_test_results(M, cts, algo):
  y = get_y_from_cts(cts)
  sure_list, unsure_list, neg_list, x, errors = _get_infected_lists(M, y, algo)
  return sure_list, unsure_list, neg_list, x

# Get y (viral loads in each test) from cycle times
def get_y_from_cts(cts):
  # First get the min cycle time. This corresponds to the test with the
  # maximum viral load. This sample will have the least amount of variance
  # in cycle time, so good to choose this as the baseline
  ctmin = np.min(cts)

  # Now filter out the positive tests
  # Treat 0 as -ve test as well. This will prevent end-user error such as
  # unchecking "Threshold Not Reached" in the app but not putting a value.
  bool_y = np.logical_and(cts < config.cycle_time_cutoff, cts > 0).astype(np.int32)
  #bool_y = (cts < config.cycle_time_cutoff).astype(np.int32)
  cts = cts * bool_y

  y = (1 + config.p) ** (ctmin - cts)
  y = y * bool_y

  return y

def build_cs_object(M, y):
  assert M is not None

  n = M.shape[1]
  t = M.shape[0]

  assert t == len(y)

  # unused params
  arr = np.zeros(n)
  mr = None
  d = 1
  s = 0.5
  l = 0.1

  cs = CS(n, t, s, d, l, arr, M, mr)
  return cs, t, n

# This function calls the decoding algorithm in class CS
# The algorithm chosen can be configured using config.app_algo
def _get_infected_lists(M, y, algo):
  cs, t, n = build_cs_object(M, y)
  bool_y = (y > 0).astype(np.int32)
  if algo == 'COMP':
    infected, infected_dd, score, tp, fp, fn, surep, unsurep,\
        num_infected_in_test = \
        cs.decode_comp_new(bool_y, compute_stats=False)
    x = np.zeros(n)
  else:
    x, infected, infected_dd, prob1, prob0, score, tp, fp, fn, uncon_negs, determined,\
        overdetermined, surep, unsurep, wrongly_undetected,\
        num_infected_in_test = cs.decode_lasso(y, algo, prefer_recall=False,
            compute_stats=False)

  errors = detect_discrepancies_in_test(t, bool_y, num_infected_in_test)
  sure_list, unsure_list, neg_list = _get_lists_from_infected(infected,
      infected_dd, n)

  return sure_list, unsure_list, neg_list, x, errors

# Get list of sure, unsure and negative people
def _get_lists_from_infected(infected, infected_dd, n):
  sure_list = []
  unsure_list = []
  neg_list = []
  for i in range(n):
    if infected_dd[i] == 1:
      sure_list.append(i+1)
    elif infected[i] == 1:
      unsure_list.append(i+1)
    else:
      neg_list.append(i+1)

  return sure_list, unsure_list, neg_list

# Detect errors using COMP and give more conservative results
def detect_errors_and_get_conservative_results(M, cts):
  # Run COMP and get errors
  y = get_y_from_cts(cts)
  bool_y = (y > 0).astype(np.int32)

  cs, t, n = build_cs_object(M, y)

  # Call COMP to get num_infected_in_tests
  _infected, _infected_dd, _score, _tp, _fp, _fn, _surep, _unsurep,\
      num_infected_in_test = \
      cs.decode_comp_new(bool_y, compute_stats=False)

  # Get discrepancies in COMP output
  errors = detect_discrepancies_in_test(t, bool_y, num_infected_in_test)

  # COMP will only output type 1 error
  assert errors['err2'] == 0 and len(errors['err2lst']) == 0
  assert len(errors['err1lst']) == errors['err1']
  total_errors = errors['err1']

  # If there are no errors, then we return
  if total_errors == 0:
    res = {}
    res['errors'] = False
    res['result_string'] = ''
    return res
  else:
    # Now get count of number of positive tests each sample is involved in
    # We will output a categorized list by number of positive tests
    num_positive_tests = cs.count_num_positive_tests(bool_y)

    # XXX: We consider only upto 3-positives. This will need to change later if our
    # matrix has more than 3 positives
    categorized_samples = { i:[] for i in range(4) }
    for sample, num_positive in enumerate(num_positive_tests):
      categorized_samples[num_positive].append(sample)

    # Now build result string
    result_string = get_result_string(errors['err1lst'], categorized_samples, t)
    res = {}
    res['errors'] = True
    res['result_string'] = result_string
    return res

def get_well_for_test(test, t):
  C = [f'{tup[0]}{tup[1]}' for tup in it.product(string.ascii_uppercase,range(1,13))]
  if t < 48:
    return C[test * 2]
  else:
    return C[test]

# Gets end-user-friendly result string in case of error
def get_result_string(err1lst, categorized_samples, t):
  assert len(err1lst) > 0
  header = '\nDiscrepancies were detected in the cycle time (CT) values entered by you.\n'
  header += 'They failed an internal consistency check.\n'
  header += 'Perhaps you forgot to input the CT value for some well(s)?\n'
  header += 'Other causes can be: \n'
  header += '  * contamination of some well from a nearby well\n'
  header += '  * very low viral load in some sample, due to which a mixed sample\n'
  header += '    did not reach cycle time threshold\n'
  header += '\nHere are the errors:\n\n'

  # List discrepancies detected by COMP
  errors_str = ''
  for test in err1lst:
    errors_str += f'Well {get_well_for_test(test, t)} is positive but no positive sample found\n'

  errors_str += '\n'
  # Give explanation to user about error categories
  explanation_str = ''
  explanation_str += 'We list the samples below, categorized by the number of tests'\
      ' they were positive in.\n\n'
  explanation_str += 'Samples appearing in 3 positive tests are most likely to be positive. \n'
  explanation_str += 'Samples appearing in 2 positive tests may also be positive\n'
  explanation_str += 'Samples appearing in 1 positive test are very unlikely to be positive\n'
  explanation_str += 'Samples appearing in 0 positive tests are negative\n\n'

  # Show categorized samples
  results_str = ''
  for tests in [3, 2, 1, 0]:
    results_str += f'Samples positive in {tests} tests: '
    results_str += ', '.join([str(item + 1) for item in categorized_samples[tests]
        ])
    results_str += '\n\n'

  final_str = header + errors_str + explanation_str + results_str
  return final_str


text:
import os
import sys

import numpy as np

import macro

def metropolis_hastings(valences,chemical_formula,iteration):

    # normalize and rounding
    mean_vect = chemical_formula[chemical_formula>1.0e-6]
    mean_sum = np.sum(mean_vect)
    mean_vect = mean_vect/mean_sum
    mean_vect = np.round(mean_vect,decimals=macro._ROUND)

    # if composition includes only one atom, return
    if len(mean_vect) == 1:
        return mean_vect, 0

    cnt = []
    for i in range(len(valences)):
        cnt.append(len(valences[i]))
    cnt = np.array(cnt,dtype=np.int32)

    buf = np.zeros(cnt.max(),dtype=np.int32)
    for i in range(len(cnt)):
        buf[cnt[i]-1] = 1
        # if there is not valence array, return
        if cnt[i] == 0:
            return mean_vect, 1

    # To reduce the complexity, if len(valence) has several values, return (must improve)
    num = 0
    for i in range(1,len(buf)):
        if buf[i] != 0:
            num = num+1
    if num > 1:
        return mean_vect, 1

    y_now = np.dot(mean_vect,valences)
    y_min = np.min(y_now)

    # if mean_vect is already balance, return
    if iteration == 0:
        return mean_vect, 0

    # set the covariance of gaussian (proposal distribution)
    cov = np.zeros((len(mean_vect),len(mean_vect)),dtype=np.float32)
    for i in range(len(mean_vect)):
        cov[i][i] = macro._SIGMA

    before_y_min = y_min
    before_mean_vect = mean_vect
    for ite in range(iteration):

        y_now = np.dot(mean_vect,valences)

        # if balance, return
        for i in range(len(y_now)):
            if abs(y_now[i]) < 1.0e-6:
                return mean_vect, 0

        x_star = np.random.multivariate_normal(mean_vect,cov,macro._SAMPLING)
        x_star = x_star.clip(min=1.0e-6)
        for i in range(len(x_star)):
            x_star[i] = x_star[i]/np.sum(x_star[i])
        x_star = np.round(x_star,macro._ROUND)

        # expand array list
        bak = np.dot(x_star,valences)
        y_star = np.zeros((macro._SAMPLING,len(y_now)),dtype=np.float32)
        for i in range(len(y_star)):
            y_star[i] = bak[i]

        A = np.exp(-y_star*y_star)/np.exp(-y_now*y_now)
        max_A = np.max(A)
        arg_A_x, arg_A_y = np.unravel_index(np.argmax(A),A.shape)

        if max_A > 1.0:
            max_A = 1.0
        p = np.random.random()
        if p < max_A and np.sum(mean_vect) > 1.0e-6:
            for i in range(len(mean_vect)):
                mean_vect[i] = x_star[arg_A_x][i]
        if y_star[arg_A_x][arg_A_y] < 1.0e-6:
            break

    return mean_vect, 0

max_stars_repo_name: yoshihidesawada/CompGAN

max_issues_repo_name: yoshihidesawada/CompGAN

max_forks_repo_name: yoshihidesawada/CompGAN


text:
import pytest
import numpy as np
from ..moreau_broto import moreau_broto
import pkg_resources

PATH = pkg_resources.resource_filename(__name__, 'test_data/')

def test_moreau_broto():
    "Test Moreau-Broto Autocorrelation"

    # load data
    X_list = open(PATH+'multiple.txt').read().splitlines()
    X_err = 'AGT2HT9'

    # test Moran's I
    mb = moreau_broto(X_list)

    # test array contents
    np.testing.assert_almost_equal(mb[0], np.array([
         0.30741363,  0.14572374,  0.24889737,  0.17774769,  0.31419645,
         0.59789614,  0.52090645,  0.54431542]))
    np.testing.assert_almost_equal(mb[1], np.array([
        -0.22151544,  0.48430045,  0.40262559,  0.65978741,  0.58163617,
         0.48620151,  1.57705014, -0.13977796]))
    np.testing.assert_almost_equal(mb[2], np.array([
         0.04011865, -0.05581411,  0.54491746,  0.48283298,  0.40420316,
         0.7053263 ,  0.07440376, -0.1527529]))

    # test ValueError (alphabetical)
    with pytest.raises(ValueError):
        mb_err = moreau_broto(X_err)

    # test ValueError (maximum lag)
    with pytest.raises(ValueError):
        mb_err = moreau_broto(X_list, d= 31)

    # test ValueError (d >= min_len)
    with pytest.raises(ValueError):
        mb_err = moreau_broto(X_err, d=7)

max_stars_repo_name: tadorfer/ProtClass

max_issues_repo_name: tadorfer/ProtClass

max_forks_repo_name: tadorfer/ProtClass


text:
import logging
import os

import numpy as np

def nonzeros(m, row):
    """ returns the non zeroes of a row in csr_matrix """
    for index in range(m.indptr[row], m.indptr[row+1]):
        yield m.indices[index], m.data[index]

def check_blas_config():
    """ checks to see if using OpenBlas/Intel MKL. If so, warn if the number of threads isn't set
    to 1 (causes severe perf issues when training - can be 10x slower) """
    if np.__config__.get_info('openblas_info') and os.environ.get('OPENBLAS_NUM_THREADS') != '1':
        logging.warning("OpenBLAS detected. Its highly recommend to set the environment variable "
                        "'export OPENBLAS_NUM_THREADS=1' to disable its internal multithreading")
    if np.__config__.get_info('blas_mkl_info') and os.environ.get('MKL_NUM_THREADS') != '1':
        logging.warning("Intel MKL BLAS detected. Its highly recommend to set the environment "
                        "variable 'export MKL_NUM_THREADS=1' to disable its internal "
                        "multithreading")

max_stars_repo_name: DeepEngagement/implicit

max_issues_repo_name: DeepEngagement/implicit

max_forks_repo_name: DeepEngagement/implicit


text:
import re
import os, sys

import numpy as np
import pandas as pd
import networkx as nx

from networkx.drawing.nx_agraph import graphviz_layout

def add_graphics_theme_to_nx_graph(
        nx_graph,
        edge_color=None,
        node_size_factor=50,
        edge_size_factor=500):
    """adjust nodes and edges
    """
    # node size, stroke
    for node_name, node_attrs in nx_graph.nodes(data=True):

        #node_size = nx_graph.nodes[node_name]["numexamples"] / float(node_size_factor)
        #node_size = nx_graph.nodes[node_name]["numexamples"] / float(nx_graph.graph["numexamples"])
        #node_size *= node_size_factor
        node_size = 100

        graphics = {
            "type": "ellipse",
            "w": node_size,
            "h": node_size,
            "fill": "#FFFFFF",
            "outline": "#000000",
            "width": 1.0,
            "fontSize": 14
        }

        if nx_graph.nodes[node_name].get("graphics") is not None:
            nx_graph.nodes[node_name]["graphics"].update(graphics)
        else:
            nx_graph.nodes[node_name]["graphics"] = graphics

    # edges
    for start_node, end_node in nx_graph.edges():
        for edge_idx in xrange(len(nx_graph[start_node][end_node])):

            #edge_width = nx_graph[start_node][end_node][edge_idx]["numexamples"] / float(
            #    edge_size_factor)

            #edge_width = nx_graph[start_node][end_node][edge_idx]["numexamples"] / float(
            #    nx_graph.graph["numexamples"])
            #edge_width *= edge_size_factor
            edge_width = 1.0

            graphics = {
                "type": "arc",
                "width": edge_width,
                "targetArrow": "delta"
            }

            if edge_color is not None:
                graphics["fill"] = edge_color

            if nx_graph[start_node][end_node][edge_idx].get("graphics") is not None:
                nx_graph[start_node][end_node][edge_idx]["graphics"].update(graphics)
            else:
                nx_graph[start_node][end_node][edge_idx]["graphics"] = graphics

    return None

def stringize_nx_graph(nx_graph):
    """preparatory function for writing out to gml
    """
    # graph attributes
    for key in nx_graph.graph.keys():
        if isinstance(nx_graph.graph[key], (list, set, np.ndarray)):
            nx_graph.graph[key] = ",".join([
                str(val) for val in list(nx_graph.graph[key])])

    # node attributes
    for node_name, node_attrs in nx_graph.nodes(data=True):
        for key in node_attrs.keys():
            if isinstance(nx_graph.nodes[node_name][key], (list, set, np.ndarray)):
                nx_graph.nodes[node_name][key] = ",".join([
                    str(val) for val in nx_graph.nodes[node_name][key]])
        # adjust node name for nice output in cytoscape
        new_node_name = re.sub(r"HCLUST.\d+_", "", node_name)
        new_node_name = new_node_name.replace(".UNK.0.A", "")
        nx_graph.nodes[node_name]["name"] = new_node_name

    # edge attributes
    for start_node, end_node in nx_graph.edges():
        for edge_idx in xrange(len(nx_graph[start_node][end_node])):
            edge_attrs = nx_graph[start_node][end_node][edge_idx]
            for key in edge_attrs.keys():
                if isinstance(edge_attrs[key], (list, set, np.ndarray)):
                    nx_graph[start_node][end_node][edge_idx][key] = ",".join([
                        str(val) for val in nx_graph[start_node][end_node][edge_idx][key]])

    return nx_graph

def main():
    """build a network view
    """
    # files
    summary_file = sys.argv[1]
    pwms_to_tfs_file = sys.argv[2]
    expressed_tfs_file = sys.argv[3] # TODO

    # TODO pull in num regions to resize things? but complicated with overlaps etc
    # TODO edit edges with type of interaction
    # TODO may want to color by trajectory, to demonstrate waves of trajectory

    # read in data
    summary = pd.read_csv(summary_file, sep="\t")
    pwms_to_tfs = pd.read_csv(pwms_to_tfs_file, sep="\t")
    pwms_to_tfs = pwms_to_tfs[pwms_to_tfs["expressed"].notna()]
    pwms_to_filt_tfs = {} # dict: key - pwm names, vals - dict of ensembl ids to hgnc ids
    for line_idx in range(pwms_to_tfs.shape[0]):
        pwm_info = pwms_to_tfs.iloc[line_idx,:]
        pwm_name = pwm_info["hclust_model_name"]
        pwm_to_tf = dict(zip(pwm_info["expressed"].split(";"), pwm_info["expressed_hgnc"].split(";")))
        pwms_to_filt_tfs[pwm_name] = pwm_to_tf

    # filter expressed hgncs for dynamic ones only
    tfs_filt = pd.read_csv(expressed_tfs_file, sep="\t", index_col=0)
    for pwm_name in pwms_to_filt_tfs.keys():
        tfs_tmp = pwms_to_filt_tfs[pwm_name]
        for ensembl_tf in tfs_tmp.keys():
            if ensembl_tf not in tfs_filt.index:
                del tfs_tmp[ensembl_tf]
        if len(tfs_tmp.keys()) == 0:
            del pwms_to_filt_tfs[pwm_name]
        pwms_to_filt_tfs[pwm_name] = tfs_tmp

    # add in tfs column
    tf1 = []
    for pwm in summary["pwm1"]:
        tf_str = []
        for ensembl_id in pwms_to_filt_tfs[pwm]:
            tf_str.append(pwms_to_filt_tfs[pwm][ensembl_id])
            # TODO try add in max point
            expression = tfs_filt.loc[ensembl_id,:]
            max_idx = np.argmax(expression.values)
            tf_str.append(str(max_idx))
        tf_str = (";").join(tf_str)
        tf1.append(tf_str)
    summary["tf1"] = tf1

    tf2 = []
    for pwm in summary["pwm2"]:
        tf_str = []
        for ensembl_id in pwms_to_filt_tfs[pwm]:
            tf_str.append(pwms_to_filt_tfs[pwm][ensembl_id])
            expression = tfs_filt.loc[ensembl_id,:]
            max_idx = np.argmax(expression.values)
            tf_str.append(str(max_idx))
        tf_str = (";").join(tf_str)
        tf2.append(tf_str)
    summary["tf2"] = tf2

    # remove failed rules
    summary = summary[~summary["interaction"].str.contains("FAILED")]

    # make graph
    graph = nx.from_pandas_edgelist(summary, "tf1", "tf2")

    # set up positions
    #pos = graphviz_layout(graph, prog="dot")
    pos = graphviz_layout(graph, prog="neato")
    scale_factor = 3
    for key in pos.keys():
        coords = pos[key]
        pos[key] = {"x": scale_factor*coords[0], "y": -scale_factor*coords[1]}
    nx.set_node_attributes(graph, pos, "graphics") # note this is diff from v1 to v2 in networkx

    # add graphics
    add_graphics_theme_to_nx_graph(graph)

    # write gml
    out_file = "summary.gml"
    nx.write_gml(stringize_nx_graph(graph), out_file, stringizer=str)

    # tfs: for each tf, get gene column

    return

main()

max_stars_repo_name: vervacity/ggr-project

max_issues_repo_name: vervacity/ggr-project

max_forks_repo_name: vervacity/ggr-project

reason: import numpy,import networkx,from networkx


text:
# test_simulator.py

import pandas as pd
import numpy as np
import torch
from intersim.utils import get_map_path, get_svt, SVT_to_stateactions
from intersim.viz import animate
from intersim import collisions
import gym

import os
opj = os.path.join
outdir  = opj('tests','output')
if not os.path.isdir(outdir):
    os.mkdir(outdir)
filestr = opj(outdir,'test_state_targeting_states')

def test_state_targeting():

    svt, svt_path = get_svt()
    osm = get_map_path()
    print('SVT path: {}'.format(svt_path))
    print('Map path: {}'.format(osm))

    states, actions = SVT_to_stateactions(svt)

    # animate from environment
    env = gym.make('intersim:intersim-v0', svt=svt, map_path=osm,
        min_acc=-np.inf, max_acc=np.inf)
    env.reset()
    done = False
    i = 0
    obs, infos = [], []
    env_states = []
    nnis = []
    max_devs = []
    while not done and i < len(actions):
        env.render()
        env_state = env.projected_state
        env_states.append(env_state)
        nni = ~torch.isnan(env_state[:,0])
        nnis.append(nni)
        norms = torch.norm(env_state[nni,:2]-states[i,nni,:2], dim=1)
        if len(norms)>0:
            max_devs.append(norms.max())
        ob, r, done, info = env.step(env.target_state(svt.simstate[i+1], mu=0.002))
        obs.append(ob)
        infos.append(info)
        i+=1
    env.render()

    # assert small max norm and no collisions

    assert max(max_devs) < 10, "Maximum environment deviation from track: %f m" %(max(max_devs))
    x = torch.stack([ob['state'] for ob in obs])
    cols = collisions.check_collisions_trajectory(x, svt.lengths, svt.widths)
    assert ~torch.any(cols), 'Error: Collisions found at indices {}'.format(cols.nonzero(as_tuple=True))

    # env.close(filestr=opj(outdir,'test_state_targeting_env'))
    # animate from states
    # animate(osm, states[:n_frames+1], svt._lengths, svt._widths, filestr=filestr)

max_stars_repo_name: sisl/InteractionSimulator

max_issues_repo_name: sisl/InteractionSimulator

max_forks_repo_name: sisl/InteractionSimulator


text:
from typing import (
    Any,
    Callable,
    List,
    Mapping,
    NamedTuple,
    Optional,
    Sequence,
    Tuple,
    Union,
)

import numpy as np
import torch

class PaddedTensor(NamedTuple):
    data: torch.Tensor
    sizes: torch.Tensor

    @classmethod
    def build(cls, data: torch.Tensor, sizes: torch.Tensor):
        assert isinstance(data, torch.Tensor)
        assert isinstance(sizes, torch.Tensor)
        assert sizes.dim() == 2, "PaddedTensor.sizes must have 2 dimensions"
        assert sizes.size(1) in (2, 3), (
            "PaddedTensor.sizes is incorrect: "
            "expected=2 (HxW) or 3 (CxHxW), "
            f"found={sizes.size(1)}"
        )
        assert data.size(0) == sizes.size(0), (
            f"Batch size {sizes.size(0)} does not match "
            f"the number of samples in the batch {data.size(0)}"
        )
        return cls(data, sizes)

    def __repr__(self) -> str:
        return (
            f"PaddedTensor("
            f"data.size()={list(self.data.size())}, "
            f"sizes={self.sizes.tolist()}, "
            f"device={str(self.data.device)})"
        )

    @property
    def device(self) -> torch.device:
        return self.data.device

def by_descending_width(x):
    # C x H x W
    return -x["img"].size(2)

class PaddingCollater:
    def __init__(self, sizes: Any, sort_key: Callable = None):
        self._sizes = sizes
        self._sort_key = sort_key

    def __call__(self, batch: Any) -> torch.Tensor:
        if self._sort_key:
            batch = sorted(batch, key=self._sort_key)
        return self.collate(batch, self._sizes)

    @staticmethod
    def get_max_sizes(
        batch: List[torch.Tensor], sizes: Optional[Tuple[Union[int, None], ...]] = None
    ) -> Tuple[int, ...]:
        # All tensors in the batch must have the same number of dimensions
        dim = batch[0].dim()
        assert all(x.dim() == dim for x in batch)
        max_sizes = [len(batch)]
        for d in range(dim):
            max_v = max(x.size(d) for x in batch)
            min_v = min(x.size(d) for x in batch)
            if sizes and sizes[d] is not None:
                assert max_v == min_v == sizes[d]
            max_sizes.append(max_v)
        return tuple(max_sizes)

    @staticmethod
    def collate_tensors(
        batch: List[torch.Tensor], max_sizes: Tuple[int, ...]
    ) -> torch.Tensor:
        out = batch[0].new_zeros(size=max_sizes)
        for i, x in enumerate(batch):
            batch_tensor = out[i]
            for d in range(batch[0].dim()):
                batch_tensor = batch_tensor.narrow(d, 0, x.size(d))
            batch_tensor.add_(x)
        return out

    def collate(self, batch: Any, sizes: Any) -> Any:
        elem, elem_type = batch[0], type(batch[0])
        if isinstance(elem, torch.Tensor):
            if any(s is None for s in sizes):
                max_sizes = PaddingCollater.get_max_sizes(batch, sizes)
                x = PaddingCollater.collate_tensors(batch, max_sizes)
                xs = torch.stack([torch.tensor(x.size()) for x in batch])
                return PaddedTensor.build(x, xs)
            return torch.stack(batch)
        if isinstance(elem, np.ndarray):
            return self.collate([torch.from_numpy(b) for b in batch], sizes)
        if isinstance(elem, Mapping):
            return {
                k: self.collate([d[k] for d in batch], sizes[k])
                if k in sizes
                else [d[k] for d in batch]
                for k in elem
            }
        if isinstance(elem, Sequence):
            return [self.collate(b, s) for b, s in zip(batch, sizes)]
        raise TypeError(
            f"Batch must contain tensors, numbers, dicts or lists. Found {elem_type}"
        )

max_stars_repo_name: eivtho/PyLaia

max_issues_repo_name: eivtho/PyLaia

max_forks_repo_name: jpuigcerver/PyLaia


text:
import sys, os
import numpy as np
import tensorflow as tf
import random

from bert_tagging import DataProcessor, BertTagging
import modeling
import optimization
import time
from tagging_eval import score_f
tf.logging.set_verbosity(tf.logging.ERROR)

def evaluate(FLAGS, sess, model, data_processor, label_list=None):
    gpuid = FLAGS.gpuid
    max_sen_len = FLAGS.max_sen_len
    train_path = FLAGS.train_path
    test_file = FLAGS.test_path
    out_dir = FLAGS.output_dir
    batch_size = 20
    EPOCH = FLAGS.epoch
    init_bert_dir = FLAGS.init_bert_path
    learning_rate = FLAGS.learning_rate
    vocab_file = '%s/vocab.txt' % init_bert_dir
    init_checkpoint = '%s/bert_model.ckpt' % init_bert_dir
    bert_config_path = '%s/bert_config.json'% init_bert_dir

    test_num = data_processor.num_examples
    test_data = data_processor.build_data_generator(batch_size)
    iterator = test_data.make_one_shot_iterator()
    input_ids, input_mask, segment_ids, lmask, label_ids, masked_sample = iterator.get_next()
    pred_loss, pred_result, gold_result, gold_mask, r_loss = model.create_model(input_ids, input_mask, segment_ids, lmask, label_ids, batch_size=batch_size, masked_sample=masked_sample, is_training=False)

    label_list = data_processor.label_list
    ans_c, ans_py, ans = [], [], []
    all_inputs, all_golds, all_preds = [], [], []
    all_fusino_preds = []
    all_inputs_sent, all_golds_sent, all_preds_sent = [], [], []
    for step in range(test_num // batch_size):
        inputs, loss_value, preds, golds, gmask = sess.run([input_ids, pred_loss, pred_result, gold_result, gold_mask])
        for k in range(batch_size):
            tmp1, tmp2, tmp3, tmps4, tmps5, tmps6, tmps7 = [], [], [], [], [], [], []
            for j in range(max_sen_len):
                if gmask[k][j] == 0: continue
                all_golds.append(golds[k][j])
                all_preds.append(preds[k][j])
                all_inputs.append(inputs[k][j])
                tmp1.append(label_list[golds[k][j]])
                tmp2.append(label_list[preds[k][j]])
                tmp3.append(label_list[inputs[k][j]])

            all_golds_sent.append(tmp1)
            all_preds_sent.append(tmp2)
            all_inputs_sent.append(tmp3)

    all_golds = [label_list[k] for k in all_golds]
    all_preds = [label_list[k] for k in all_preds]
    all_inputs = [label_list[k] for k in all_inputs]

    print('zi result:')
    p, r, f = score_f((all_inputs, all_golds, all_preds), only_check=False)

    return f

def train(FLAGS):
    gpuid = FLAGS.gpuid
    max_sen_len = FLAGS.max_sen_len
    train_path = FLAGS.train_path
    test_file = FLAGS.test_path
    test_file_merr = FLAGS.test_path_merr
    out_dir = FLAGS.output_dir
    batch_size = FLAGS.batch_size
    EPOCH = FLAGS.epoch
    alpha = FLAGS.alpha
    init_bert_dir = FLAGS.init_bert_path
    learning_rate = FLAGS.learning_rate
    vocab_file = '%s/vocab.txt' % init_bert_dir
    init_checkpoint = '%s/bert_model.ckpt' % init_bert_dir
    bert_config_path = '%s/bert_config.json'% init_bert_dir

    if os.path.exists(out_dir) is False:
        os.mkdir(out_dir)

    os.environ["CUDA_VISIBLE_DEVICES"] = gpuid
    keep_prob = FLAGS.keep_prob
    print('test_file=', test_file)
    test_data_processor = DataProcessor(test_file, max_sen_len, vocab_file, out_dir, label_list=None, is_training=False)
    print('test_file_merr=', test_file_merr)
    test_data_processor_merr = DataProcessor(test_file_merr, max_sen_len, vocab_file, out_dir, label_list=None, is_training=False)
    print('train_file=', train_path)
    data_processor = DataProcessor(train_path, max_sen_len, vocab_file, out_dir, label_list=None, is_training=True)

    train_num = data_processor.num_examples
    train_data = data_processor.build_data_generator(batch_size)
    iterator = train_data.make_one_shot_iterator()
    input_ids, input_mask, segment_ids, lmask, label_ids, masked_sample = iterator.get_next()

    model = BertTagging(bert_config_path, num_class=len(data_processor.get_label_list()), max_sen_len=max_sen_len, alpha=alpha, keep_prob=keep_prob)
    (loss, probs, golds, mask, r_loss) = model.create_model(input_ids, input_mask, segment_ids, lmask, label_ids, batch_size=batch_size, masked_sample=masked_sample, is_training=True)

    tf_config = tf.ConfigProto(log_device_placement=False)
    tf_config.gpu_options.allow_growth = True
    with tf.Session(config=tf_config) as sess:
        if init_checkpoint is not None:
            tvars = tf.trainable_variables()
            (assignment_map, initialized_variable_names) = modeling.get_assignment_map_from_checkpoint(tvars, init_checkpoint)
            keys = [x for x in assignment_map.keys()]
            for key in keys:
                print(key, '\t', assignment_map[key])
            tf.train.init_from_checkpoint(init_checkpoint, assignment_map)

        num_steps = train_num // batch_size * EPOCH
        num_warmup_steps = num_steps // 10
        train_op = optimization.create_optimizer(loss, learning_rate, num_steps, num_warmup_steps, use_tpu=False)

        init = tf.global_variables_initializer()
        sess.run(init)

        loss_values = []
        r_loss_values = []
        saver = tf.train.Saver()
        best_score = 0.0
        best_model_path = os.path.join(out_dir, 'best.ckpt')
        total_step = 0
        for epoch in range(EPOCH):
            for step in range(int(train_num / batch_size)):
                total_step += 1
                start_time = time.time()
                train_loss, _, train_r_loss = sess.run([loss,  train_op, r_loss])
                loss_values.append(train_loss)
                r_loss_values.append(train_r_loss)
                if step % 500 == 0:
                    duration = time.time() - start_time
                    examples_per_sec = float(duration) / batch_size
                    format_str = ('Epoch {} step {},  train loss = {:.4f},{:.4f},{:.4f} ( {:.4f} examples/sec; {:.4f} ''sec/batch)')
                    print (format_str.format(epoch, step, np.mean(loss_values),np.mean(loss_values[-500:]),np.mean(r_loss_values[-500:]), examples_per_sec, duration))
                    loss_values = loss_values[-500:]
                    r_loss_values = r_loss_values[-500:]
                    print('multi-error result:')
                    evaluate(FLAGS, sess, model, test_data_processor_merr)
                    print('overall result:')
                    f1 = evaluate(FLAGS, sess, model, test_data_processor)
                    if f1 > best_score:
                        saver.save(sess, best_model_path)
                        best_score = f1
                    sys.stdout.flush()
            f1 = evaluate(FLAGS, sess, model, test_data_processor)
            if f1 > best_score:
                saver.save(sess, best_model_path)
                best_score = f1
            sys.stdout.flush()
        print ('best f value:', best_score)

if __name__ == '__main__':

    flags = tf.flags
    ## Required parameters
    flags.DEFINE_string("gpuid", '0', "The gpu NO. ")

    ## Optional
    flags.DEFINE_string("train_path", '', "train path ")
    flags.DEFINE_string("test_path", '', "train path ")
    flags.DEFINE_string("test_path_merr", '', "train path ")
    flags.DEFINE_string("output_dir", '', "out dir ")
    flags.DEFINE_string("init_bert_path", '', "out dir ")
    flags.DEFINE_string("label_list", '', 'max_sen_len')
    flags.DEFINE_integer("max_sen_len", 64, 'max_sen_len')
    flags.DEFINE_integer("batch_size", 32, 'batch_size')
    flags.DEFINE_integer("single_text", '0', 'single_text')
    flags.DEFINE_integer("epoch", 2, 'batch_size')
    flags.DEFINE_float("learning_rate", 5e-5, 'filter_punc')
    flags.DEFINE_float("keep_prob", 0.9, 'keep prob in dropout')
    flags.DEFINE_float("alpha", 0.05, 'trade-off factor')

    flags.mark_flag_as_required('gpuid')
    flags.mark_flag_as_required('train_path')
    flags.mark_flag_as_required('test_path')
    flags.mark_flag_as_required('init_bert_path')
    flags.mark_flag_as_required('output_dir')
    flags.mark_flag_as_required('label_list')
    flags.mark_flag_as_required('max_sen_len')

    FLAGS = flags.FLAGS
    print ('Confings:')
    print ('\tlearning_rate=', FLAGS.learning_rate)
    print ('\ttrain_path=', FLAGS.train_path)
    print ('\ttest_path=', FLAGS.test_path)
    print ('\tinit_bert_path=', FLAGS.init_bert_path)
    print ('\talpha=', FLAGS.alpha)
    print ('\tmax_sen_len=', FLAGS.max_sen_len)
    train(FLAGS)

max_stars_repo_name: liushulinle/CRACSpell

max_issues_repo_name: liushulinle/CRASpell

max_forks_repo_name: liushulinle/CRASpell


text:
import time
import matplotlib.pyplot as plt
import matplotlib.animation as animation
from mpl_toolkits.axes_grid1 import make_axes_locatable
from IPython.display import HTML

import numpy as np
import numpy.matlib
import matplotlib.pyplot as plt
import copy as copy
import math
import os
import scipy.io as sio
from scipy.signal import hilbert
from shutil import copy2

import sys

__all__ = ['Envelopt_fun','Correlate_fun']

__docformat__ = "restructuredtext en"

def Envelopt_fun(dobs, dpred, p=2.0):
    dpred_Hilbert = hilbert(dpred, axis=0).imag
    dobs_Hilbert = hilbert(dobs, axis=0).imag

    dpred_envelope = dpred**2.0 + dpred_Hilbert**2.0
    dobs_envelope = dobs**2.0 + dobs_Hilbert**2.0

    resid = dpred_envelope**(p/2.0) - dobs_envelope**(p/2.0)

    denvelope_ddata = p * dpred_envelope**(p/2.0 - 1.0) * dpred
    adjoint_src = denvelope_ddata * resid

    denvelope_ddataH = p * dpred_envelope**(p/2.0 - 1.0) * dpred_Hilbert
    adjoint_src += (-hilbert(denvelope_ddataH * resid, axis=0)).imag

    return -resid, -adjoint_src

def Correlate_fun(dobs, dpred, dt):
    shape_dobs = np.shape(dobs)
    n_correlate_data = shape_dobs[0]
    resid = np.zeros([n_correlate_data, shape_dobs[1]])
    adjoint_src = np.zeros([shape_dobs[0], shape_dobs[1]])
    W  = np.zeros(n_correlate_data)
    if np.mod(n_correlate_data, 2) == 0:
        W[0:n_correlate_data//2] = np.linspace(-dt, -dt*(shape_dobs[0])/2.0, n_correlate_data/2)
        W[n_correlate_data//2:n_correlate_data] = np.flipud(W[0:n_correlate_data//2])
    else:
        W[0:(n_correlate_data+1)//2] = np.linspace(0.0, -dt*(shape_dobs[0]-1)/2.0, (n_correlate_data+1)/2)
        W[(n_correlate_data-1)//2:n_correlate_data] = np.flipud(W[0:(n_correlate_data+1)//2])

    for i in range(0, shape_dobs[1]):
        correlate_data = correlate_fun(dobs[:,i], dpred[:,i])
        Wf = W * correlate_data
        correlate_data_norm2 = np.dot(correlate_data, correlate_data)
        Wf_norm2 = np.dot(Wf, Wf)
        resid[:, i] = Wf / np.sqrt(correlate_data_norm2)

        adjoint_src[:,i] = (2.0*correlate_data_norm2*correlate_fun(dobs[:,i], W*Wf, mode='adj') - 2.0*Wf_norm2*correlate_fun(dobs[:,i], correlate_data, mode='adj')) / correlate_data_norm2**2.0

    return resid, -1*adjoint_src

max_stars_repo_name: zfang-slim/FTools

max_issues_repo_name: zfang-slim/FTools

max_forks_repo_name: zfang-slim/FTools

reason: import numpy,import scipy,from scipy


text:
import numpy as np
from tetrisgym import Move, TetrisGym, TETRIS_HEIGHT, TETRIS_WIDTH
from collections import deque
from keras.models import Sequential
from keras.layers import Dense, MaxPooling2D, Conv2D, Flatten
from keras.optimizers import Adam

import matplotlib.pyplot as plt
import random
import time

class DQN:
    """
    An agent that explores the game space and learns to play.
    """
    def __init__(self):
        self.memory = deque(maxlen=10000)
        self.action_size = 4
        self.state_size = 200
        self.improvement_rate = 0.95
        self.randomness = 1.0
        self.randomness_min = 0.01
        self.randomness_decay = 0.999
        self.learning_rate = 0.001
        self.model = self.compile_model()

    def compile_model(self):
        """ Returns a compiled network """
        model = Sequential()
        model.add(Conv2D(32, kernel_size=(3, 3), strides=(1, 1),
                         activation='relu',
                         input_shape=(TETRIS_HEIGHT, TETRIS_WIDTH, 1)))
        model.add(MaxPooling2D(pool_size=(2, 2), strides=(2, 2)))
        model.add(Conv2D(64, (2, 2), activation='relu'))
        model.add(MaxPooling2D(pool_size=(2, 2)))
        model.add(Flatten())
        model.add(Dense(1000, activation='relu'))
        model.add(Dense(self.action_size, activation='softmax'))
        adam = Adam(lr=self.learning_rate)
        model.compile(loss='mse', optimizer=adam)

        return model

    def act(self, state):
        """ Predicts an action on a state, or acts randomly depending on state exploration. """
        if np.random.rand() <= self.randomness:
            return random.randrange(self.action_size)
        act_values = self.model.predict(state.reshape((1, TETRIS_HEIGHT, TETRIS_WIDTH, 1)))
        return np.argmax(act_values[0])

    def integer_to_action(self, input):
        """ Translate the generated action from an integer to an action the game understands. """
        if input == 0:
            return Move.Down
        elif input == 1:
            return Move.Left
        elif input == 2:
            return Move.Right
        elif input == 3:
            return Move.Rotate

    def store_experience(self, state, action, reward, next_state, done):
        """ Store the experience for replay. """
        self.memory.append((state, action, reward, next_state, done))

    def replay_experiences(self, batch_size):
        """ Train on a batch of stored experiences. """
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            next_state = next_state
            state = state
            old_q = self.model.predict(state.reshape((1, TETRIS_HEIGHT, TETRIS_WIDTH, 1)))
            new_q = self.model.predict(next_state.reshape((1, TETRIS_HEIGHT, TETRIS_WIDTH, 1)))

            if not done:
                scaled_old = (1 - self.improvement_rate) * old_q[0][int(action)]
                scaled_new = self.improvement_rate * (reward + new_q[0][int(action)])
                target = scaled_old + scaled_new

                old_q[0][int(action)] = target
            self.model.fit(state.reshape((1, TETRIS_HEIGHT, TETRIS_WIDTH, 1)), old_q, epochs=1, verbose=0)
        if self.randomness > self.randomness_min:
            self.randomness *= self.randomness_decay

gym = TetrisGym()
batch_size = 64
finished = False
agent = DQN()
scores = []
iterations = 2000

for game_iteration in range(iterations):
    gym = TetrisGym()
    current_state = gym.reset_game()
    runtime = 0
    while not gym.game_over:
        action = agent.act(current_state)
        update, current_state, done = gym.update(agent.integer_to_action(action))
        gym.render()
        time.sleep(0.01)

        if update is None:
            pass
        else:
            for state, action, reward, next_state in update:
                reward = reward
                next_state = next_state
                state = state
                agent.store_experience(state, action, reward, next_state, False)
                state = next_state
        if done:
            gym.render()
            print("{}/{}, score: {}, randomness: {:.2}"
                  .format(game_iteration, iterations, runtime, agent.randomness))
            scores.append(runtime)
            break
        else:
            runtime += 1
    if len(agent.memory) > batch_size:
        agent.replay_experiences(batch_size)

plt.plot(scores)
plt.ylabel('Steps Survived')
plt.xlabel('Episode')
plt.show()
print ("done!")

max_stars_repo_name: bretcolloff/tetrisdqn

max_issues_repo_name: bretcolloff/tetrisdqn

max_forks_repo_name: bretcolloff/tetrisdqn


text:
import numpy as np

###### Constants you should be using in your code ######

# You are welcome to alter these SIZES, but you MUST make
# DISPLAY_SIZE a multiple of GRID_SIZE, or else things will break
# For example, by default DISPLAY_SIZE = 540 = 13 * 40 = 13 * GRID_SIZE
# So, it creates a 13x13 grid, where the border is the wall, and the inner
# 12x12 is playable space
DISPLAY_SIZE = 560
GRID_SIZE = 40

UP = 0
DOWN = 1
LEFT = 2
RIGHT = 3

######

WALL_SIZE = GRID_SIZE

# Colors for pygame
WHITE = (255, 255, 255)
RED = (255, 0, 0)
BLUE = (72, 61, 139)
BLACK = (0, 0, 0)
GREEN = (0, 255, 0)

NUM_FOOD_DIR_X=3
NUM_FOOD_DIR_Y=3
NUM_ADJOINING_WALL_X_STATES=3
NUM_ADJOINING_WALL_Y_STATES=3
NUM_ADJOINING_BODY_TOP_STATES=2
NUM_ADJOINING_BODY_BOTTOM_STATES=2
NUM_ADJOINING_BODY_LEFT_STATES=2
NUM_ADJOINING_BODY_RIGHT_STATES=2
NUM_ACTIONS = 4

CHECKPOINT = 'checkpoint.npy'

def create_q_table():
	return np.zeros((NUM_FOOD_DIR_X, NUM_FOOD_DIR_Y, NUM_ADJOINING_WALL_X_STATES, NUM_ADJOINING_WALL_Y_STATES,
					 NUM_ADJOINING_BODY_TOP_STATES, NUM_ADJOINING_BODY_BOTTOM_STATES, NUM_ADJOINING_BODY_LEFT_STATES,
					 NUM_ADJOINING_BODY_RIGHT_STATES, NUM_ACTIONS))

def sanity_check(arr):
	if (type(arr) is np.ndarray and
		arr.shape==(NUM_FOOD_DIR_X, NUM_FOOD_DIR_Y, NUM_ADJOINING_WALL_X_STATES, NUM_ADJOINING_WALL_Y_STATES,
					 NUM_ADJOINING_BODY_TOP_STATES, NUM_ADJOINING_BODY_BOTTOM_STATES, NUM_ADJOINING_BODY_LEFT_STATES,
					 NUM_ADJOINING_BODY_RIGHT_STATES,NUM_ACTIONS)):
		return True
	else:
		return False

def save(filename, arr):
	if sanity_check(arr):
		np.save(filename,arr)
		return True
	else:
		print("Failed to save model")
		return False

def load(filename):
	try:
		arr = np.load(filename)
		if sanity_check(arr):
			print("Loaded model successfully")
			return arr
		print("Model loaded is not in the required format")
		return None
	except:
		print("Filename doesnt exist")
		return None

max_stars_repo_name: annabellej/snake-reinforcement-learning

max_issues_repo_name: annabellej/snake-reinforcement-learning

max_forks_repo_name: annabellej/snake-reinforcement-learning


text:
import os
import sys
import json
import random
import argparse
import datetime
import numpy as np
from pathlib import Path
from easydict import EasyDict as edict

import torch
import torch.backends.cudnn as cudnn
from torch.utils.data import DataLoader

from lib.utils import Logger
import lib.data as datasets
import lib.trainer as trainer

def main(config, resume, save_dir, log_dir):
    # DATA LOADERS
    train_set = getattr(datasets, config.datamanager.type)(
        config.datamanager.root, config.datamanager.dataset_dir,
        width=config.datamanager.width,
        height=config.datamanager.height,
        mean=config.datamanager.norm_mean,
        std=config.datamanager.norm_std,
        mode="train")
    val_set = getattr(datasets, config.datamanager.type)(
        config.datamanager.root, config.datamanager.dataset_dir,
        width=config.datamanager.width,
        height=config.datamanager.height,
        mean=config.datamanager.norm_mean,
        std=config.datamanager.norm_std,
        mode="test")

    train_loader = DataLoader(dataset=train_set, num_workers=config.datamanager.workers,
                              batch_size=config.datamanager.batch_size_train, shuffle=True,
                              pin_memory=config.datamanager.pin_memory)
    val_loader = DataLoader(dataset=val_set, num_workers=config.datamanager.workers,
                            batch_size=config.datamanager.batch_size_test, shuffle=False,
                            pin_memory=config.datamanager.pin_memory)

    print(f'\n{train_loader.dataset}\n')
    print(f'\n{val_loader.dataset}\n')

    # TRAINING
    runner = getattr(trainer, config.trainer.type)(
        config=config,
        resume=resume,
        save_dir=save_dir,
        log_dir=log_dir,
        train_loader=train_loader,
        val_loader=val_loader)

    runner.train()

if __name__ == '__main__':
    # PARSE THE ARGS
    parser = argparse.ArgumentParser(description='PyTorch Training')
    parser.add_argument('-c', '--config', default='./configs/selfie2anime.json', type=str,
                        help='Path to the config file (default: PSPNet.json)')
    parser.add_argument('-r', '--resume',
                        default="",
                        type=str,
                        help='Path to the .pth model checkpoint to resume training')
    parser.add_argument('-d', '--device', default=None, type=str,
                        help='indices of GPUs to enable (default: all)')
    args = parser.parse_args()

    # Setup seeds
    torch.manual_seed(123)
    torch.cuda.manual_seed(123)
    np.random.seed(123)
    random.seed(123)
    cudnn.benchmark = True

    config = json.load(open(args.config))
    if args.resume:
        config = torch.load(args.resume)['config']
    if args.device:
        os.environ["CUDA_VISIBLE_DEVICES"] = args.device
    config = edict(config)

    start_time = datetime.datetime.now().strftime('%m-%d_%H-%M')
    save_dir = Path('.').resolve() / config.trainer.save_dir / config.name / start_time
    log_dir = save_dir / 'logs'
    log_dir.mkdir(parents=True, exist_ok=True)
    sys.stdout = Logger(log_dir / 'train.txt')

    main(config, args.resume, save_dir, log_dir)


text:
import matplotlib.pyplot as plt
import numpy as np
from numpy.linalg import svd

from sympy.matrices import Matrix
from sympy import init_printing
from sympy import latex
import math

init_printing()

def printm(m):
    display(Matrix(m))

def printl(m):
    print(latex(Matrix(np.array(m*100, dtype=int)/100)))

def gen_rank_k(k, U, W, Vt):
    temp = np.matmul(U[:,:k], np.diag(W[:k]))
    return np.matmul(temp, Vt[:k, :])

reason: import numpy,from numpy,from sympy