text:
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import unittest

import numpy as np

from measurement_stats import distributions
from measurement_stats.distributions import boxes
from measurement_stats import value

def get_area_under_curve(x_values, y_values):
    area = 0.0
    for i in range(len(x_values) - 1):
        dx = x_values[i + 1] - x_values[i]
        area += dx * y_values[i]
    return area

class TestDensity(unittest.TestCase):

    def test_singleDensity(self):
        """ A single measurement value should have a total probability of
            unity
        """

        x_values = np.linspace(-10.0, 10.0, 100)
        measurements = [value.ValueUncertainty()]

        dist = distributions.Distribution(measurements=measurements)

        area = get_area_under_curve(x_values, dist.probabilities_at(x_values))
        self.assertAlmostEqual(area, 1.0, 3)

    def test_doubleDensityOverlap(self):
        """ Two overlapping measurement values should have a total probability
            of unity
        """
        x_values = np.linspace(-10.0, 10.0, 100)
        measurements = [value.ValueUncertainty(), value.ValueUncertainty()]

        dist = distributions.Distribution(measurements=measurements)
        area = get_area_under_curve(x_values, dist.probabilities_at(x_values))
        self.assertAlmostEqual(area, 1.0, 3)

    def test_doubleDensityOffset(self):
        """ Two measurements with different values and uncertainties should
            still result in a total probability of unity
        """

        x_values = np.linspace(-10.0, 25.0, 100)
        measurements = [
            value.ValueUncertainty(),
            value.ValueUncertainty(2.0, 2.0)
        ]

        dist = distributions.Distribution(measurements=measurements)
        area = get_area_under_curve(x_values, dist.probabilities_at(x_values))
        self.assertAlmostEqual(area, 1.0, 3)

    def test_singleDensityMedian(self):
        """ The median of a single measurement should be at the value of
            that measurement
        """

        for i in range(10):
            measurements = [value.ValueUncertainty.create_random(-100, 100)]

            dist = distributions.Distribution(measurements=measurements)
            median = distributions.distributions_ops.percentile(dist)

            self.assertAlmostEqual(median, measurements[0].value, delta=0.1)

    def test_tukey_box(self):
        measurements = []
        while len(measurements) < 6:
            measurements.append(value.ValueUncertainty())

        dist = distributions.Distribution(measurements=measurements)
        unweighted = boxes.unweighted_tukey(dist)
        weighted = boxes.weighted_tukey(dist)

    def test_generalizedGetMedian(self):
        for i in range(10):
            measurements = [
                value.ValueUncertainty.create_random(-100.0, -5.0),
                value.ValueUncertainty.create_random(-500.0, -20.0),
                value.ValueUncertainty.create_random(-100.0, -50.0),
                value.ValueUncertainty.create_random(),
                value.ValueUncertainty.create_random(),
                value.ValueUncertainty.create_random()
            ]

            dist = distributions.Distribution(measurements=measurements)
            result = distributions.distributions_ops.percentile(dist)

    def test_overlap(self):
        d1 = distributions.Distribution(measurements=[value.ValueUncertainty()])
        d2 = distributions.Distribution(measurements=[value.ValueUncertainty()])
        self.assertAlmostEqual(distributions.distributions_ops.overlap(d1, d2), 1.0)

    def test_compareAgainstGaussian2(self):
        d1 = distributions.Distribution(measurements=[value.ValueUncertainty()])
        d2 = distributions.Distribution(
            measurements=[value.ValueUncertainty(uncertainty=0.5)]
        )
        self.assertLess(distributions.distributions_ops.overlap(d1, d2), 0.7)

    def test_compareAgainstGaussian3(self):
        d1 = distributions.Distribution(measurements=[value.ValueUncertainty()])
        d2 = distributions.Distribution(measurements=[
            value.ValueUncertainty(5.0),
            value.ValueUncertainty(8.0),
            value.ValueUncertainty(10.0, 2.0)
        ])

        self.assertGreaterEqual(distributions.distributions_ops.overlap(d1, d2), 0.0)
        self.assertLess(distributions.distributions_ops.overlap(d1, d2), 0.06)

    def test_fromValuesOnly(self):
        values = [11, 15, 3, 7, 2]
        dd = distributions.create_distribution(values)
        self.assertAlmostEqual(dd.minimum_value().value, min(*values))
        self.assertAlmostEqual(dd.maximum_value().value, max(*values))

    def test_getAdaptiveRange(self):
        dist = distributions.Distribution(measurements=[value.ValueUncertainty()])
        result = distributions.distributions_ops.adaptive_range(dist, 10.0)

    def test_getAdaptiveRangeMulti(self):
        measurements = [
            value.ValueUncertainty(),
            value.ValueUncertainty(2.0, 0.5) ]
        dist = distributions.Distribution(measurements=measurements)
        result = distributions.distributions_ops.adaptive_range(dist, 10.0)

if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(TestDensity)
    unittest.TextTestRunner(verbosity=2).run(suite)

max_stars_repo_name: sernst/RefinedStatistics

max_issues_repo_name: sernst/RefinedStatistics

max_forks_repo_name: sernst/RefinedStatistics


text:
# Created by Pearu Peterson, June 2003
from __future__ import division, print_function, absolute_import

import numpy as np
from numpy.testing import (assert_equal, assert_almost_equal, assert_array_equal,
        assert_array_almost_equal, assert_allclose)
from scipy._lib._numpy_compat import suppress_warnings
from pytest import raises as assert_raises

from numpy import array, diff, linspace, meshgrid, ones, pi, shape
from scipy.interpolate.fitpack import bisplrep, bisplev
from scipy.interpolate.fitpack2 import (UnivariateSpline,
        LSQUnivariateSpline, InterpolatedUnivariateSpline,
        LSQBivariateSpline, SmoothBivariateSpline, RectBivariateSpline,
        LSQSphereBivariateSpline, SmoothSphereBivariateSpline,
        RectSphereBivariateSpline)

class TestUnivariateSpline(object):
    def test_linear_constant(self):
        x = [1,2,3]
        y = [3,3,3]
        lut = UnivariateSpline(x,y,k=1)
        assert_array_almost_equal(lut.get_knots(),[1,3])
        assert_array_almost_equal(lut.get_coeffs(),[3,3])
        assert_almost_equal(lut.get_residual(),0.0)
        assert_array_almost_equal(lut([1,1.5,2]),[3,3,3])

    def test_preserve_shape(self):
        x = [1, 2, 3]
        y = [0, 2, 4]
        lut = UnivariateSpline(x, y, k=1)
        arg = 2
        assert_equal(shape(arg), shape(lut(arg)))
        assert_equal(shape(arg), shape(lut(arg, nu=1)))
        arg = [1.5, 2, 2.5]
        assert_equal(shape(arg), shape(lut(arg)))
        assert_equal(shape(arg), shape(lut(arg, nu=1)))

    def test_linear_1d(self):
        x = [1,2,3]
        y = [0,2,4]
        lut = UnivariateSpline(x,y,k=1)
        assert_array_almost_equal(lut.get_knots(),[1,3])
        assert_array_almost_equal(lut.get_coeffs(),[0,4])
        assert_almost_equal(lut.get_residual(),0.0)
        assert_array_almost_equal(lut([1,1.5,2]),[0,1,2])

    def test_subclassing(self):
        # See #731

        class ZeroSpline(UnivariateSpline):
            def __call__(self, x):
                return 0*array(x)

        sp = ZeroSpline([1,2,3,4,5], [3,2,3,2,3], k=2)
        assert_array_equal(sp([1.5, 2.5]), [0., 0.])

    def test_empty_input(self):
        # Test whether empty input returns an empty output. Ticket 1014
        x = [1,3,5,7,9]
        y = [0,4,9,12,21]
        spl = UnivariateSpline(x, y, k=3)
        assert_array_equal(spl([]), array([]))

    def test_resize_regression(self):
        """Regression test for #1375."""
        x = [-1., -0.65016502, -0.58856235, -0.26903553, -0.17370892,
             -0.10011001, 0., 0.10011001, 0.17370892, 0.26903553, 0.58856235,
             0.65016502, 1.]
        y = [1.,0.62928599, 0.5797223, 0.39965815, 0.36322694, 0.3508061,
             0.35214793, 0.3508061, 0.36322694, 0.39965815, 0.5797223,
             0.62928599, 1.]
        w = [1.00000000e+12, 6.88875973e+02, 4.89314737e+02, 4.26864807e+02,
             6.07746770e+02, 4.51341444e+02, 3.17480210e+02, 4.51341444e+02,
             6.07746770e+02, 4.26864807e+02, 4.89314737e+02, 6.88875973e+02,
             1.00000000e+12]
        spl = UnivariateSpline(x=x, y=y, w=w, s=None)
        desired = array([0.35100374, 0.51715855, 0.87789547, 0.98719344])
        assert_allclose(spl([0.1, 0.5, 0.9, 0.99]), desired, atol=5e-4)

    def test_out_of_range_regression(self):
        # Test different extrapolation modes. See ticket 3557
        x = np.arange(5, dtype=float)
        y = x**3

        xp = linspace(-8, 13, 100)
        xp_zeros = xp.copy()
        xp_zeros[np.logical_or(xp_zeros < 0., xp_zeros > 4.)] = 0
        xp_clip = xp.copy()
        xp_clip[xp_clip < x[0]] = x[0]
        xp_clip[xp_clip > x[-1]] = x[-1]

        for cls in [UnivariateSpline, InterpolatedUnivariateSpline]:
            spl = cls(x=x, y=y)
            for ext in [0, 'extrapolate']:
                assert_allclose(spl(xp, ext=ext), xp**3, atol=1e-16)
                assert_allclose(cls(x, y, ext=ext)(xp), xp**3, atol=1e-16)
            for ext in [1, 'zeros']:
                assert_allclose(spl(xp, ext=ext), xp_zeros**3, atol=1e-16)
                assert_allclose(cls(x, y, ext=ext)(xp), xp_zeros**3, atol=1e-16)
            for ext in [2, 'raise']:
                assert_raises(ValueError, spl, xp, **dict(ext=ext))
            for ext in [3, 'const']:
                assert_allclose(spl(xp, ext=ext), xp_clip**3, atol=1e-16)
                assert_allclose(cls(x, y, ext=ext)(xp), xp_clip**3, atol=1e-16)

        # also test LSQUnivariateSpline [which needs explicit knots]
        t = spl.get_knots()[3:4]  # interior knots w/ default k=3
        spl = LSQUnivariateSpline(x, y, t)
        assert_allclose(spl(xp, ext=0), xp**3, atol=1e-16)
        assert_allclose(spl(xp, ext=1), xp_zeros**3, atol=1e-16)
        assert_raises(ValueError, spl, xp, **dict(ext=2))
        assert_allclose(spl(xp, ext=3), xp_clip**3, atol=1e-16)

        # also make sure that unknown values for `ext` are caught early
        for ext in [-1, 'unknown']:
            spl = UnivariateSpline(x, y)
            assert_raises(ValueError, spl, xp, **dict(ext=ext))
            assert_raises(ValueError, UnivariateSpline,
                    **dict(x=x, y=y, ext=ext))

    def test_lsq_fpchec(self):
        xs = np.arange(100) * 1.
        ys = np.arange(100) * 1.
        knots = np.linspace(0, 99, 10)
        bbox = (-1, 101)
        assert_raises(ValueError, LSQUnivariateSpline, xs, ys, knots,
                      bbox=bbox)

    def test_derivative_and_antiderivative(self):
        # Thin wrappers to splder/splantider, so light smoke test only.
        x = np.linspace(0, 1, 70)**3
        y = np.cos(x)

        spl = UnivariateSpline(x, y, s=0)
        spl2 = spl.antiderivative(2).derivative(2)
        assert_allclose(spl(0.3), spl2(0.3))

        spl2 = spl.antiderivative(1)
        assert_allclose(spl2(0.6) - spl2(0.2),
                        spl.integral(0.2, 0.6))

    def test_derivative_extrapolation(self):
        # Regression test for gh-10195: for a const-extrapolation spline
        # its derivative evaluates to zero for extrapolation
        x_values = [1, 2, 4, 6, 8.5]
        y_values = [0.5, 0.8, 1.3, 2.5, 5]
        f = UnivariateSpline(x_values, y_values, ext='const', k=3)

        x = [-1, 0, -0.5, 9, 9.5, 10]
        assert_allclose(f.derivative()(x), 0, atol=1e-15)

    def test_nan(self):
        # bail out early if the input data contains nans
        x = np.arange(10, dtype=float)
        y = x**3
        w = np.ones_like(x)
        # also test LSQUnivariateSpline [which needs explicit knots]
        spl = UnivariateSpline(x, y, check_finite=True)
        t = spl.get_knots()[3:4]  # interior knots w/ default k=3
        y_end = y[-1]
        for z in [np.nan, np.inf, -np.inf]:
            y[-1] = z
            assert_raises(ValueError, UnivariateSpline,
                    **dict(x=x, y=y, check_finite=True))
            assert_raises(ValueError, InterpolatedUnivariateSpline,
                    **dict(x=x, y=y, check_finite=True))
            assert_raises(ValueError, LSQUnivariateSpline,
                    **dict(x=x, y=y, t=t, check_finite=True))
            y[-1] = y_end  # check valid y but invalid w
            w[-1] = z
            assert_raises(ValueError, UnivariateSpline,
                    **dict(x=x, y=y, w=w, check_finite=True))
            assert_raises(ValueError, InterpolatedUnivariateSpline,
                    **dict(x=x, y=y, w=w, check_finite=True))
            assert_raises(ValueError, LSQUnivariateSpline,
                    **dict(x=x, y=y, t=t, w=w, check_finite=True))

    def test_strictly_increasing_x(self):
        # Test the x is required to be strictly increasing for
        # UnivariateSpline if s=0 and for InterpolatedUnivariateSpline,
        # but merely increasing for UnivariateSpline if s>0
        # and for LSQUnivariateSpline; see gh-8535
        xx = np.arange(10, dtype=float)
        yy = xx**3
        x = np.arange(10, dtype=float)
        x[1] = x[0]
        y = x**3
        w = np.ones_like(x)
        # also test LSQUnivariateSpline [which needs explicit knots]
        spl = UnivariateSpline(xx, yy, check_finite=True)
        t = spl.get_knots()[3:4]  # interior knots w/ default k=3
        spl2 = UnivariateSpline(x=x, y=y, w=w, s=1, check_finite=True)
        spl3 = LSQUnivariateSpline(x=x, y=y, t=t, w=w, check_finite=True)
        assert_raises(ValueError, UnivariateSpline,
                **dict(x=x, y=y, s=0, check_finite=True))
        assert_raises(ValueError, InterpolatedUnivariateSpline,
                **dict(x=x, y=y, check_finite=True))

    def test_increasing_x(self):
        # Test that x is required to be increasing, see gh-8535
        xx = np.arange(10, dtype=float)
        yy = xx**3
        x = np.arange(10, dtype=float)
        x[1] = x[0] - 1.0
        y = x**3
        w = np.ones_like(x)
        # also test LSQUnivariateSpline [which needs explicit knots]
        spl = UnivariateSpline(xx, yy, check_finite=True)
        t = spl.get_knots()[3:4]  # interior knots w/ default k=3
        assert_raises(ValueError, UnivariateSpline,
                **dict(x=x, y=y, check_finite=True))
        assert_raises(ValueError, InterpolatedUnivariateSpline,
                **dict(x=x, y=y, check_finite=True))
        assert_raises(ValueError, LSQUnivariateSpline,
                **dict(x=x, y=y, t=t, w=w, check_finite=True))

class TestLSQBivariateSpline(object):
    # NOTE: The systems in this test class are rank-deficient
    def test_linear_constant(self):
        x = [1,1,1,2,2,2,3,3,3]
        y = [1,2,3,1,2,3,1,2,3]
        z = [3,3,3,3,3,3,3,3,3]
        s = 0.1
        tx = [1+s,3-s]
        ty = [1+s,3-s]
        with suppress_warnings() as sup:
            r = sup.record(UserWarning, "\nThe coefficients of the spline")
            lut = LSQBivariateSpline(x,y,z,tx,ty,kx=1,ky=1)
            assert_equal(len(r), 1)

        assert_almost_equal(lut(2,2), 3.)

    def test_bilinearity(self):
        x = [1,1,1,2,2,2,3,3,3]
        y = [1,2,3,1,2,3,1,2,3]
        z = [0,7,8,3,4,7,1,3,4]
        s = 0.1
        tx = [1+s,3-s]
        ty = [1+s,3-s]
        with suppress_warnings() as sup:
            # This seems to fail (ier=1, see ticket 1642).
            sup.filter(UserWarning, "\nThe coefficients of the spline")
            lut = LSQBivariateSpline(x,y,z,tx,ty,kx=1,ky=1)

        tx, ty = lut.get_knots()
        for xa, xb in zip(tx[:-1], tx[1:]):
            for ya, yb in zip(ty[:-1], ty[1:]):
                for t in [0.1, 0.5, 0.9]:
                    for s in [0.3, 0.4, 0.7]:
                        xp = xa*(1-t) + xb*t
                        yp = ya*(1-s) + yb*s
                        zp = (+ lut(xa, ya)*(1-t)*(1-s)
                              + lut(xb, ya)*t*(1-s)
                              + lut(xa, yb)*(1-t)*s
                              + lut(xb, yb)*t*s)
                        assert_almost_equal(lut(xp,yp), zp)

    def test_integral(self):
        x = [1,1,1,2,2,2,8,8,8]
        y = [1,2,3,1,2,3,1,2,3]
        z = array([0,7,8,3,4,7,1,3,4])

        s = 0.1
        tx = [1+s,3-s]
        ty = [1+s,3-s]
        with suppress_warnings() as sup:
            r = sup.record(UserWarning, "\nThe coefficients of the spline")
            lut = LSQBivariateSpline(x, y, z, tx, ty, kx=1, ky=1)
            assert_equal(len(r), 1)
        tx, ty = lut.get_knots()
        tz = lut(tx, ty)
        trpz = .25*(diff(tx)[:,None]*diff(ty)[None,:]
                    * (tz[:-1,:-1]+tz[1:,:-1]+tz[:-1,1:]+tz[1:,1:])).sum()

        assert_almost_equal(lut.integral(tx[0], tx[-1], ty[0], ty[-1]),
                            trpz)

    def test_empty_input(self):
        # Test whether empty inputs returns an empty output. Ticket 1014
        x = [1,1,1,2,2,2,3,3,3]
        y = [1,2,3,1,2,3,1,2,3]
        z = [3,3,3,3,3,3,3,3,3]
        s = 0.1
        tx = [1+s,3-s]
        ty = [1+s,3-s]
        with suppress_warnings() as sup:
            r = sup.record(UserWarning, "\nThe coefficients of the spline")
            lut = LSQBivariateSpline(x, y, z, tx, ty, kx=1, ky=1)
            assert_equal(len(r), 1)

        assert_array_equal(lut([], []), np.zeros((0,0)))
        assert_array_equal(lut([], [], grid=False), np.zeros((0,)))

class TestSmoothBivariateSpline(object):
    def test_linear_constant(self):
        x = [1,1,1,2,2,2,3,3,3]
        y = [1,2,3,1,2,3,1,2,3]
        z = [3,3,3,3,3,3,3,3,3]
        lut = SmoothBivariateSpline(x,y,z,kx=1,ky=1)
        assert_array_almost_equal(lut.get_knots(),([1,1,3,3],[1,1,3,3]))
        assert_array_almost_equal(lut.get_coeffs(),[3,3,3,3])
        assert_almost_equal(lut.get_residual(),0.0)
        assert_array_almost_equal(lut([1,1.5,2],[1,1.5]),[[3,3],[3,3],[3,3]])

    def test_linear_1d(self):
        x = [1,1,1,2,2,2,3,3,3]
        y = [1,2,3,1,2,3,1,2,3]
        z = [0,0,0,2,2,2,4,4,4]
        lut = SmoothBivariateSpline(x,y,z,kx=1,ky=1)
        assert_array_almost_equal(lut.get_knots(),([1,1,3,3],[1,1,3,3]))
        assert_array_almost_equal(lut.get_coeffs(),[0,0,4,4])
        assert_almost_equal(lut.get_residual(),0.0)
        assert_array_almost_equal(lut([1,1.5,2],[1,1.5]),[[0,0],[1,1],[2,2]])

    def test_integral(self):
        x = [1,1,1,2,2,2,4,4,4]
        y = [1,2,3,1,2,3,1,2,3]
        z = array([0,7,8,3,4,7,1,3,4])

        with suppress_warnings() as sup:
            # This seems to fail (ier=1, see ticket 1642).
            sup.filter(UserWarning, "\nThe required storage space")
            lut = SmoothBivariateSpline(x, y, z, kx=1, ky=1, s=0)

        tx = [1,2,4]
        ty = [1,2,3]

        tz = lut(tx, ty)
        trpz = .25*(diff(tx)[:,None]*diff(ty)[None,:]
                    * (tz[:-1,:-1]+tz[1:,:-1]+tz[:-1,1:]+tz[1:,1:])).sum()
        assert_almost_equal(lut.integral(tx[0], tx[-1], ty[0], ty[-1]), trpz)

        lut2 = SmoothBivariateSpline(x, y, z, kx=2, ky=2, s=0)
        assert_almost_equal(lut2.integral(tx[0], tx[-1], ty[0], ty[-1]), trpz,
                            decimal=0)  # the quadratures give 23.75 and 23.85

        tz = lut(tx[:-1], ty[:-1])
        trpz = .25*(diff(tx[:-1])[:,None]*diff(ty[:-1])[None,:]
                    * (tz[:-1,:-1]+tz[1:,:-1]+tz[:-1,1:]+tz[1:,1:])).sum()
        assert_almost_equal(lut.integral(tx[0], tx[-2], ty[0], ty[-2]), trpz)

    def test_rerun_lwrk2_too_small(self):
        # in this setting, lwrk2 is too small in the default run. Here we
        # check for equality with the bisplrep/bisplev output because there,
        # an automatic re-run of the spline representation is done if ier>10.
        x = np.linspace(-2, 2, 80)
        y = np.linspace(-2, 2, 80)
        z = x + y
        xi = np.linspace(-1, 1, 100)
        yi = np.linspace(-2, 2, 100)
        tck = bisplrep(x, y, z)
        res1 = bisplev(xi, yi, tck)
        interp_ = SmoothBivariateSpline(x, y, z)
        res2 = interp_(xi, yi)
        assert_almost_equal(res1, res2)

class TestLSQSphereBivariateSpline(object):
    def setup_method(self):
        # define the input data and coordinates
        ntheta, nphi = 70, 90
        theta = linspace(0.5/(ntheta - 1), 1 - 0.5/(ntheta - 1), ntheta) * pi
        phi = linspace(0.5/(nphi - 1), 1 - 0.5/(nphi - 1), nphi) * 2. * pi
        data = ones((theta.shape[0], phi.shape[0]))
        # define knots and extract data values at the knots
        knotst = theta[::5]
        knotsp = phi[::5]
        knotdata = data[::5, ::5]
        # calculate spline coefficients
        lats, lons = meshgrid(theta, phi)
        lut_lsq = LSQSphereBivariateSpline(lats.ravel(), lons.ravel(),
                                           data.T.ravel(), knotst, knotsp)
        self.lut_lsq = lut_lsq
        self.data = knotdata
        self.new_lons, self.new_lats = knotsp, knotst

    def test_linear_constant(self):
        assert_almost_equal(self.lut_lsq.get_residual(), 0.0)
        assert_array_almost_equal(self.lut_lsq(self.new_lats, self.new_lons),
                                  self.data)

    def test_empty_input(self):
        assert_array_almost_equal(self.lut_lsq([], []), np.zeros((0,0)))
        assert_array_almost_equal(self.lut_lsq([], [], grid=False), np.zeros((0,)))

class TestSmoothSphereBivariateSpline(object):
    def setup_method(self):
        theta = array([.25*pi, .25*pi, .25*pi, .5*pi, .5*pi, .5*pi, .75*pi,
                       .75*pi, .75*pi])
        phi = array([.5 * pi, pi, 1.5 * pi, .5 * pi, pi, 1.5 * pi, .5 * pi, pi,
                     1.5 * pi])
        r = array([3, 3, 3, 3, 3, 3, 3, 3, 3])
        self.lut = SmoothSphereBivariateSpline(theta, phi, r, s=1E10)

    def test_linear_constant(self):
        assert_almost_equal(self.lut.get_residual(), 0.)
        assert_array_almost_equal(self.lut([1, 1.5, 2],[1, 1.5]),
                                  [[3, 3], [3, 3], [3, 3]])

    def test_empty_input(self):
        assert_array_almost_equal(self.lut([], []), np.zeros((0,0)))
        assert_array_almost_equal(self.lut([], [], grid=False), np.zeros((0,)))

class TestRectBivariateSpline(object):
    def test_defaults(self):
        x = array([1,2,3,4,5])
        y = array([1,2,3,4,5])
        z = array([[1,2,1,2,1],[1,2,1,2,1],[1,2,3,2,1],[1,2,2,2,1],[1,2,1,2,1]])
        lut = RectBivariateSpline(x,y,z)
        assert_array_almost_equal(lut(x,y),z)

    def test_evaluate(self):
        x = array([1,2,3,4,5])
        y = array([1,2,3,4,5])
        z = array([[1,2,1,2,1],[1,2,1,2,1],[1,2,3,2,1],[1,2,2,2,1],[1,2,1,2,1]])
        lut = RectBivariateSpline(x,y,z)

        xi = [1, 2.3, 5.3, 0.5, 3.3, 1.2, 3]
        yi = [1, 3.3, 1.2, 4.0, 5.0, 1.0, 3]
        zi = lut.ev(xi, yi)
        zi2 = array([lut(xp, yp)[0,0] for xp, yp in zip(xi, yi)])

        assert_almost_equal(zi, zi2)

    def test_derivatives_grid(self):
        x = array([1,2,3,4,5])
        y = array([1,2,3,4,5])
        z = array([[1,2,1,2,1],[1,2,1,2,1],[1,2,3,2,1],[1,2,2,2,1],[1,2,1,2,1]])
        dx = array([[0,0,-20,0,0],[0,0,13,0,0],[0,0,4,0,0],
            [0,0,-11,0,0],[0,0,4,0,0]])/6.
        dy = array([[4,-1,0,1,-4],[4,-1,0,1,-4],[0,1.5,0,-1.5,0],
            [2,.25,0,-.25,-2],[4,-1,0,1,-4]])
        dxdy = array([[40,-25,0,25,-40],[-26,16.25,0,-16.25,26],
            [-8,5,0,-5,8],[22,-13.75,0,13.75,-22],[-8,5,0,-5,8]])/6.
        lut = RectBivariateSpline(x,y,z)
        assert_array_almost_equal(lut(x,y,dx=1),dx)
        assert_array_almost_equal(lut(x,y,dy=1),dy)
        assert_array_almost_equal(lut(x,y,dx=1,dy=1),dxdy)

    def test_derivatives(self):
        x = array([1,2,3,4,5])
        y = array([1,2,3,4,5])
        z = array([[1,2,1,2,1],[1,2,1,2,1],[1,2,3,2,1],[1,2,2,2,1],[1,2,1,2,1]])
        dx = array([0,0,2./3,0,0])
        dy = array([4,-1,0,-.25,-4])
        dxdy = array([160,65,0,55,32])/24.
        lut = RectBivariateSpline(x,y,z)
        assert_array_almost_equal(lut(x,y,dx=1,grid=False),dx)
        assert_array_almost_equal(lut(x,y,dy=1,grid=False),dy)
        assert_array_almost_equal(lut(x,y,dx=1,dy=1,grid=False),dxdy)

    def test_broadcast(self):
        x = array([1,2,3,4,5])
        y = array([1,2,3,4,5])
        z = array([[1,2,1,2,1],[1,2,1,2,1],[1,2,3,2,1],[1,2,2,2,1],[1,2,1,2,1]])
        lut = RectBivariateSpline(x,y,z)
        assert_allclose(lut(x, y), lut(x[:,None], y[None,:], grid=False))

class TestRectSphereBivariateSpline(object):
    def test_defaults(self):
        y = linspace(0.01, 2*pi-0.01, 7)
        x = linspace(0.01, pi-0.01, 7)
        z = array([[1,2,1,2,1,2,1],[1,2,1,2,1,2,1],[1,2,3,2,1,2,1],
                   [1,2,2,2,1,2,1],[1,2,1,2,1,2,1],[1,2,2,2,1,2,1],
                   [1,2,1,2,1,2,1]])
        lut = RectSphereBivariateSpline(x,y,z)
        assert_array_almost_equal(lut(x,y),z)

    def test_evaluate(self):
        y = linspace(0.01, 2*pi-0.01, 7)
        x = linspace(0.01, pi-0.01, 7)
        z = array([[1,2,1,2,1,2,1],[1,2,1,2,1,2,1],[1,2,3,2,1,2,1],
                   [1,2,2,2,1,2,1],[1,2,1,2,1,2,1],[1,2,2,2,1,2,1],
                   [1,2,1,2,1,2,1]])
        lut = RectSphereBivariateSpline(x,y,z)
        yi = [0.2, 1, 2.3, 2.35, 3.0, 3.99, 5.25]
        xi = [1.5, 0.4, 1.1, 0.45, 0.2345, 1., 0.0001]
        zi = lut.ev(xi, yi)
        zi2 = array([lut(xp, yp)[0,0] for xp, yp in zip(xi, yi)])
        assert_almost_equal(zi, zi2)

    def test_derivatives_grid(self):
        y = linspace(0.01, 2*pi-0.01, 7)
        x = linspace(0.01, pi-0.01, 7)
        z = array([[1,2,1,2,1,2,1],[1,2,1,2,1,2,1],[1,2,3,2,1,2,1],
                   [1,2,2,2,1,2,1],[1,2,1,2,1,2,1],[1,2,2,2,1,2,1],
                   [1,2,1,2,1,2,1]])

        lut = RectSphereBivariateSpline(x,y,z)

        y = linspace(0.02, 2*pi-0.02, 7)
        x = linspace(0.02, pi-0.02, 7)

        assert_allclose(lut(x, y, dtheta=1), _numdiff_2d(lut, x, y, dx=1),
                        rtol=1e-4, atol=1e-4)
        assert_allclose(lut(x, y, dphi=1), _numdiff_2d(lut, x, y, dy=1),
                        rtol=1e-4, atol=1e-4)
        assert_allclose(lut(x, y, dtheta=1, dphi=1), _numdiff_2d(lut, x, y, dx=1, dy=1, eps=1e-6),
                        rtol=1e-3, atol=1e-3)

    def test_derivatives(self):
        y = linspace(0.01, 2*pi-0.01, 7)
        x = linspace(0.01, pi-0.01, 7)
        z = array([[1,2,1,2,1,2,1],[1,2,1,2,1,2,1],[1,2,3,2,1,2,1],
                   [1,2,2,2,1,2,1],[1,2,1,2,1,2,1],[1,2,2,2,1,2,1],
                   [1,2,1,2,1,2,1]])

        lut = RectSphereBivariateSpline(x,y,z)

        y = linspace(0.02, 2*pi-0.02, 7)
        x = linspace(0.02, pi-0.02, 7)

        assert_equal(lut(x, y, dtheta=1, grid=False).shape, x.shape)
        assert_allclose(lut(x, y, dtheta=1, grid=False),
                        _numdiff_2d(lambda x,y: lut(x,y,grid=False), x, y, dx=1),
                        rtol=1e-4, atol=1e-4)
        assert_allclose(lut(x, y, dphi=1, grid=False),
                        _numdiff_2d(lambda x,y: lut(x,y,grid=False), x, y, dy=1),
                        rtol=1e-4, atol=1e-4)
        assert_allclose(lut(x, y, dtheta=1, dphi=1, grid=False),
                        _numdiff_2d(lambda x,y: lut(x,y,grid=False), x, y, dx=1, dy=1, eps=1e-6),
                        rtol=1e-3, atol=1e-3)

def _numdiff_2d(func, x, y, dx=0, dy=0, eps=1e-8):
    if dx == 0 and dy == 0:
        return func(x, y)
    elif dx == 1 and dy == 0:
        return (func(x + eps, y) - func(x - eps, y)) / (2*eps)
    elif dx == 0 and dy == 1:
        return (func(x, y + eps) - func(x, y - eps)) / (2*eps)
    elif dx == 1 and dy == 1:
        return (func(x + eps, y + eps) - func(x - eps, y + eps)
                - func(x + eps, y - eps) + func(x - eps, y - eps)) / (2*eps)**2
    else:
        raise ValueError("invalid derivative order")

ext: py

reason: import numpy,from numpy,from scipy


text:
from copy import deepcopy
from functools import partial
from slm_lab.agent import net, memory
from slm_lab.agent.algorithm import math_util, policy_util
from slm_lab.agent.algorithm.actor_critic import ActorCritic
from slm_lab.agent.net import net_util
from slm_lab.lib import logger, util
from slm_lab.lib.decorator import lab_api
import numpy as np
import torch
import pydash as ps

logger = logger.get_logger(__name__)

class SIL(ActorCritic):
    '''
    Implementation of Self-Imitation Learning (SIL) https://arxiv.org/abs/1806.05635
    This is actually just A2C with an extra SIL loss function

    e.g. algorithm_spec
    "algorithm": {
        "name": "SIL",
        "action_pdtype": "default",
        "action_policy": "default",
        "action_policy_update": "no_update",
        "explore_var_start": null,
        "explore_var_end": null,
        "explore_anneal_epi": null,
        "gamma": 0.99,
        "use_gae": true,
        "lam": 1.0,
        "use_nstep": false,
        "num_step_returns": 100,
        "add_entropy": true,
        "entropy_coef": 0.01,
        "policy_loss_coef": 1.0,
        "val_loss_coef": 0.01,
        "sil_policy_loss_coef": 1.0,
        "sil_val_loss_coef": 0.01,
        "continuous_action_clip": 2.0,
        "training_batch_epoch": 8,
        "training_frequency": 1,
        "training_epoch": 8
    }
    e.g. special memory_spec
    "memory": {
        "name": "OnPolicyReplay",
        "sil_replay_name": "Replay",
        "batch_size": 32,
        "max_size": 10000,
        "use_cer": true
    }
    '''
    @lab_api
    def post_body_init(self):
        '''Initializes the part of algorithm needing a body to exist first.'''
        self.body = self.agent.nanflat_body_a[0]  # single-body algo
        # create the extra replay memory for SIL
        memory_name = self.memory_spec['sil_replay_name']
        MemoryClass = getattr(memory, memory_name)
        self.body.replay_memory = MemoryClass(self.memory_spec, self, self.body)
        self.init_algorithm_params()
        self.init_nets()
        logger.info(util.self_desc(self))

    @lab_api
    def init_algorithm_params(self):
        '''Initialize other algorithm parameters'''
        # set default
        util.set_attr(self, dict(
            action_pdtype='default',
            action_policy='default',
            action_policy_update='no_update',
            explore_var_start=np.nan,
            explore_var_end=np.nan,
            explore_anneal_epi=np.nan,
        ))
        util.set_attr(self, self.algorithm_spec, [
            'action_pdtype',
            'action_policy',
            # theoretically, AC does not have policy update; but in this implementation we have such option
            'action_policy_update',
            'explore_var_start',
            'explore_var_end',
            'explore_anneal_epi',
            'gamma',  # the discount factor
            'use_gae',
            'lam',
            'use_nstep',
            'num_step_returns',
            'add_entropy',
            'entropy_coef',
            'policy_loss_coef',
            'val_loss_coef',
            'sil_policy_loss_coef',
            'sil_val_loss_coef',
            'continuous_action_clip',
            'training_frequency',
            'training_batch_epoch',
            'training_epoch',
        ])
        self.to_train = 0
        self.action_policy = getattr(policy_util, self.action_policy)
        self.action_policy_update = getattr(policy_util, self.action_policy_update)
        for body in self.agent.nanflat_body_a:
            body.explore_var = self.explore_var_start
        self.calc_advs_v_targets = self.calc_gae_advs_v_targets

    def sample(self):
        '''Modify the onpolicy sample to also append to replay'''
        batches = [body.memory.sample() for body in self.agent.nanflat_body_a]
        batch = util.concat_batches(batches)
        data_keys = self.body.replay_memory.data_keys
        for idx in range(len(batch['dones'])):
            tuples = [batch[k][idx] for k in data_keys]
            self.body.replay_memory.add_experience(*tuples)
        batch = util.to_torch_batch(batch, self.net.gpu)
        return batch

    def replay_sample(self):
        '''Samples a batch from memory'''
        batches = [body.replay_memory.sample() for body in self.agent.nanflat_body_a]
        batch = util.concat_batches(batches)
        batch = util.to_torch_batch(batch, self.net.gpu)
        assert not torch.isnan(batch['states']).any()
        return batch

    def calc_log_probs(self, batch):
        '''Helper method to calculate log_probs for a randomly sampled batch'''
        states, actions = batch['states'], batch['actions']
        # get ActionPD, don't append to state_buffer
        ActionPD, _pdparam, _body = policy_util.init_action_pd(states[0].cpu().numpy(), self, self.body, append=False)
        # construct log_probs for each state-action
        pdparams = self.calc_pdparam(states, evaluate=False)
        log_probs = []
        for idx, pdparam in enumerate(pdparams):
            _action, action_pd = policy_util.sample_action_pd(ActionPD, pdparam, self.body)
            log_prob = action_pd.log_prob(actions[idx])
            log_probs.append(log_prob)
        log_probs = torch.stack(log_probs)
        return log_probs

    def calc_sil_policy_val_loss(self, batch):
        '''
        Calculate the SIL policy losses for actor and critic
        sil_policy_loss = -log_prob * max(R - v_pred, 0)
        sil_val_loss = (max(R - v_pred, 0)^2) / 2
        This is called on a randomly-sample batch from experience replay
        '''
        returns = math_util.calc_returns(batch, self.gamma)
        v_preds = self.calc_v(batch['states'], evaluate=False)
        clipped_advs = torch.clamp(returns - v_preds, min=0.0)
        log_probs = self.calc_log_probs(batch)

        sil_policy_loss = self.sil_policy_loss_coef * torch.mean(- log_probs * clipped_advs)
        sil_val_loss = self.sil_val_loss_coef * torch.pow(clipped_advs, 2) / 2
        sil_val_loss = torch.mean(sil_val_loss)

        if torch.cuda.is_available() and self.net.gpu:
            sil_policy_loss = sil_policy_loss.cuda()
            sil_val_loss = sil_val_loss.cuda()
        logger.debug(f'SIL actor policy loss: {sil_policy_loss:.2f}')
        logger.debug(f'SIL critic value loss: {sil_val_loss:.2f}')
        return sil_policy_loss, sil_val_loss

    def train_shared(self):
        '''
        Trains the network when the actor and critic share parameters
        '''
        if self.to_train == 1:
            # onpolicy a2c update
            a2c_loss = super(SIL, self).train_shared()
            # offpolicy sil update with random minibatch
            total_sil_loss = torch.tensor(0.0)
            for _ in range(self.training_epoch):
                batch = self.replay_sample()
                sil_policy_loss, sil_val_loss = self.calc_sil_policy_val_loss(batch)
                sil_loss = sil_policy_loss + sil_val_loss
                self.net.training_step(loss=sil_loss)
                total_sil_loss += sil_loss.cpu()
            sil_loss = total_sil_loss / self.training_epoch
            loss = a2c_loss + sil_loss
            self.last_loss = loss.item()
        return self.last_loss

    def train_separate(self):
        '''
        Trains the network when the actor and critic are separate networks
        '''
        if self.to_train == 1:
            # onpolicy a2c update
            a2c_loss = super(SIL, self).train_separate()
            # offpolicy sil update with random minibatch
            total_sil_loss = torch.tensor(0.0)
            for _ in range(self.training_epoch):
                batch = self.replay_sample()
                sil_policy_loss, sil_val_loss = self.calc_sil_policy_val_loss(batch)
                self.net.training_step(loss=sil_policy_loss, retain_graph=True)
                self.critic.training_step(loss=sil_val_loss)
                total_sil_loss += sil_policy_loss + sil_val_loss
            sil_loss = total_sil_loss / self.training_epoch
            loss = a2c_loss + sil_loss
            self.last_loss = loss.item()
        return self.last_loss

max_stars_repo_name: cclauss/SLM-Lab

max_issues_repo_name: cclauss/SLM-Lab

max_forks_repo_name: cclauss/SLM-Lab

reason: import numpy


text:
# coding: utf-8
"""
使用NLI训练得到的whitening参数, 在下游任务测试.

@env: python3, pytorch>=1.7.1, transformers==4.2.0
@author: Weijie Liu
@date: 20/01/2020
"""
import os
import torch
import numpy as np
from tqdm import tqdm
import scipy.stats
from all_utils import *
import senteval
import logging
from prettytable import PrettyTable

MAX_LENGTH = 64
BATCH_SIZE = 256
TEST_PATH = './data/'
logging.basicConfig(format='%(asctime)s : %(message)s', level=logging.DEBUG)

MODEL_ZOOS = {

    'BERTbase-first_last_avg': {
        'encoder': './model/bert-base-uncased',
        'pooling': 'first_last_avg'
    },

    'BERTbase-cls': {
        'encoder': './model/bert-base-uncased',
        'pooling': 'cls'
    },

    'BERTlarge-first_last_avg': {
        'encoder': './model/bert-large-uncased',
        'pooling': 'first_last_avg'
    },

    'BERTlarge-cls': {
        'encoder': './model/bert-large-uncased',
        'pooling': 'cls'
    },

    'SBERTbase-nli-first_last_avg': {
        'encoder': './model/bert-base-nli-mean-tokens',
        'pooling': 'first_last_avg'
    },

    'SBERTbase-nli-first_last_avg': {
        'encoder': './model/bert-base-nli-mean-tokens',
        'pooling': 'cls'
    },

    'SBERTlarge-nli-first_last_avg': {
        'encoder': './model/bert-large-nli-mean-tokens',
        'pooling': 'first_last_avg'
    },

    'SBERTlarge-nli-first_last_avg': {
        'encoder': './model/bert-large-nli-mean-tokens',
        'pooling': 'cls'
    },
}

def prepare(params, samples):
    return None

def batcher(params, batch):
    batch = [' '.join(sent) if sent != [] else '.' for sent in batch]
    embeddings = []
    for sent in batch:
        vec = sent_to_vec(sent, params['tokenizer'], \
                params['encoder'], params['pooling'], MAX_LENGTH)
        embeddings.append(vec)
    embeddings = np.vstack(embeddings)
    return embeddings

def run(model_name, test_path):

    model_config = MODEL_ZOOS[model_name]
    logging.info(f"{model_name} configs: {model_config}")

    tokenizer, encoder = build_model(model_config['encoder'])
    logging.info("Building {} tokenizer and model successfuly.".format(model_config['encoder']))

    # Set params for senteval
    params_senteval = {
            'task_path': test_path,
            'usepytorch': True,
            'tokenizer': tokenizer,
            'encoder': encoder,
            'pooling': model_config['pooling'],
            'batch_size': BATCH_SIZE
        }

    se = senteval.engine.SE(params_senteval, batcher, prepare)
    transfer_tasks = [
            'STS12', 'STS13', 'STS14', 'STS15', 'STS16',
            'SICKRelatednessCosin',
            'STSBenchmarkCosin'
        ]
    results = se.eval(transfer_tasks)

    # Show results
    table = PrettyTable(["Task", "Spearman"])
    for task in transfer_tasks:
        if task in ['STS12', 'STS13', 'STS14', 'STS15', 'STS16']:
            metric = results[task]['all']['spearman']['wmean']
        elif task in ['SICKRelatednessCosin', 'STSBenchmarkCosin']:
            metric = results[task]['spearman']
        table.add_row([task, metric])
    logging.info(f"{model_name} results:\n" + str(table))

def run_all_model():

    for model_name in MODEL_ZOOS:
        run(model_name, TEST_PATH)

if __name__ == "__main__":
    # run('BERTbase-first_last_avg', TEST_PATH)
    run_all_model()

max_stars_repo_name: dumpmemory/BERT-whitening-pytorch

max_issues_repo_name: dumpmemory/BERT-whitening-pytorch

max_forks_repo_name: dumpmemory/BERT-whitening-pytorch

reason: import numpy,import scipy


text:
import tensorflow as tf
import numpy as np

def preprocessing(document, summary, bert=False):
    # for decoder sequence
    summary = summary.apply(lambda x: '<start> ' + x + ' <end>')

    #tokenizing
    filters = '!"#$%&()*+,-./:;=?@[\\]^_`{|}~\t\n'
    oov_token = '<unk>'
    if not bert:
        document_tokenizer = tf.keras.preprocessing.text.Tokenizer(oov_token=oov_token)
        document_tokenizer.fit_on_texts(document)
        inputs = document_tokenizer.texts_to_sequences(document)
        encoder_vocab_size = len(document_tokenizer.word_index) + 1
    else:
        inputs = None
        encoder_vocab_size = None
        document_tokenizer = None

    summary_tokenizer = tf.keras.preprocessing.text.Tokenizer(filters=filters, oov_token=oov_token)
    summary_tokenizer.fit_on_texts(summary)

    targets = summary_tokenizer.texts_to_sequences(summary)

    decoder_vocab_size = len(summary_tokenizer.word_index) + 1

    return inputs, targets, encoder_vocab_size, decoder_vocab_size, document_tokenizer, summary_tokenizer

def encode_sentence(s, tokenizer):
   tokens = list(tokenizer.tokenize(s))
   tokens.append('[SEP]')
   return tokenizer.convert_tokens_to_ids(tokens)

def bert_encode(glue_dict, tokenizer):

    article = tf.ragged.constant([
        encode_sentence(s, tokenizer)
        for s in np.array(glue_dict)])
    # highlights = tf.ragged.constant([
    #     encode_sentence(s, tokenizer)
    #     for s in np.array(glue_dict["highlights"])])

    cls = [tokenizer.convert_tokens_to_ids(['[CLS]'])]*article.shape[0]
    input_word_ids = tf.concat([cls, article], axis=-1)

    input_mask = tf.ones_like(input_word_ids).to_tensor()

    type_cls = tf.zeros_like(cls)
    type_s1 = tf.zeros_like(article)
    # type_s2 = tf.ones_like(highlights)
    input_type_ids = tf.concat(
        [type_cls, type_s1], axis=-1).to_tensor()

    inputs = {
        'input_word_ids': input_word_ids.to_tensor(),
        'input_mask': input_mask,
        'input_type_ids': input_type_ids}

    return inputs


text:
from __future__ import print_function
import matplotlib.pyplot as plt
import seaborn as sns
sns.set(font_scale=2)
import numpy as np
import scipy as sp
import sys

def dNdEPlot_pts(energies,numSpec,bin_left,bin_right,sst12,m12,PID=None):
    num_points = len(energies)
    opacity = 0.9
    fix, ax = plt.subplots()
    plt.plot(energies,numSpec,'ro', alpha=opacity, color='b')
    plt.hlines(numSpec,bin_left,bin_right, color = 'b')
    plt.vlines(bin_left,numSpec, \
            0.0000000001, color = 'b')
    plt.vlines(bin_right,numSpec, \
            0.0000000001, color = 'b')
    ax.annotate(r'$\sin^{2}(\theta _{12})$ =' + str(sst12) + '\n' + \
            r'$\Delta m^{2}_{21}$ = ' + str(m12), xy=(7,125),
            xytext=(6.5,125))
    plt.ylim(0,np.max(numSpec) + 1)
    plt.ylabel(r'Events/ 200 keV')
    if PID=='pos':
        plt.xlabel('Prompt Energy (MeV)')
        plt.title(r'Neutrino spectrum at given location in positron energy')
    if PID=='nu':
        plt.xlabel('Antineutrino Energy (MeV)')
        plt.title(r'Neutrino spectrum at given location in antineutrino energy')
    plt.show()

#Takes in a Histogram object as defined in /lib/histogram and plots it
def plot_EventHist(Histogram,sst12,m12):
    num_points = len(Histogram.bin_centers)
    opacity = 0.9
    fix, ax = plt.subplots()
    plt.plot(Histogram.bin_centers,Histogram.bin_values,'ro', \
            alpha=opacity, color='b')
    plt.hlines(Histogram.bin_values,Histogram.bin_lefts, \
            Histogram.bin_rights, color = 'b')
    plt.vlines(Histogram.bin_lefts,Histogram.bin_values, \
            0.0000000001, color = 'b')
    plt.vlines(Histogram.bin_rights,Histogram.bin_values, \
            0.0000000001, color = 'b')
    ax.annotate(r'$\sin^{2}(\theta _{12})$ =' + str(sst12) + '\n' + \
            r'$\Delta m^{2}_{21}$ = ' + str(m12), xy=(7,200),
            xytext=(6.5,200))
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'events/$10^{32}proton-years/MeV$')
    plt.title(r'Neutrino spectrum in TNU at input location')
    plt.show()

#Takes in a Histogram object as defined in /lib/histogram and plots it
def plot_TwoEventHist(Hist1,Hist2,sst12,m12):
    num_points = len(Hist1.bin_centers)
    opacity = 0.9
    fix, ax = plt.subplots()
    plt.plot(Hist1.bin_centers,Hist1.bin_values,'ro', \
            alpha=opacity, color='r')
    plt.hlines(Hist1.bin_values,Hist1.bin_lefts, \
            Hist1.bin_rights, color = 'r')
    plt.plot(Hist2.bin_centers,Hist2.bin_values,'bo', \
            alpha=opacity, color='b')
    plt.hlines(Hist2.bin_values,Hist2.bin_lefts, \
            Hist2.bin_rights, color = 'b')
    ax.annotate(r'$\sin^{2}(\theta _{12})$ =' + str(sst12) + '\n' + \
            r'$\Delta m^{2}_{21}$ = ' + str(m12), xy=(7,200),
            xytext=(6.5,200))
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'events/$10^{32}proton-years/MeV$')
    plt.title(r'Comparison of two spectrum histograms')
    plt.show()

def dNdEPlot_line(energies,numSpec,sst12,m12,PID=None):
    num_points = len(energies)
    opacity = 0.9
    fig, ax = plt.subplots()
    plt.plot(energies,numSpec, alpha=opacity, color='g')
    plt.fill_between(energies, 1e-10, numSpec, facecolor ='g',alpha = 0.4)
    ax.annotate(r'$\sin^{2}(\theta _{12})$ =' + str(sst12) + '\n' + \
            r'$\Delta m^{2}_{21}$ = ' + str(m12), xy=(7,35),
            xytext=(6.5,35))
#    plt.xaxis.get_label().set_fontproperties(30)
    if PID=='pos':
        plt.xlabel('Prompt Energy (MeV)')
        plt.ylabel(r'$dN/dE_{prompt}$ (MeV)')
        plt.title(r'Event Spectrum as a function of prompt positron energy')
    if PID=='nu':
        plt.xlabel('Energy (MeV)')
        plt.ylabel(r'$dN/dE_{\nu}$ (MeV)')
        plt.title(r'Neutrino event spectrum')
    else:
        print("PID not recognized.  Not making plot")
        return
    plt.show()

def dNdEPlot_line_TNU(energies,numSpec,sst12,m12,PID=None):
    num_points = len(energies)
    opacity = 0.9
    fix, ax = plt.subplots()
    ax.plot(energies,numSpec, alpha=opacity, color='g')
    ax.fill_between(energies, 1e-10, numSpec, facecolor ='g',alpha = 0.4)
    ax.annotate(r'$\sin^{2}(\theta _{12})$ =' + str(sst12) + '\n' + \
            r'$\Delta m^{2}_{21}$ = ' + str(m12), xy=(7,50),
            xytext=(6.5,50))
    for tick in ax.xaxis.get_major_ticks():
        tick.label.set_fontsize(18)
    for tick in ax.yaxis.get_major_ticks():
        tick.label.set_fontsize(18)
    if PID=='pos':
        plt.xlabel('Prompt Energy (MeV)')
        plt.ylabel(r'$dN/dE_{prompt}$ (TNU/MeV)')
        plt.title(r'Event Spectrum as a function of prompt positron energy')
    if PID=='nu':
        plt.xlabel('Energy (MeV)')
        plt.ylabel(r'$dN/dE_{\nu}$ (TNU/MeV)')
        plt.title(r'Neutrino event spectrum')
    else:
        print("PID not recognized.  Not making plot")
        return
    plt.show()

def CAspectrumPlot(energies,spectrum):
    '''
    Takes an array evaluated at energy points and plots them vs. energy.
    Specific to Canadian spectrum to have labels ready.
    '''
    num_points = len(energies)
    opacity = 0.9
    fig, ax = plt.subplots()
    #plt.gcf().subplots_adjust(bottom=0.2)
    plt.plot(energies, spectrum, alpha=opacity, color='b')
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'(Sum of Oscillated Spectra$m^{-2}$)')
    plt.title(r'Plot of oscillated neutrino spectrum at input location for all' + \
            'Canadian plants')
    #plt.xticks(index + bar_width, x, y=0.001)
    #plt.legend()
    #plt.tight_layout()  #could use instead of the subplots_adjust line
    plt.show()

def plotSumOscSpectrum(OscSpectra):
    """
    Plots the "core_number"s spectrum in the passed in UnoscSpectra class using
    Matplotlib.pyplot.
    """
    num_points = len(OscSpectra.Summed_Spectra)
    spectrum = OscSpectra.Summed_Spectra
    energies = OscSpectra.E_arr
    opacity = 0.9
    fig, ax = plt.subplots()
    #plt.gcf().subplots_adjust(bottom=0.2)
    plt.plot(energies, spectrum, alpha=opacity, color='g')
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'(Sum of Oscillated Spectra$m^{-2}$)')
    plt.title(r'Plot of oscillated Core Spectrums for Plant ' +  \
            str(OscSpectra.ReacDetails.index) + ' at input location')
    #plt.xticks(index + bar_width, x, y=0.001)
    #plt.legend()
    #plt.tight_layout()  #could use instead of the subplots_adjust line
    plt.show()

def plotCoreOscSpectrum(core_number,OscSpectra):
    """
    Plots the "core_number"s spectrum in the passed in UnoscSpectra class using
    Matplotlib.pyplot.
    """
    num_points = len(OscSpectra.Osc_Spectra[core_number])
    spectrum = OscSpectra.Osc_Spectra[core_number]
    energies = OscSpectra.E_arr
    opacity = 0.9
    fig, ax = plt.subplots()
    #plt.gcf().subplots_adjust(bottom=0.2)
    plt.plot(energies, spectrum, alpha=opacity, color='g')
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'Unoscillated spectra ($m^{-2}$)')
    plt.title(r'Plot of oscillated Core Spectrum for the ' + str(core_number) + \
    r'th core of Plant ' + str(OscSpectra.ReacDetails.index) + ' at input location')
    #plt.xticks(index + bar_width, x, y=0.001)
    #plt.legend()
    #plt.tight_layout()  #could use instead of the subplots_adjust line
    plt.show()

def plotCoreUnoscSpectrum(core_number,UnoscSpectra):
    """
    Plots the "core_number"s spectrum in the passed in UnoscSpectra class using
    Matplotlib.pyplot.
    """
    num_points = len(UnoscSpectra.Unosc_Spectra[core_number])
    spectrum = UnoscSpectra.Unosc_Spectra[core_number]
    energies = UnoscSpectra.E_arr
    opacity = 0.9
    fig, ax = plt.subplots()
    #plt.gcf().subplots_adjust(bottom=0.2)
    plt.plot(energies, spectrum, alpha=opacity, color='g')
    plt.xlabel('Energy (MeV)')
    plt.ylabel(r'Unoscillated spectra ($m^{-2}$)')
    plt.title(r'Plot of Unoscillated Core Spectrum for the ' + str(core_number) + \
    r'th core of Plant ' + str(UnoscSpectra.ReacDetails.index) + ' at input location')
    #plt.xticks(index + bar_width, x, y=0.001)
    #plt.legend()
    #plt.tight_layout()  #could use instead of the subplots_adjust line
    plt.show()

if __name__ == '__main__':
    print("No main loop implemented.  It's just a library, get real.")

max_stars_repo_name: pershint/reacdb

max_issues_repo_name: pershint/reacdb

max_forks_repo_name: pershint/reacdb


text:
# -*- coding: utf-8 -*-
"""
Created on 12 Jul 2019 22:45:39

@author: jiahuei
"""
import tensorflow as tf
import numpy as np
import os
import logging
from tensorflow.contrib.model_pruning.python import pruning_utils
# from tensorflow.contrib.model_pruning.python import pruning
from common import ops_v1 as ops
from common.mask_prune import masked_layer
from common.mask_prune import sampler

logger = logging.getLogger(__name__)
# _NBINS = 256
_NBINS = 512
LOSS_TYPE = ['L1', 'L2', 'hinge_L1']
pjoin = os.path.join
_shape = ops.shape

def calculate_weight_sparsities(weights, weight_op_names=None):
    return calculate_sparsities(tensor_list=weights,
                                count_nnz_fn=lambda x: tf.count_nonzero(x, axis=None, dtype=tf.float32),
                                tensor_op_names=weight_op_names)

def calculate_mask_sparsities(sampled_masks, mask_op_names):
    return calculate_sparsities(tensor_list=sampled_masks,
                                count_nnz_fn=tf.reduce_sum,
                                tensor_op_names=mask_op_names)

def calculate_sparsities(tensor_list, count_nnz_fn, tensor_op_names=None):
    if tensor_op_names is not None:
        assert isinstance(tensor_op_names, list)
    tensor_sizes = [tf.to_float(tf.reduce_prod(_shape(t))) for t in tensor_list]
    tensor_nnz = []
    tensor_sps = []
    for i, m in enumerate(tensor_list):
        m_nnz = count_nnz_fn(m)
        m_sps = tf.subtract(1.0, tf.divide(m_nnz, tensor_sizes[i]))
        tensor_nnz.append(m_nnz)
        if tensor_op_names is None:
            m_name = ''
        else:
            # m_name = '/'.join(tensor_op_names[i].split('/')[-3:])
            m_name = tensor_op_names[i]
        tensor_sps.append((m_name, m_sps))
        # tf.summary.scalar(m_name, m_sps)
    total_nnz = tf.add_n(tensor_nnz)
    total_size = tf.add_n(tensor_sizes)
    total_sparsity = tf.subtract(1.0, tf.divide(total_nnz, total_size))
    return total_sparsity, total_nnz, tensor_sps

def mask_sparsity_summaries(masks_list, mask_op_names):
    """
    Add summary ops for mask sparsity levels. The masks provided must have binary values (either 0. or 1.).
    :param masks_list:
    :param mask_op_names:
    :return:
    """
    with tf.name_scope('sparsity'):
        total_sparsity, total_nnz, mask_sps = calculate_mask_sparsities(masks_list, mask_op_names)
        for sps in mask_sps:
            tf.summary.scalar(*sps)
    tf.summary.scalar('total_nnz', total_nnz)
    tf.summary.scalar('total_sparsity', total_sparsity)
    return total_sparsity

def write_sparsities_to_file(log_dir, val):
    assert 'global_step' in val
    assert 'total_sparsity' in val
    assert 'total_nnz' in val
    assert 'mask_sps' in val

    out = [
        '{}'.format(val['global_step']),
        '{:9.7f}'.format(val['total_sparsity']),
        '{:d}'.format(int(val['total_nnz']))
    ]
    out += ['{:9.7f}'.format(sps[1]) for sps in val['mask_sps']]
    out = '\r\n' + ','.join(out)

    fpath = pjoin(log_dir, 'sparsity_values.csv')
    if not os.path.isfile(fpath):
        headers = 'Global step,Total sparsity,Total NNZ,'
        headers += ','.join([str(sps[0]) for sps in val['mask_sps']])
        out = headers + out
    with open(fpath, 'a') as f:
        f.write(out)

def get_masks(sampling_method='binarise_round', exclude_scopes=None):
    masks = tf.contrib.model_pruning.get_masks()
    mask_sampled_ref = tf.get_collection('masks_sampled')
    is_mag_prune = len(mask_sampled_ref) == 0
    if exclude_scopes is not None:
        assert isinstance(exclude_scopes, (list, tuple))
        masks = tf.contrib.framework.filter_variables(
            var_list=masks,
            include_patterns=None,
            exclude_patterns=exclude_scopes,
            reg_search=True)
        mask_sampled_ref = tf.contrib.framework.filter_variables(
            var_list=mask_sampled_ref,
            include_patterns=None,
            exclude_patterns=exclude_scopes,
            reg_search=True)
    if is_mag_prune:
        logger.debug('get_mask(): Should be magnitude pruning')
        return masks, masks
    else:
        assert sampling_method in ['binarise_round', 'rand', 'sigmoid']
        if sampling_method == 'rand':
            sampled_masks = mask_sampled_ref[:]
        else:
            sampled_masks = []
            for m, m_sampled in zip(masks, mask_sampled_ref):
                if sampling_method == 'binarise_round':
                    m = sampler.binarise_sigmoid(m)
                else:
                    raise NotImplementedError
                    m = tf.nn.sigmoid(m)
                m_sampled_s = _shape(m_sampled)
                if _shape(m) != m_sampled_s:
                    # Mask mode is Structured
                    m = tf.tile(m, multiples=[m_sampled_s[0], m_sampled_s[1] // _shape(m)[-1]])
                sampled_masks.append(m)
        return sampled_masks, masks

def get_weights(exclude_scopes=None):
    weights = tf.contrib.model_pruning.get_weights()
    if exclude_scopes is not None:
        assert isinstance(exclude_scopes, (list, tuple))
        weights = tf.contrib.framework.filter_variables(
            var_list=weights,
            include_patterns=None,
            exclude_patterns=exclude_scopes,
            reg_search=True)
    return weights

def get_mask_assign_ops(mask_type, sparsity_target, exclude_scopes, loss=None):
    if mask_type in masked_layer.MAG_PRUNE_MASKS + [masked_layer.LOTTERY]:
        masks, _ = get_masks(exclude_scopes=exclude_scopes)
        weights = get_weights(exclude_scopes=exclude_scopes)
    else:
        raise ValueError('Invalid mask type. Must be one of {}'.format(
            # masked_layer.MAG_PRUNE_MASKS + masked_layer.MASK_PRUNE))
            masked_layer.MAG_PRUNE_MASKS))
    assert len(weights) == len(masks)
    assert len(masks) > 0

    with tf.name_scope('mask_assign_ops'):
        if mask_type == masked_layer.SNIP:
            # Maybe accumulate saliency
            with tf.variable_scope('accum_saliency'):
                zero_init = tf.initializers.zeros(loss.dtype)
                var_kwargs = dict(dtype=loss.dtype, initializer=zero_init, trainable=False)
                saliency = [tf.get_variable('saliency_m{}'.format(i), shape=_shape(m), **var_kwargs)
                            for i, m in enumerate(masks)]
                # saliency_batch = [tf.abs(s) for s in tf.gradients(ys=loss, xs=masks)]
                saliency_batch = [s for s in tf.gradients(ys=loss, xs=masks)]
                # Ops for accumulating saliency
                accum_ops = [sal.assign_add(sal_b) for (sal, sal_b) in zip(saliency, saliency_batch)]
            # saliency = [tf.abs(s) for s in tf.gradients(ys=loss, xs=masks)]
            mask_ori_shape = [_shape(m) for m in masks]
            mask_num_elems = [np.prod(m) for m in mask_ori_shape]
            saliency_vec = tf.concat([tf.reshape(s, [-1]) for s in saliency], axis=0)
            saliency_vec = tf.abs(saliency_vec)
            saliency_vec = tf.divide(saliency_vec, tf.reduce_sum(saliency_vec))
            num_params = _shape(saliency_vec)[0]
            kappa = int(round(num_params * (1. - sparsity_target)))
            _, ind = tf.nn.top_k(saliency_vec, k=kappa, sorted=True)
            mask_sparse_vec = tf.sparse_to_dense(ind, tf.shape(saliency_vec),
                                                 tf.ones_like(ind, dtype=tf.float32),
                                                 validate_indices=False)
            mask_sparse_split = tf.split(mask_sparse_vec, mask_num_elems)
            mask_sparse = [tf.reshape(m, ms) for m, ms in zip(mask_sparse_split, mask_ori_shape)]
            assign_ops = [tf.assign(m, new_mask) for m, new_mask in zip(masks, mask_sparse)]
            return assign_ops, accum_ops

        elif mask_type == masked_layer.MAG_DIST:
            # Magnitude pruning, class-distribution
            # Calculate standard dev of each class
            # Transform weights as positive factor of standard dev, ie w' = | (w - mean) / std_dev |
            # Reshape and concat all factorised weights, and calculate threshold
            # The rest of the operations are same as class-blind
            abs_weights = []
            for w in weights:
                mean, var = tf.nn.moments(w, axes=list(range(len(_shape(w)))))
                std_dev = tf.sqrt(var)
                w = tf.abs(tf.divide(tf.subtract(w, mean), std_dev))
                abs_weights.append(w)
            criterion = [tf.concat([tf.reshape(w, [-1]) for w in abs_weights], axis=0)]

        else:
            abs_weights = [tf.abs(w) for w in weights]
            # if mask_type in masked_layer.MAG_BLIND + masked_layer.MASK_BLIND:
            if mask_type == masked_layer.MAG_UNIFORM:
                # Magnitude pruning, class-uniform
                criterion = abs_weights
            elif mask_type in (masked_layer.MAG_BLIND, masked_layer.LOTTERY):
                # Magnitude pruning, class-blind
                # We reshape all the weights into a vector, and concat them
                criterion = [tf.concat([tf.reshape(w, [-1]) for w in abs_weights], axis=0)]

        # len == 1 for class-blind, and len == len(weights) for others
        thresholds = [_get_threshold(c, sparsity_target, nbins=_NBINS) for c in criterion]
        if len(thresholds) != len(masks):
            assert len(thresholds) == 1, 'Threshold list should be either of length 1 or equal length as masks list.'

        assign_ops = []
        # new_masks = []
        for index, mask in enumerate(masks):
            abs_w = abs_weights[index]
            threshold = thresholds[min(index, len(thresholds) - 1)]
            new_mask = tf.cast(tf.greater(abs_w, threshold), tf.float32)
            assign_ops.append(tf.assign(mask, new_mask))
            # new_masks.append(new_mask)
        # Assign ops need to be executed for the summaries to capture correct values
        # mask_sparsity_summaries(masks, [m.op.name for m in masks])
    return assign_ops

def conditional_mask_update_op(exclude_scopes,
                               pruning_scheme,
                               global_step,
                               initial_sparsity,
                               final_sparsity,
                               pruning_start_step,
                               pruning_end_step,
                               prune_frequency):
    """
    Conditional mask update ops for gradual pruning.
    https://arxiv.org/abs/1710.01878
    https://github.com/tensorflow/tensorflow/blob/r1.10/tensorflow/contrib/model_pruning

    :param exclude_scopes:
    :param pruning_scheme:
    :param global_step:
    :param initial_sparsity:
    :param final_sparsity:
    :param pruning_start_step:
    :param pruning_end_step:
    :param prune_frequency:
    :return:
    """
    assert pruning_scheme in masked_layer.MAG_PRUNE_MASKS
    if (pruning_end_step - pruning_start_step) % prune_frequency != 0:
        raise ValueError('Pruning end step must be equal to start step added by multiples of frequency.')

    def maybe_update_masks():
        with tf.name_scope('mask_update'):
            is_step_within_pruning_range = tf.logical_and(
                tf.greater_equal(global_step, pruning_start_step),
                # If end_pruning_step is negative, keep pruning forever!
                tf.logical_or(
                    tf.less_equal(global_step, pruning_end_step), tf.less(pruning_end_step, 0)))
            is_pruning_step = tf.equal(
                tf.floormod(tf.subtract(global_step, pruning_start_step), prune_frequency), 0)
            is_pruning_step = tf.logical_and(is_step_within_pruning_range, is_pruning_step)
            return is_pruning_step

    def mask_update_op():
        current_sparsity = _get_current_sparsity(global_step=global_step,
                                                 initial_sparsity=initial_sparsity,
                                                 final_sparsity=final_sparsity,
                                                 pruning_start_step=pruning_start_step,
                                                 pruning_end_step=pruning_end_step)
        # tf.summary.scalar('sparsity_target', current_sparsity)
        mask_assign_ops = get_mask_assign_ops(
            mask_type=pruning_scheme, sparsity_target=current_sparsity, exclude_scopes=exclude_scopes)
        with tf.control_dependencies(mask_assign_ops):
            # logger.info('Updating masks.')
            return tf.no_op('mask_update')
            # return tf.identity(global_step)

    def no_update_op():
        return tf.no_op()
        # return tf.identity(global_step)

    return tf.cond(maybe_update_masks(), mask_update_op, no_update_op)

def _get_current_sparsity(global_step,
                          initial_sparsity,
                          final_sparsity,
                          pruning_start_step,
                          pruning_end_step):
    """
    Get current sparsity level for gradual pruning.
    https://arxiv.org/abs/1710.01878
    https://github.com/tensorflow/tensorflow/blob/r1.10/tensorflow/contrib/model_pruning

    :param global_step:
    :param initial_sparsity:
    :param final_sparsity:
    :param pruning_start_step:
    :param pruning_end_step:
    :return:
    """
    si = initial_sparsity
    sf = final_sparsity
    t = global_step
    t0 = pruning_start_step
    tn = pruning_end_step

    p = tf.div(tf.cast(t - t0, tf.float32), tn - t0)
    p = tf.minimum(1.0, tf.maximum(0.0, p))
    st = tf.add(sf, tf.multiply(si - sf, tf.pow(1 - p, 3)))
    return st

def sparsity_loss(sparsity_target,
                  loss_type='L1',
                  exclude_scopes=None):
    """
    Loss for controlling sparsity of Supermasks.
    :param sparsity_target: Desired sparsity rate.
    :param loss_type: The distance metric.
    :param exclude_scopes: Mask scopes to exclude.
    :return: Scalar loss value.
    """
    assert loss_type in LOSS_TYPE, 'Valid loss functions: {}'.format(LOSS_TYPE)
    if loss_type == 'L1':
        loss_fn = _l1_loss
    elif loss_type == 'L2':
        loss_fn = _l2_loss
    elif loss_type == 'hinge_L1':
        loss_fn = _hinge_l1_loss
    else:
        raise ValueError()
    logger.debug('Using mask sparsity loss: `{}`'.format(loss_type))
    sampled_masks, masks = get_masks(exclude_scopes=exclude_scopes)
    if len(masks) == 0:
        return 0.
    total_sparsity = mask_sparsity_summaries(sampled_masks, [m.op.name for m in masks])
    with tf.name_scope('sparsity'):
        # Log average mask value
        mask_vec = tf.concat([tf.reshape(m, [-1]) for m in masks], axis=0)
        mask_av = tf.reduce_mean(mask_vec)
        tf.summary.scalar('mask_average_val', mask_av)

        with tf.name_scope('loss'):
            loss = loss_fn(total_sparsity, sparsity_target)
        total_size_np = int(sum([np.prod(_shape(m)) for m in sampled_masks]))
        logger.debug('mask_loss: Total mask size: {:,d}'.format(total_size_np))
    return loss

def _l1_loss(curr, target):
    with tf.name_scope('l1'):
        return tf.abs(tf.subtract(target, curr))
        # return tf.abs(tf.subtract(curr, target))

def _l2_loss(curr, target):
    with tf.name_scope('l2'):
        return tf.squared_difference(curr, target)

def _hinge_l1_loss(curr, target):
    with tf.name_scope('hinge_l1'):
        return tf.nn.relu(tf.subtract(target, curr))

def _get_threshold(abs_weights, sparsity_target, nbins, use_tpu=False):
    with tf.name_scope('get_threshold'):
        max_value = tf.reduce_max(abs_weights)
        cdf_fn = pruning_utils.compute_cdf_from_histogram
        if use_tpu:
            cdf_fn = pruning_utils.compute_cdf

        norm_cdf = cdf_fn(abs_weights, [0.0, max_value], nbins=nbins)
        prune_nbins = tf.reduce_sum(tf.cast(tf.less(norm_cdf, sparsity_target), tf.float32))
        threshold = tf.multiply(tf.div(prune_nbins, float(nbins)), max_value)
    return threshold

max_stars_repo_name: jiahuei/tf-sparse-captioning

max_issues_repo_name: jiahuei/tf-sparse-captioning

max_forks_repo_name: jiahuei/tf-sparse-captioning


text:
import torch
import numpy as np

from utils.misc import soft_update

from model.BCAgent import BCAgent
from model.utils.model import *

class BahaviorClone(object):

    def __init__(self, name, params):

        self.name = name
        self.lr = params.lr
        self.gamma = params.gamma
        self.tau = params.tau

        self.obs_dim = params.obs_dim
        self.action_dim = params.action_dim
        self.batch_size = params.batch_size // 2
        self.device = params.device
        self.discrete_action = params.discrete_action_space

        self.agent_index = params.agent_index
        self.num_agents = len(self.agent_index)

        self.cross_entropy_loss = torch.nn.CrossEntropyLoss()

        params.critic.obs_dim = (self.obs_dim + self.action_dim)

        self.agents = [BCAgent(params) for _ in range(self.num_agents)]
        [agent.to(self.device) for agent in self.agents]

    def act(self, observations, sample=False):
        observations = torch.Tensor(observations).to(self.device)

        actions = []
        for agent, obs in zip(self.agents, observations):
            agent.eval()
            actions.append(agent.act(obs, explore=sample).squeeze())
            agent.train()
        return np.array(actions)

    def update(self, replay_buffer, logger, step):
        sample = replay_buffer.sample(self.batch_size, nth=self.agent_index)
        obses, actions, rewards, next_obses, dones = sample

        # split each joint into two single trajectories
        # need to check
        obses = (torch.cat([obses[:, 0], obses[:, 1]], dim=0))
        actions = (torch.cat([actions[:, 0], actions[:, 1]], dim=0))
        rewards = (torch.cat([rewards[:, 0], rewards[:, 1]], dim=0))
        next_obses = (torch.cat([next_obses[:, 0], next_obses[:, 1]], dim=0))
        dones = (torch.cat([dones[:, 0], dones[:, 1]], dim=0))

        if self.discrete_action:
            actions = number_to_onehot(actions)
            actions = torch.max(actions.long(), 1)[1]

        for agent_i, agent in enumerate(self.agents):
            agent.policy_optimizer.zero_grad()
            agent_actions = agent.policy(obses)
            loss = self.cross_entropy_loss(agent_actions, actions)
            loss.backward()
            agent.policy_optimizer.step()

    def save(self, filename):
        raise NotImplementedError

    def load(self, filename):
        raise NotImplementedError


text:
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import mplcursors

data = {'jenis': ['set1', 'set2', 'set3', 'set4', 'set5'],
        'data1': [0.80443, 0.84176, 0.84278, 0.82316, 0.82260],
        'data2': [0.71956, 0.77691, 0.77279, 0.74522, 0.74747],
        'data3': [0.84256, 0.83268, 0.84152, 0.84204, 0.83775],
        'data4': [0.71956, 0.77691, 0.77279, 0.74522, 0.74747],
        'data5': [0.80320, 0.83787, 0.83933, 0.82087, 0.82008],
        'data6': [0.71956, 0.77043, 0.76772, 0.74286, 0.74432],
        'data7': [0.83641, 0.83009, 0.83847, 0.83743, 0.83333],
        'data8': [0.71956, 0.77043, 0.76772, 0.74286, 0.74432]}
df = pd.DataFrame.from_dict(data)

fig = plt.figure(figsize=(8, 4))
ax1 = plt.subplot(111)
for i, data_col in enumerate( df.columns[1:]):
    ax1.plot(df.jenis, [x * 100 for x in df[data_col]], 's-', label=f'Skenario {i+1}')
plt.ylim([60, 100])

for data_col in df.columns[1:]:
    for i,j in df[data_col].items():
         ax1.annotate(str(j), xy=(i, j*100))

mplcursors.cursor(hover=True)

plt.ylabel('Akurasi')
plt.title('Performance Measurements')
plt.legend(loc='best')
plt.show()

max_stars_repo_name: adityadees/grafik

max_issues_repo_name: adityadees/grafik

max_forks_repo_name: adityadees/grafik


text:
from numpy import mean, std
import pickle
import sys

num_runs = sys.argv[1]
U = sys.argv[2]

f = open('result_runs%s_U%s.pkl' % (num_runs, U), 'rb')
data = pickle.load(f)
f.close()
ae_unsw_train = data['ae']['unsw']['train']
ae_nsl_train = data['ae']['nsl']['train']
ae_unsw_test = data['ae']['unsw']['test']
ae_nsl_test = data['ae']['nsl']['test']

sae_unsw_train = data['sae']['unsw']['train']
sae_nsl_train = data['sae']['nsl']['train']
sae_unsw_test = data['sae']['unsw']['test']
sae_nsl_test = data['sae']['nsl']['test']

ae_unified_unsw_train = data['ae_unified']['unsw']['train']
ae_unified_nsl_train = data['ae_unified']['nsl']['train']
ae_unified_unsw_test = data['ae_unified']['unsw']['test']
ae_unified_nsl_test = data['ae_unified']['nsl']['test']

sae_unified_unsw_train = data['sae_unified']['unsw']['train']
sae_unified_nsl_train = data['sae_unified']['nsl']['train']
sae_unified_unsw_test = data['sae_unified']['unsw']['test']
sae_unified_nsl_test = data['sae_unified']['nsl']['test']

unsw_train_avgs = [mean(ae_unsw_train), mean(sae_unsw_train),
                   mean(ae_unified_unsw_train),
                   mean(sae_unified_unsw_train)]
nsl_train_avgs = [mean(ae_nsl_train), mean(sae_nsl_train),
                  mean(ae_unified_nsl_train), mean(sae_unified_nsl_train)]
unsw_test_avgs = [mean(ae_unsw_test), mean(sae_unsw_test),
                  mean(ae_unified_unsw_test), mean(sae_unified_unsw_test)]
nsl_test_avgs = [mean(ae_nsl_test), mean(sae_nsl_test),
                 mean(ae_unified_nsl_test), mean(sae_unified_nsl_test)]

unsw_train_stds = [std(ae_unsw_train), std(sae_unsw_train),
                   std(ae_unified_unsw_train), std(sae_unified_unsw_train)]
nsl_train_stds = [std(ae_nsl_train), std(sae_nsl_train),
                  std(ae_unified_nsl_train), std(sae_unified_nsl_train)]
unsw_test_stds = [std(ae_unsw_test), std(sae_unsw_test),
                  std(ae_unified_unsw_test), std(sae_unified_unsw_test)]
nsl_test_stds = [std(ae_nsl_test), std(sae_nsl_test),
                 std(ae_unified_nsl_test), std(sae_unified_nsl_test)]
print('Use %s shared neurons for %s runs: ' %
      (U, len(sae_unified_unsw_test)))
# pprint(unsw_train_avgs)
# pprint(nsl_train_avgs)
print(unsw_test_avgs)
print(nsl_test_avgs)

max_stars_repo_name: littlepretty/NetLearner

max_issues_repo_name: JinLabIIT/NetLearner

max_forks_repo_name: JinLabIIT/NetLearner

reason: from numpy


text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

'''
gnpy.core.network
=================

This module contains functions for constructing networks of network elements.
'''

from gnpy.core.convert import convert_file
from networkx import DiGraph
from numpy import arange
from scipy.interpolate import interp1d
from logging import getLogger
from os import path
from operator import itemgetter, attrgetter
from gnpy.core import elements
from gnpy.core.elements import Fiber, Edfa, Transceiver, Roadm, Fused, RamanFiber
from gnpy.core.equipment import edfa_nf
from gnpy.core.exceptions import ConfigurationError, NetworkTopologyError
from gnpy.core.units import UNITS
from gnpy.core.utils import (load_json, save_json, round2float, db2lin,
                            merge_amplifier_restrictions)
from gnpy.core.science_utils import SimParams
from collections import namedtuple

logger = getLogger(__name__)

def load_network(filename, equipment, name_matching = False):
    json_filename = ''
    if filename.suffix.lower() == '.xls':
        logger.info('Automatically generating topology JSON file')
        json_filename = convert_file(filename, name_matching)
    elif filename.suffix.lower() == '.json':
        json_filename = filename
    else:
        raise ValueError(f'unsuported topology filename extension {filename.suffix.lower()}')
    json_data = load_json(json_filename)
    return network_from_json(json_data, equipment)

def save_network(filename, network):
    filename_output = path.splitext(filename)[0] + '_auto_design.json'
    json_data = network_to_json(network)
    save_json(json_data, filename_output)

def network_from_json(json_data, equipment):
    # NOTE|dutc: we could use the following, but it would tie our data format
    #            too closely to the graph library
    # from networkx import node_link_graph
    g = DiGraph()
    for el_config in json_data['elements']:
        typ = el_config.pop('type')
        variety = el_config.pop('type_variety', 'default')
        if typ in equipment and variety in equipment[typ]:
            extra_params = equipment[typ][variety]
            temp = el_config.setdefault('params', {})
            temp = merge_amplifier_restrictions(temp, extra_params.__dict__)
            el_config['params'] = temp
        elif typ in ['Edfa', 'Fiber']: # catch it now because the code will crash later!
            raise ConfigurationError(f'The {typ} of variety type {variety} was not recognized:'
                    '\nplease check it is properly defined in the eqpt_config json file')
        cls = getattr(elements, typ)
        el = cls(**el_config)
        g.add_node(el)

    nodes = {k.uid: k for k in g.nodes()}

    for cx in json_data['connections']:
        from_node, to_node = cx['from_node'], cx['to_node']
        try:
            if isinstance(nodes[from_node], Fiber):
                edge_length = nodes[from_node].params.length
            else:
                edge_length = 0.01
            g.add_edge(nodes[from_node], nodes[to_node], weight = edge_length)
        except KeyError:
            raise NetworkTopologyError(f'can not find {from_node} or {to_node} defined in {cx}')

    return g

def network_to_json(network):
    data = {
        'elements': [n.to_json for n in network]
        }
    connections = {
        'connections': [{"from_node": n.uid,
                         "to_node": next_n.uid}
                        for n in network
                        for next_n in network.successors(n) if next_n is not None]
        }
    data.update(connections)
    return data

def select_edfa(raman_allowed, gain_target, power_target, equipment, uid, restrictions=None):
    """amplifer selection algorithm
    @Orange Jean-Luc Augé
    """
    Edfa_list = namedtuple('Edfa_list', 'variety power gain_min nf')
    TARGET_EXTENDED_GAIN = equipment['Span']['default'].target_extended_gain

    # for roadm restriction only: create a dict including not allowed for design amps
    # because main use case is to have specific radm amp which are not allowed for ILA
    # with the auto design
    edfa_dict = {name: amp for (name, amp) in equipment['Edfa'].items()
        if restrictions is None or name in restrictions}

    pin = power_target - gain_target

    # create 2 list of available amplifiers with relevant attributes for their selection

    # edfa list with:
    # extended gain min allowance of 3dB: could be parametrized, but a bit complex
    # extended gain max allowance TARGET_EXTENDED_GAIN is coming from eqpt_config.json
    # power attribut include power AND gain limitations
    edfa_list = [Edfa_list(
                variety=edfa_variety,
                power=min(
                    pin
                    +edfa.gain_flatmax
                    +TARGET_EXTENDED_GAIN,
                    edfa.p_max
                    )
                    -power_target,
                gain_min=
                    gain_target+3
                    -edfa.gain_min,
                nf=edfa_nf(gain_target, edfa_variety, equipment)) \
                for edfa_variety, edfa in edfa_dict.items()
                if ((edfa.allowed_for_design or restrictions is not None) and not edfa.raman)]

    #consider a Raman list because of different gain_min requirement:
    #do not allow extended gain min for Raman
    raman_list = [Edfa_list(
                variety=edfa_variety,
                power=min(
                    pin
                    +edfa.gain_flatmax
                    +TARGET_EXTENDED_GAIN,
                    edfa.p_max
                    )
                    -power_target,
                gain_min=
                    gain_target
                    -edfa.gain_min,
                nf=edfa_nf(gain_target, edfa_variety, equipment))
                for edfa_variety, edfa in edfa_dict.items()
                if (edfa.allowed_for_design and edfa.raman)] \
                if raman_allowed else []

    #merge raman and edfa lists
    amp_list = edfa_list + raman_list

    #filter on min gain limitation:
    acceptable_gain_min_list = [x for x in amp_list if x.gain_min>0]

    if len(acceptable_gain_min_list) < 1:
        #do not take this empty list into account for the rest of the code
        #but issue a warning to the user and do not consider Raman
        #Raman below min gain should not be allowed because i is meant to be a design requirement
        #and raman padding at the amplifier input is impossible!

        if len(edfa_list) < 1:
            raise ConfigurationError(f'auto_design could not find any amplifier \
                    to satisfy min gain requirement in node {uid} \
                    please increase span fiber padding')
        else:
            # TODO: convert to logging
            print(
                f'\x1b[1;31;40m'\
                + f'WARNING: target gain in node {uid} is below all available amplifiers min gain: \
                    amplifier input padding will be assumed, consider increase span fiber padding instead'\
                + '\x1b[0m'
                )
            acceptable_gain_min_list = edfa_list

    #filter on gain+power limitation:
    #this list checks both the gain and the power requirement
    #because of the way .power is calculated in the list
    acceptable_power_list = [x for x in acceptable_gain_min_list if x.power>0]
    if len(acceptable_power_list) < 1:
        #no amplifier satisfies the required power, so pick the highest power(s):
        power_max = max(acceptable_gain_min_list, key=attrgetter('power')).power
        #check and pick if other amplifiers may have a similar gain/power
        #allow a 0.3dB power range
        #this allows to chose an amplifier with a better NF subsequentely
        acceptable_power_list = [x for x in acceptable_gain_min_list
                                 if x.power-power_max>-0.3]

    # gain and power requirements are resolved,
    #       =>chose the amp with the best NF among the acceptable ones:
    selected_edfa = min(acceptable_power_list, key=attrgetter('nf')) #filter on NF
    #check what are the gain and power limitations of this amp
    power_reduction = round(min(selected_edfa.power, 0),2)
    if power_reduction < -0.5:
        print(
            f'\x1b[1;31;40m'\
            + f'WARNING: target gain and power in node {uid}\n \
    is beyond all available amplifiers capabilities and/or extended_gain_range:\n\
    a power reduction of {power_reduction} is applied\n'\
            + '\x1b[0m'
            )

    return selected_edfa.variety, power_reduction

def target_power(network, node, equipment): #get_fiber_dp
    SPAN_LOSS_REF = 20
    POWER_SLOPE = 0.3
    power_mode = equipment['Span']['default'].power_mode
    dp_range = list(equipment['Span']['default'].delta_power_range_db)
    node_loss = span_loss(network, node)

    try:
        dp = round2float((node_loss - SPAN_LOSS_REF) * POWER_SLOPE, dp_range[2])
        dp = max(dp_range[0], dp)
        dp = min(dp_range[1], dp)
    except KeyError:
        raise ConfigurationError(f'invalid delta_power_range_db definition in eqpt_config[Span]'
              f'delta_power_range_db: [lower_bound, upper_bound, step]')

    if isinstance(node, Roadm):
        dp = 0

    return dp

def prev_node_generator(network, node):
    """fused spans interest:
    iterate over all predecessors while they are Fused or Fiber type"""
    try:
        prev_node = next(n for n in network.predecessors(node))
    except StopIteration:
        raise NetworkTopologyError(f'Node {node.uid} is not properly connected, please check network topology')
    # yield and re-iterate
    if isinstance(prev_node, Fused) or isinstance(node, Fused) and not isinstance(prev_node, Roadm):
        yield prev_node
        yield from prev_node_generator(network, prev_node)
    else:
        StopIteration

def next_node_generator(network, node):
    """fused spans interest:
    iterate over all successors while they are Fused or Fiber type"""
    try:
        next_node = next(n for n in network.successors(node))
    except StopIteration:
        raise NetworkTopologyError('Node {node.uid} is not properly connected, please check network topology')
    # yield and re-iterate
    if isinstance(next_node, Fused) or isinstance(node, Fused) and not isinstance(next_node, Roadm):
        yield next_node
        yield from next_node_generator(network, next_node)
    else:
        StopIteration

def span_loss(network, node):
    """Fused span interest:
    return the total span loss of all the fibers spliced by a Fused node"""
    loss = node.loss if node.passive else 0
    try:
        prev_node = next(n for n in network.predecessors(node))
        if isinstance(prev_node, Fused):
            loss += sum(n.loss for n in prev_node_generator(network, node))
    except StopIteration:
        pass
    try:
        next_node = next(n for n in network.successors(node))
        if isinstance(next_node, Fused):
            loss += sum(n.loss for n in next_node_generator(network, node))
    except StopIteration:
        pass
    return loss

def find_first_node(network, node):
    """Fused node interest:
    returns the 1st node at the origin of a succession of fused nodes
    (aka no amp in between)"""
    this_node = node
    for this_node in prev_node_generator(network, node):
        pass
    return this_node

def find_last_node(network, node):
    """Fused node interest:
    returns the last node in a succession of fused nodes
    (aka no amp in between)"""
    this_node = node
    for this_node in next_node_generator(network, node):
        pass
    return this_node

def set_amplifier_voa(amp, power_target, power_mode):
    VOA_MARGIN = 1 #do not maximize the VOA optimization
    if amp.out_voa is None:
        if power_mode:
            gain_target = amp.effective_gain
            voa = min(amp.params.p_max-power_target,
                      amp.params.gain_flatmax-amp.effective_gain)
            voa = max(round2float(max(voa, 0), 0.5) - VOA_MARGIN, 0) if amp.params.out_voa_auto else 0
            amp.delta_p = amp.delta_p + voa
            amp.effective_gain = amp.effective_gain + voa
        else:
            voa = 0 # no output voa optimization in gain mode
        amp.out_voa = voa

def set_egress_amplifier(network, roadm, equipment, pref_total_db):
    power_mode = equipment['Span']['default'].power_mode
    next_oms = (n for n in network.successors(roadm) if not isinstance(n, Transceiver))
    for oms in next_oms:
        #go through all the OMS departing from the Roadm
        node = roadm
        prev_node = roadm
        next_node = oms
        # if isinstance(next_node, Fused): #support ROADM wo egress amp for metro applications
        #     node = find_last_node(next_node)
        #     next_node = next(n for n in network.successors(node))
        #     next_node = find_last_node(next_node)

        if node.per_degree_target_pch_out_db:
            # find the target power on this degree
            try:
                prev_dp = next(el["target_pch_out_db"] for el in \
                    node.per_degree_target_pch_out_db if el["to_node"]==next_node.uid)
            except StopIteration:
                # if no target power is defined on this degree use the global one
                prev_dp = getattr(node.params, 'target_pch_out_db', 0)
        else:
            # if no per degree target power is given use the global one
            prev_dp = getattr(node.params, 'target_pch_out_db', 0)
        dp = prev_dp
        prev_voa = 0
        voa = 0
        while True:
        #go through all nodes in the OMS (loop until next Roadm instance)
            if isinstance(node, Edfa):
                node_loss = span_loss(network, prev_node)
                voa = node.out_voa if node.out_voa else 0
                if node.delta_p is None:
                    dp = target_power(network, next_node, equipment)
                else:
                    dp = node.delta_p
                gain_from_dp = node_loss + dp - prev_dp + prev_voa
                if node.effective_gain is None or power_mode:
                    gain_target = gain_from_dp
                else: #gain mode with effective_gain
                    gain_target = node.effective_gain
                    dp = prev_dp - node_loss + gain_target

                power_target = pref_total_db + dp

                raman_allowed = False
                if isinstance(prev_node, Fiber):
                    max_fiber_lineic_loss_for_raman = \
                            equipment['Span']['default'].max_fiber_lineic_loss_for_raman
                    raman_allowed = prev_node.params.loss_coef < max_fiber_lineic_loss_for_raman

                # implementation of restrictions on roadm boosters
                if isinstance(prev_node,Roadm):
                    if prev_node.restrictions['booster_variety_list']:
                        restrictions = prev_node.restrictions['booster_variety_list']
                    else:
                        restrictions = None
                elif isinstance(next_node,Roadm):
                    # implementation of restrictions on roadm preamp
                    if next_node.restrictions['preamp_variety_list']:
                        restrictions = next_node.restrictions['preamp_variety_list']
                    else:
                        restrictions = None
                else:
                    restrictions = None

                if node.params.type_variety == '':
                    edfa_variety, power_reduction = select_edfa(raman_allowed,
                                   gain_target, power_target, equipment, node.uid, restrictions)
                    extra_params = equipment['Edfa'][edfa_variety]
                    node.params.update_params(extra_params.__dict__)
                    dp += power_reduction
                    gain_target += power_reduction
                elif node.params.raman and not raman_allowed:
                    print(
                        f'\x1b[1;31;40m'\
                        + f'WARNING: raman is used in node {node.uid}\n \
                but fiber lineic loss is above threshold\n'\
                        + '\x1b[0m'
                        )

                node.delta_p = dp if power_mode else None
                node.effective_gain = gain_target
                set_amplifier_voa(node, power_target, power_mode)
            if isinstance(next_node, Roadm) or isinstance(next_node, Transceiver):
                break
            prev_dp = dp
            prev_voa = voa
            prev_node = node
            node = next_node
            # print(f'{node.uid}')
            next_node = next(n for n in network.successors(node))

def add_egress_amplifier(network, node):
    next_nodes = [n for n in network.successors(node)
        if not (isinstance(n, Transceiver) or isinstance(n, Fused) or isinstance(n, Edfa))]
        #no amplification for fused spans or TRX
    for i, next_node in enumerate(next_nodes):
        network.remove_edge(node, next_node)
        amp = Edfa(
                    uid = f'Edfa{i}_{node.uid}',
                    params = {},
                    metadata = {
                        'location': {
                            'latitude':  (node.lat * 2 + next_node.lat * 2) / 4,
                            'longitude': (node.lng * 2 + next_node.lng * 2) / 4,
                            'city':      node.loc.city,
                            'region':    node.loc.region,
                        }
                    },
                    operational = {
                        'gain_target': None,
                        'tilt_target': 0,
                    })
        network.add_node(amp)
        if isinstance(node,Fiber):
            edgeweight = node.params.length
        else:
            edgeweight = 0.01
        network.add_edge(node, amp, weight = edgeweight)
        network.add_edge(amp, next_node, weight = 0.01)

def calculate_new_length(fiber_length, bounds, target_length):
    if fiber_length < bounds.stop:
        return fiber_length, 1

    n_spans = int(fiber_length // target_length)

    length1 = fiber_length / (n_spans+1)
    delta1 = target_length-length1
    result1 = (length1, n_spans+1)

    length2 = fiber_length / n_spans
    delta2 = length2-target_length
    result2 = (length2, n_spans)

    if (bounds.start<=length1<=bounds.stop) and not(bounds.start<=length2<=bounds.stop):
        result = result1
    elif (bounds.start<=length2<=bounds.stop) and not(bounds.start<=length1<=bounds.stop):
        result = result2
    else:
        result = result1 if delta1 < delta2 else result2

    return result

def split_fiber(network, fiber, bounds, target_length, equipment):
    new_length, n_spans = calculate_new_length(fiber.length, bounds, target_length)
    if n_spans == 1:
        return

    try:
        next_node = next(network.successors(fiber))
        prev_node = next(network.predecessors(fiber))
    except StopIteration:
        raise NetworkTopologyError(f'Fiber {fiber.uid} is not properly connected, please check network topology')

    network.remove_node(fiber)

    fiber_params = fiber.params._asdict()
    fiber_params['length'] = new_length / UNITS[fiber.params.length_units]
    fiber_params['con_in'] = fiber.con_in
    fiber_params['con_out'] = fiber.con_out

    f = interp1d([prev_node.lng, next_node.lng], [prev_node.lat, next_node.lat])
    xpos = [prev_node.lng + (next_node.lng - prev_node.lng) * (n+1)/(n_spans+1) for n in range(n_spans)]
    ypos = f(xpos)
    for span, lng, lat in zip(range(n_spans), xpos, ypos):
        new_span = Fiber(uid = f'{fiber.uid}_({span+1}/{n_spans})',
                          metadata = {
                            'location': {
                                'latitude':  lat,
                                'longitude': lng,
                                'city':      fiber.loc.city,
                                'region':    fiber.loc.region,
                            }
                          },
                          params = fiber_params)
        if isinstance(prev_node,Fiber):
            edgeweight = prev_node.params.length
        else:
            edgeweight = 0.01
        network.add_edge(prev_node, new_span, weight = edgeweight)
        prev_node = new_span
    if isinstance(prev_node,Fiber):
        edgeweight = prev_node.params.length
    else:
        edgeweight = 0.01
    network.add_edge(prev_node, next_node, weight = edgeweight)

def add_connector_loss(network, fibers, default_con_in, default_con_out, EOL):
    for fiber in fibers:
        if fiber.con_in is None: fiber.con_in = default_con_in
        if fiber.con_out is None: fiber.con_out = default_con_out
        next_node = next(n for n in network.successors(fiber))
        if not isinstance(next_node, Fused):
            fiber.con_out += EOL

def add_fiber_padding(network, fibers, padding):
    """last_fibers = (fiber for n in network.nodes()
                         if not (isinstance(n, Fiber) or isinstance(n, Fused))
                         for fiber in network.predecessors(n)
                         if isinstance(fiber, Fiber))"""
    for fiber in fibers:
        this_span_loss = span_loss(network, fiber)
        try:
            next_node = next(network.successors(fiber))
        except StopIteration:
            raise NetworkTopologyError(f'Fiber {fiber.uid} is not properly connected, please check network topology')
        if this_span_loss < padding and not (isinstance(next_node, Fused)):
            #add a padding att_in at the input of the 1st fiber:
            #address the case when several fibers are spliced together
            first_fiber = find_first_node(network, fiber)
            # in order to support no booster , fused might be placed
            # just after a roadm: need to check that first_fiber is really a fiber
            if isinstance(first_fiber,Fiber):
                if first_fiber.att_in is None:
                    first_fiber.att_in = padding - this_span_loss
                else:
                    first_fiber.att_in = first_fiber.att_in + padding - this_span_loss

def build_network(network, equipment, pref_ch_db, pref_total_db):
    default_span_data = equipment['Span']['default']
    max_length = int(default_span_data.max_length * UNITS[default_span_data.length_units])
    min_length = max(int(default_span_data.padding/0.2*1e3),50_000)
    bounds = range(min_length, max_length)
    target_length = max(min_length, 90_000)
    default_con_in = default_span_data.con_in
    default_con_out = default_span_data.con_out
    padding = default_span_data.padding

    #set roadm loss for gain_mode before to build network
    fibers = [f for f in network.nodes() if isinstance(f, Fiber)]
    add_connector_loss(network, fibers, default_con_in, default_con_out, default_span_data.EOL)
    add_fiber_padding(network, fibers, padding)
    # don't group split fiber and add amp in the same loop
    # =>for code clarity (at the expense of speed):
    for fiber in fibers:
        split_fiber(network, fiber, bounds, target_length, equipment)

    amplified_nodes = [n for n in network.nodes()
                        if isinstance(n, Fiber) or isinstance(n, Roadm)]

    for node in amplified_nodes:
        add_egress_amplifier(network, node)

    roadms = [r for r in network.nodes() if isinstance(r, Roadm)]
    for roadm in roadms:
        set_egress_amplifier(network, roadm, equipment, pref_total_db)

    #support older json input topology wo Roadms:
    if len(roadms) == 0:
        trx = [t for t in network.nodes() if isinstance(t, Transceiver)]
        for t in trx:
            set_egress_amplifier(network, t, equipment, pref_total_db)

def load_sim_params(filename):
    sim_params = load_json(filename)
    return SimParams(params=sim_params)

def configure_network(network, sim_params):
    for node in network.nodes:
        if isinstance(node, RamanFiber):
            node.sim_params = sim_params

reason: from numpy,from scipy,from networkx


text:
#https://medium.com/octopus-labs-london/how-to-build-a-rest-api-in-python-with-tornado-fc717c33824a
#https://medium.com/octopus-labs-london/how-to-build-a-rest-api-in-python-with-tornado-part-2-8ebc423bd4fa
#https://stackoverflow.com/questions/4383571/importing-files-from-different-folder
from tornado.web import Application, RequestHandler
from tornado.ioloop import IOLoop
import json
import sys
import numpy
import os
# sys.path.append('/Applications/XAMPP/xamppfiles/htdocs/CC1-Project/pymodel')
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from notebook_recommender import *
from urllib.parse import *
topics = []

class QueryHandler(RequestHandler):
  def get(self):
    self.write({'topics': topics})

class ListTopics(RequestHandler):
  #https://stackoverflow.com/questions/35254742/tornado-server-enable-cors-requests/49504274
  def set_default_headers(self):
    self.set_header("Access-Control-Allow-Origin", "*")
    self.set_header("Access-Control-Allow-Headers", "content-type")
    self.set_header('Access-Control-Allow-Methods', 'POST, GET, OPTIONS')

  def post(self):
    global notebooks
    #clear the notebooks array
    notebooks = []
    text = self.request.body
    text = json.loads(text)
    output = query(str(text['text']))
    for i in output:
      notebooks.append(json.loads(json.dumps(i)))

    self.write({'notebooks': notebooks})

  def get(self):
    self.write('some get')

  def options(self):
    self.set_status(204)
    self.finish()

def make_app():
  urls = [("/", QueryHandler),("/api/rec/jupyter", ListTopics)]
  return Application(urls, debug=True)

if __name__ == '__main__':
    app = make_app()
    app.listen(9200)
    IOLoop.instance().start()

max_stars_repo_name: Xiyao-Cheng/OnTimeEvidence

max_issues_repo_name: Xiyao-Cheng/OnTimeEvidence

max_forks_repo_name: Xiyao-Cheng/OnTimeEvidence