text:
import time
import math

from OpenGL.GL import *
from OpenGL.GL.shaders import *
from OpenGL.GL.shaders import glDeleteShader

import numpy as np

from PyEngine3D.Common import logger
from PyEngine3D.App import CoreManager
from PyEngine3D.Render import ScreenQuad
from PyEngine3D.Utilities import Attributes
from .Constants import *
from .Model import *

class Luminance:
    NONE = 0
    APPROXIMATE = 1
    PRECOMPUTED = 2

class Atmosphere:
    def __init__(self, **object_data):
        self.name = object_data.get('name', 'atmosphere')
        self.attributes = Attributes()
        self.is_render_atmosphere = object_data.get('is_render_atmosphere', True)
        self.use_constant_solar_spectrum = False
        self.use_ozone = True
        self.use_combined_textures = True
        self.luminance_type = Luminance.NONE
        self.num_precomputed_wavelengths = 15 if Luminance.PRECOMPUTED == self.luminance_type else 3
        self.do_white_balance = False
        self.show_help = True
        self.view_distance_meters = 9000.0
        self.view_zenith_angle_radians = 1.47
        self.view_azimuth_angle_radians = -0.1
        self.sun_zenith_angle_radians = 1.3
        self.sun_azimuth_angle_radians = 2.9
        self.sun_direction = Float3()

        self.white_point = Float3()
        self.earth_center = Float3(0.0, -kBottomRadius / kLengthUnitInMeters, 0.0)
        self.sun_size = Float2(math.tan(kSunAngularRadius), math.cos(kSunAngularRadius))

        self.kSky = Float3(1.0, 1.0, 1.0)
        self.kSun = Float3(1.0, 1.0, 1.0)

        self.atmosphere_exposure = 0.0001

        # cloud constants
        self.cloud_exposure = 0.1
        self.cloud_altitude = 100.0
        self.cloud_height = 500.0
        self.cloud_speed = 0.01
        self.cloud_absorption = 0.15

        self.cloud_contrast = 2.0
        self.cloud_coverage = 0.8
        self.cloud_tiling = 0.0004

        self.inscatter_power = 0.25

        self.noise_contrast = 1.0
        self.noise_coverage = 1.0
        self.noise_tiling = 0.0003

        self.model = None
        self.atmosphere_material_instance = None

        self.transmittance_texture = None
        self.scattering_texture = None
        self.irradiance_texture = None
        self.optional_single_mie_scattering_texture = None

        self.cloud_texture = None
        self.noise_texture = None

        self.quad = None

        self.load_data(object_data)

    def get_attribute(self):
        save_data = self.get_save_data()
        attribute_names = list(save_data.keys())
        attribute_names.sort()

        for attribute_name in attribute_names:
            self.attributes.set_attribute(attribute_name, save_data[attribute_name])

        return self.attributes

    def set_attribute(self, attribute_name, attribute_value, item_info_history, attribute_index):
        if hasattr(self, attribute_name):
            setattr(self, attribute_name, attribute_value)

    def load_data(self, object_data):
        for key, value in object_data.items():
            if hasattr(self, key):
                setattr(self, key, value)

    def get_save_data(self):
        save_data = dict(
            is_render_atmosphere=self.is_render_atmosphere,
            atmosphere_exposure=self.atmosphere_exposure,
            cloud_exposure=self.cloud_exposure,
            cloud_altitude=self.cloud_altitude,
            cloud_height=self.cloud_height,
            cloud_tiling=self.cloud_tiling,
            cloud_speed=self.cloud_speed,
            cloud_contrast=self.cloud_contrast,
            cloud_coverage=self.cloud_coverage,
            cloud_absorption=self.cloud_absorption,
            inscatter_power=self.inscatter_power,
            noise_tiling=self.noise_tiling,
            noise_contrast=self.noise_contrast,
            noise_coverage=self.noise_coverage,
            sun_size=self.sun_size,
        )
        return save_data

    def initialize(self):
        resource_manager = CoreManager.instance().resource_manager

        self.quad = ScreenQuad.get_vertex_array_buffer()

        self.atmosphere_material_instance = resource_manager.get_material_instance(
            'precomputed_atmosphere.atmosphere',
            macros={
                'USE_LUMINANCE': 1 if self.luminance_type else 0,
                'COMBINED_SCATTERING_TEXTURES': 1 if self.use_combined_textures else 0
            }
        )

        # precompute constants
        max_sun_zenith_angle = 120.0 / 180.0 * kPi

        rayleigh_layer = DensityProfileLayer(0.0, 1.0, -1.0 / kRayleighScaleHeight, 0.0, 0.0)
        mie_layer = DensityProfileLayer(0.0, 1.0, -1.0 / kMieScaleHeight, 0.0, 0.0)

        ozone_density = [DensityProfileLayer(25000.0, 0.0, 0.0, 1.0 / 15000.0, -2.0 / 3.0),
                         DensityProfileLayer(0.0, 0.0, 0.0, -1.0 / 15000.0, 8.0 / 3.0)]

        wavelengths = []
        solar_irradiance = []
        rayleigh_scattering = []
        mie_scattering = []
        mie_extinction = []
        absorption_extinction = []
        ground_albedo = []

        for i in range(kLambdaMin, kLambdaMax + 1, 10):
            L = float(i) * 1e-3  # micro-meters
            mie = kMieAngstromBeta / kMieScaleHeight * math.pow(L, -kMieAngstromAlpha)
            wavelengths.append(i)
            if self.use_constant_solar_spectrum:
                solar_irradiance.append(kConstantSolarIrradiance)
            else:
                solar_irradiance.append(kSolarIrradiance[int((i - kLambdaMin) / 10)])
            rayleigh_scattering.append(kRayleigh * math.pow(L, -4))
            mie_scattering.append(mie * kMieSingleScatteringAlbedo)
            mie_extinction.append(mie)
            if self.use_ozone:
                absorption_extinction.append(kMaxOzoneNumberDensity * kOzoneCrossSection[int((i - kLambdaMin) / 10)])
            else:
                absorption_extinction.append(0.0)
            ground_albedo.append(kGroundAlbedo)

        rayleigh_density = [rayleigh_layer, ]
        mie_density = [mie_layer, ]

        if Luminance.PRECOMPUTED == self.luminance_type:
            self.kSky[...] = [MAX_LUMINOUS_EFFICACY, MAX_LUMINOUS_EFFICACY, MAX_LUMINOUS_EFFICACY]
        else:
            self.kSky[...] = ComputeSpectralRadianceToLuminanceFactors(wavelengths, solar_irradiance, -3)
        self.kSun[...] = ComputeSpectralRadianceToLuminanceFactors(wavelengths, solar_irradiance, 0)

        # generate precomputed textures
        if not resource_manager.texture_loader.hasResource('precomputed_atmosphere.transmittance') or \
                not resource_manager.texture_loader.hasResource('precomputed_atmosphere.scattering') or \
                not resource_manager.texture_loader.hasResource('precomputed_atmosphere.irradiance') or \
                not resource_manager.texture_loader.hasResource(
                    'precomputed_atmosphere.optional_single_mie_scattering') and not self.use_combined_textures:
            model = Model(wavelengths,
                          solar_irradiance,
                          kSunAngularRadius,
                          kBottomRadius,
                          kTopRadius,
                          rayleigh_density,
                          rayleigh_scattering,
                          mie_density,
                          mie_scattering,
                          mie_extinction,
                          kMiePhaseFunctionG,
                          ozone_density,
                          absorption_extinction,
                          ground_albedo,
                          max_sun_zenith_angle,
                          kLengthUnitInMeters,
                          self.num_precomputed_wavelengths,
                          Luminance.PRECOMPUTED == self.luminance_type,
                          self.use_combined_textures)
            model.generate()

        self.transmittance_texture = resource_manager.get_texture('precomputed_atmosphere.transmittance')
        self.scattering_texture = resource_manager.get_texture('precomputed_atmosphere.scattering')
        self.irradiance_texture = resource_manager.get_texture('precomputed_atmosphere.irradiance')

        if not self.use_combined_textures:
            self.optional_single_mie_scattering_texture = resource_manager.get_texture(
                'precomputed_atmosphere.optional_single_mie_scattering')

        self.cloud_texture = resource_manager.get_texture('precomputed_atmosphere.cloud_3d')
        self.noise_texture = resource_manager.get_texture('precomputed_atmosphere.noise_3d')

    def update(self, main_light):
        if not self.is_render_atmosphere:
            return

        self.sun_direction[...] = main_light.transform.front

    def bind_precomputed_atmosphere(self, material_instance):
        material_instance.bind_uniform_data("transmittance_texture", self.transmittance_texture)
        material_instance.bind_uniform_data("scattering_texture", self.scattering_texture)
        material_instance.bind_uniform_data("irradiance_texture", self.irradiance_texture)

        if not self.use_combined_textures:
            material_instance.bind_uniform_data(
                "single_mie_scattering_texture", self.optional_single_mie_scattering_texture)

        material_instance.bind_uniform_data("SKY_RADIANCE_TO_LUMINANCE", self.kSky * self.atmosphere_exposure)
        material_instance.bind_uniform_data("SUN_RADIANCE_TO_LUMINANCE", self.kSun * self.atmosphere_exposure)

        material_instance.bind_uniform_data("atmosphere_exposure", self.atmosphere_exposure)
        material_instance.bind_uniform_data("earth_center", self.earth_center)

    def bind_cloud(self, material_instance):
        material_instance.bind_uniform_data("texture_cloud", self.cloud_texture)
        material_instance.bind_uniform_data("texture_noise", self.noise_texture)

        material_instance.bind_uniform_data('cloud_exposure', self.cloud_exposure)
        material_instance.bind_uniform_data('cloud_altitude', self.cloud_altitude)
        material_instance.bind_uniform_data('cloud_height', self.cloud_height)
        material_instance.bind_uniform_data('cloud_speed', self.cloud_speed)
        material_instance.bind_uniform_data('cloud_absorption', self.cloud_absorption)

        material_instance.bind_uniform_data('cloud_tiling', self.cloud_tiling)
        material_instance.bind_uniform_data('cloud_contrast', self.cloud_contrast)
        material_instance.bind_uniform_data('cloud_coverage', self.cloud_coverage)

        material_instance.bind_uniform_data('noise_tiling', self.noise_tiling)
        material_instance.bind_uniform_data('noise_contrast', self.noise_contrast)
        material_instance.bind_uniform_data('noise_coverage', self.noise_coverage)

    def render_precomputed_atmosphere(self, texture_linear_depth, texture_shadow, render_light_probe_mode):
        if not self.is_render_atmosphere:
            return

        self.atmosphere_material_instance.use_program()
        self.atmosphere_material_instance.bind_material_instance()
        self.atmosphere_material_instance.bind_uniform_data("texture_linear_depth", texture_linear_depth)
        self.atmosphere_material_instance.bind_uniform_data("texture_shadow", texture_shadow)
        self.atmosphere_material_instance.bind_uniform_data("sun_size", self.sun_size)
        self.atmosphere_material_instance.bind_uniform_data("render_light_probe_mode", render_light_probe_mode)
        self.bind_precomputed_atmosphere(self.atmosphere_material_instance)
        self.bind_cloud(self.atmosphere_material_instance)
        self.quad.draw_elements()

max_forks_repo_name: ubuntunux/GuineaPig


text:
#!/usr/bin/env python3

"""
This module contains unit tests of the :mod:`ccg.backend` module.
"""

import unittest

import numpy as np
import pybel
from rdkit import Chem

from ccg.backend import rdk

class RDKitMolTest(unittest.TestCase):
    """
    Contains unit tests of RDKitMol Class
    """
    def setUp(self):
        self.smiles = 'C'
        rd_mol = Chem.MolFromSmiles('C')
        self.rd_mol = Chem.AddHs(rd_mol)
        self.vdw_mat = np.array([[0., 0., 0., 0., 0.],
                                [0., 0., 0.96, 0.96, 0.96],
                                [0., 0., 0., 0.96, 0.96],
                                [0., 0., 0., 0., 0.96],
                                [0., 0., 0., 0., 0.]])
        self.xyz = '''14

O       1.85073500   -1.14075800   -0.80808500
O       1.64413700   -0.98958800    0.61920900
C       0.89993200    0.19167500    0.83332300
C       1.57997200    1.46553500    0.32458600
C       0.87059800    2.73251700    0.76632100
O       2.94131500    1.54377500    0.86706900
O       3.78534700    0.75777800    0.22699800
H       2.74479300   -0.76185300   -0.87545700
H       0.78501500    0.22753800    1.92086600
H      -0.09098800    0.11998200    0.36665200
H       1.69085200    1.41865800   -0.75897600
H       0.84097800    2.80358600    1.85617500
H      -0.15579100    2.73441900    0.39232200
H       1.37972400    3.61444100    0.37512000
'''
        self.ob_mol = pybel.readstring('xyz', self.xyz).OBMol
        self.rdkitmol = rdk.RDKitMol.FromOBMol(self.ob_mol)
        self.torsions = [[7, 0, 1, 2], [0, 1, 2, 3], [1, 2, 3, 4],
                         [2, 3, 4, 11], [2, 3, 5, 6]]

    def test_init(self):
        methane = rdk.RDKitMol(self.rd_mol)
        self.assertEqual(methane.GetNumAtoms(), 5)
        self.assertEqual(methane.GetNumBonds(), 4)
        atomic_nums = [atom.GetAtomicNum() for atom in methane.GetAtoms()]
        self.assertEqual(atomic_nums.count(6), 1)
        self.assertEqual(atomic_nums.count(1), 4)

    def test_FromSmiles(self):
        methane = rdk.RDKitMol.FromSmiles(self.smiles)
        self.assertEqual(methane.GetNumAtoms(), 5)
        self.assertEqual(methane.GetNumBonds(), 4)
        atomic_nums = [atom.GetAtomicNum() for atom in methane.GetAtoms()]
        self.assertEqual(atomic_nums.count(6), 1)
        self.assertEqual(atomic_nums.count(1), 4)

    def test_FromOBMol(self):
        self.assertEqual(self.rdkitmol.GetNumAtoms(), 14)
        self.assertEqual(self.rdkitmol.GetNumBonds(), 13)
        atomic_nums = [atom.GetAtomicNum() for atom in self.rdkitmol.GetAtoms()]
        self.assertEqual(atomic_nums.count(6), 3)
        self.assertEqual(atomic_nums.count(8), 4)
        self.assertEqual(atomic_nums.count(1), 7)

    def test_GetTorsionalModes(self):
        # TODO: Needs more examples
        self.assertEqual(self.rdkitmol.GetTorsionalModes(), self.torsions)

    def test_EmbedConformer(self):
        self.rdkitmol.EmbedConformer()
        self.assertEqual(self.rdkitmol.GetNumConformers(), 1)

    def test_EmbedMultipleConfs(self):
        self.rdkitmol.EmbedMultipleConfs(10)
        self.assertEqual(self.rdkitmol.GetNumConformers(), 10)

    def test_GetConformer(self):
        self.rdkitmol.RemoveAllConformers()
        with self.assertRaises(ValueError):
            self.rdkitmol.GetConformer()
        self.rdkitmol.EmbedConformer()
        conf = self.rdkitmol.GetConformer()
        self.assertTrue(isinstance(conf, rdk.RDKitConf))

    def test_GetConformers(self):
        self.rdkitmol.RemoveAllConformers()
        with self.assertRaises(ValueError):
            self.rdkitmol.GetConformers([1, 5])
        self.rdkitmol.EmbedMultipleConfs(8)
        confs = self.rdkitmol.GetConformers([1, 5])
        self.assertEqual(len(confs), 2)
        for conf in confs:
            self.assertTrue(isinstance(conf, rdk.RDKitConf))
            self.assertIn(conf.GetId(), [1, 5])

    def test_GetAllConformers(self):
        self.rdkitmol.RemoveAllConformers()
        confs = self.rdkitmol.GetAllConformers()
        self.assertEqual(len(confs), 0)
        self.rdkitmol.EmbedMultipleConfs(8)
        confs = self.rdkitmol.GetAllConformers()
        self.assertEqual(len(confs), 8)

    def test_ToRDMol(self):
        rd_mol = self.rdkitmol.ToRDMol()
        self.assertTrue(isinstance(rd_mol, Chem.rdchem.Mol))
        self.assertEqual(rd_mol.GetNumAtoms(), 14)
        self.assertEqual(rd_mol.GetNumBonds(), 13)
        atomic_nums = [atom.GetAtomicNum()
                       for atom in rd_mol.GetAtoms()]
        self.assertEqual(atomic_nums.count(6), 3)
        self.assertEqual(atomic_nums.count(8), 4)
        self.assertEqual(atomic_nums.count(1), 7)

    def test_GetVdwMatrix(self):
        methane = rdk.RDKitMol(self.rd_mol)
        methane.SetVdwMatrix()
        m = methane.GetVdwMatrix()
        self.assertTrue(np.allclose(m,
                                    self.vdw_mat))

    def test_SetVdwMatrix(self):
        methane = rdk.RDKitMol(self.rd_mol)
        methane.SetVdwMatrix(threshold=0.4)
        m1 = methane.GetVdwMatrix()
        self.assertTrue(np.allclose(m1, self.vdw_mat))
        methane.SetVdwMatrix(threshold=0.5)
        m2 = methane.GetVdwMatrix()
        self.assertTrue(np.all(m1 <= m2))
        self.assertTrue(np.allclose(m1 / 4 * 5, m2))

if __name__ == '__main__':
    unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))

ext: py

max_stars_repo_name: xiaoruiDong/CCG

max_issues_repo_name: xiaoruiDong/CCG

max_forks_repo_name: xiaoruiDong/CCG

reason: import numpy


text:
# create the optimization problem
from pymoo.model.termination import MaximumGenerationTermination, Termination

from pymoo.optimize import minimize

from pymop.factory import get_problem

problem = get_problem("zdt1")
pf = problem.pareto_front()

# ----------------------------------------------
# Number of Generations
# ----------------------------------------------
res = minimize(problem,
               method='nsga2',
               method_args={'pop_size': 20},
               termination=('n_gen', 5),
               pf=pf,
               disp=True)

# ----------------------------------------------
# Number of Function Evaluations
# ----------------------------------------------

res = minimize(problem,
               method='nsga2',
               method_args={'pop_size': 20},
               termination=('n_eval', 500),
               pf=pf,
               disp=True)

# ----------------------------------------------
# Until specific IGD is reached
# ----------------------------------------------

res = minimize(problem,
               method='nsga2',
               method_args={'pop_size': 100},
               termination=('igd', 0.1),
               pf=pf,
               disp=True)

# ----------------------------------------------
# Using a Custom Object - (Custom Termination Criteria can be defined)
# ----------------------------------------------

res = minimize(problem,
               method='nsga2',
               method_args={'pop_size': 40},
               termination=MaximumGenerationTermination(25),
               pf=pf,
               disp=True)

# ----------------------------------------------
# Define your own termination - Here just random termination :-D
# ----------------------------------------------

class MyTermination(Termination):

    def __init__(self, perc_improvement) -> None:
        super().__init__()
        self.perc_improvement = perc_improvement

    def _do_continue(self, algorithm):
        import numpy as np
        return np.random.random() > 0.05

res = minimize(problem,
               method='nsga2',
               method_args={'pop_size': 40},
               termination=MyTermination(0.01),
               save_history=True,
               pf=pf,
               disp=True)


text:
# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License").
# You may not use this file except in compliance with the License.
# A copy of the License is located at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# or in the "license" file accompanying this file. This file is distributed
# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
# express or implied. See the License for the specific language governing
# permissions and limitations under the License.

from typing import Iterator, List, Tuple, Optional

import numpy as np
import torch
from scipy.special import erf, erfinv

from pts.core.component import validated
from pts.dataset import DataEntry
from pts.exception import assert_pts
from .transform import (
    SimpleTransformation,
    MapTransformation,
    FlatMapTransformation,
)

class AsNumpyArray(SimpleTransformation):
    """
    Converts the value of a field into a numpy array.

    Parameters
    ----------
    expected_ndim
        Expected number of dimensions. Throws an exception if the number of
        dimensions does not match.
    dtype
        numpy dtype to use.
    """

    @validated()
    def __init__(
        self, field: str, expected_ndim: int, dtype: np.dtype = np.float32
    ) -> None:
        self.field = field
        self.expected_ndim = expected_ndim
        self.dtype = dtype

    def transform(self, data: DataEntry) -> DataEntry:
        value = data[self.field]
        if not isinstance(value, float):
            # this lines produces "ValueError: setting an array element with a
            # sequence" on our test
            # value = np.asarray(value, dtype=np.float32)
            # see https://stackoverflow.com/questions/43863748/
            value = np.asarray(list(value), dtype=self.dtype)
        else:
            # ugly: required as list conversion will fail in the case of a
            # float
            value = np.asarray(value, dtype=self.dtype)
        assert_pts(
            value.ndim >= self.expected_ndim,
            'Input for field "{self.field}" does not have the required'
            "dimension (field: {self.field}, ndim observed: {value.ndim}, "
            "expected ndim: {self.expected_ndim})",
            value=value,
            self=self,
        )
        data[self.field] = value
        return data

class ExpandDimArray(SimpleTransformation):
    """
    Expand dims in the axis specified, if the axis is not present does nothing.
    (This essentially calls np.expand_dims)

    Parameters
    ----------
    field
        Field in dictionary to use
    axis
        Axis to expand (see np.expand_dims for details)
    """

    @validated()
    def __init__(self, field: str, axis: Optional[int] = None) -> None:
        self.field = field
        self.axis = axis

    def transform(self, data: DataEntry) -> DataEntry:
        if self.axis is not None:
            data[self.field] = np.expand_dims(data[self.field], axis=self.axis)
        return data

class VstackFeatures(SimpleTransformation):
    """
    Stack fields together using ``np.vstack``.

    Fields with value ``None`` are ignored.

    Parameters
    ----------
    output_field
        Field name to use for the output
    input_fields
        Fields to stack together
    drop_inputs
        If set to true the input fields will be dropped.
    """

    @validated()
    def __init__(
        self,
        output_field: str,
        input_fields: List[str],
        drop_inputs: bool = True,
    ) -> None:
        self.output_field = output_field
        self.input_fields = input_fields
        self.cols_to_drop = (
            []
            if not drop_inputs
            else [
                fname for fname in self.input_fields if fname != output_field
            ]
        )

    def transform(self, data: DataEntry) -> DataEntry:
        r = [
            data[fname]
            for fname in self.input_fields
            if data[fname] is not None
        ]
        output = np.vstack(r)
        data[self.output_field] = output
        for fname in self.cols_to_drop:
            del data[fname]
        return data

class ConcatFeatures(SimpleTransformation):
    """
    Concatenate fields together using ``np.concatenate``.

    Fields with value ``None`` are ignored.

    Parameters
    ----------
    output_field
        Field name to use for the output
    input_fields
        Fields to stack together
    drop_inputs
        If set to true the input fields will be dropped.
    """

    @validated()
    def __init__(
        self,
        output_field: str,
        input_fields: List[str],
        drop_inputs: bool = True,
    ) -> None:
        self.output_field = output_field
        self.input_fields = input_fields
        self.cols_to_drop = (
            []
            if not drop_inputs
            else [
                fname for fname in self.input_fields if fname != output_field
            ]
        )

    def transform(self, data: DataEntry) -> DataEntry:
        r = [
            data[fname]
            for fname in self.input_fields
            if data[fname] is not None
        ]
        output = np.concatenate(r)
        data[self.output_field] = output
        for fname in self.cols_to_drop:
            del data[fname]
        return data

class SwapAxes(SimpleTransformation):
    """
    Apply `np.swapaxes` to fields.

    Parameters
    ----------
    input_fields
        Field to apply to
    axes
        Axes to use
    """

    @validated()
    def __init__(self, input_fields: List[str], axes: Tuple[int, int]) -> None:
        self.input_fields = input_fields
        self.axis1, self.axis2 = axes

    def transform(self, data: DataEntry) -> DataEntry:
        for field in self.input_fields:
            data[field] = self.swap(data[field])
        return data

    def swap(self, v):
        if isinstance(v, np.ndarray):
            return np.swapaxes(v, self.axis1, self.axis2)
        if isinstance(v, list):
            return [self.swap(x) for x in v]
        else:
            raise ValueError(
                f"Unexpected field type {type(v).__name__}, expected "
                f"np.ndarray or list[np.ndarray]"
            )

class ListFeatures(SimpleTransformation):
    """
    Creates a new field which contains a list of features.

    Parameters
    ----------
    output_field
        Field name for output
    input_fields
        Fields to combine into list
    drop_inputs
        If true the input fields will be removed from the result.
    """

    @validated()
    def __init__(
        self,
        output_field: str,
        input_fields: List[str],
        drop_inputs: bool = True,
    ) -> None:
        self.output_field = output_field
        self.input_fields = input_fields
        self.cols_to_drop = (
            []
            if not drop_inputs
            else [
                fname for fname in self.input_fields if fname != output_field
            ]
        )

    def transform(self, data: DataEntry) -> DataEntry:
        data[self.output_field] = [data[fname] for fname in self.input_fields]
        for fname in self.cols_to_drop:
            del data[fname]
        return data

class TargetDimIndicator(SimpleTransformation):
    """
    Label-encoding of the target dimensions.
    """

    @validated()
    def __init__(self, field_name: str, target_field: str) -> None:
        self.field_name = field_name
        self.target_field = target_field

    def transform(self, data: DataEntry) -> DataEntry:
        data[self.field_name] = np.arange(0, data[self.target_field].shape[0])
        return data

class SampleTargetDim(FlatMapTransformation):
    """
    Samples random dimensions from the target at training time.
    """

    @validated()
    def __init__(
        self,
        field_name: str,
        target_field: str,
        observed_values_field: str,
        num_samples: int,
        shuffle: bool = True,
    ) -> None:
        self.field_name = field_name
        self.target_field = target_field
        self.observed_values_field = observed_values_field
        self.num_samples = num_samples
        self.shuffle = shuffle

    def flatmap_transform(
        self, data: DataEntry, is_train: bool, slice_future_target: bool = True
    ) -> Iterator[DataEntry]:
        if not is_train:
            yield data
        else:
            # (target_dim,)
            target_dimensions = data[self.field_name]

            if self.shuffle:
                np.random.shuffle(target_dimensions)

            target_dimensions = target_dimensions[: self.num_samples]

            data[self.field_name] = target_dimensions
            # (seq_len, target_dim) -> (seq_len, num_samples)

            for field in [
                f"past_{self.target_field}",
                f"future_{self.target_field}",
                f"past_{self.observed_values_field}",
                f"future_{self.observed_values_field}",
            ]:
                data[field] = data[field][:, target_dimensions]

            yield data

class CDFtoGaussianTransform(MapTransformation):
    """
    Marginal transformation that transforms the target via an empirical CDF
    to a standard gaussian as described here: https://arxiv.org/abs/1910.03002

    To be used in conjunction with a multivariate gaussian to from a copula.
    Note that this transformation is currently intended for multivariate
    targets only.
    """

    @validated()
    def __init__(
        self,
        target_dim: int,
        target_field: str,
        observed_values_field: str,
        cdf_suffix="_cdf",
        max_context_length: Optional[int] = None,
    ) -> None:
        """
        Constructor for CDFtoGaussianTransform.

        Parameters
        ----------
        target_dim
            Dimensionality of the target.
        target_field
            Field that will be transformed.
        observed_values_field
            Field that indicates observed values.
        cdf_suffix
            Suffix to mark the field with the transformed target.
        max_context_length
            Sets the maximum context length for the empirical CDF.
        """
        self.target_field = target_field
        self.past_target_field = "past_" + self.target_field
        self.future_target_field = "future_" + self.target_field
        self.past_observed_field = f"past_{observed_values_field}"
        self.sort_target_field = f"past_{target_field}_sorted"
        self.slopes_field = "slopes"
        self.intercepts_field = "intercepts"
        self.cdf_suffix = cdf_suffix
        self.max_context_length = max_context_length
        self.target_dim = target_dim

    def map_transform(self, data: DataEntry, is_train: bool) -> DataEntry:
        self._preprocess_data(data, is_train=is_train)
        self._calc_pw_linear_params(data)

        for target_field in [self.past_target_field, self.future_target_field]:
            data[target_field + self.cdf_suffix] = self.standard_gaussian_ppf(
                self._empirical_cdf_forward_transform(
                    data[self.sort_target_field],
                    data[target_field],
                    data[self.slopes_field],
                    data[self.intercepts_field],
                )
            )
        return data

    def _preprocess_data(self, data: DataEntry, is_train: bool):
        """
        Performs several preprocess operations for computing the empirical CDF.
        1) Reshaping the data.
        2) Normalizing the target length.
        3) Adding noise to avoid zero slopes (training only)
        4) Sorting the target to compute the empirical CDF

        Parameters
        ----------
        data
            DataEntry with input data.
        is_train
            if is_train is True, this function adds noise to the target to
            avoid zero slopes in the piece-wise linear function.
        Returns
        -------

        """
        # (target_length, target_dim)
        past_target_vec = data[self.past_target_field].copy()

        # pick only observed values
        target_length, target_dim = past_target_vec.shape

        # (target_length, target_dim)
        past_observed = (data[self.past_observed_field] > 0) * (
            data["past_is_pad"].reshape((-1, 1)) == 0
        )
        assert past_observed.ndim == 2
        assert target_dim == self.target_dim

        past_target_vec = past_target_vec[past_observed.min(axis=1)]

        assert past_target_vec.ndim == 2
        assert past_target_vec.shape[1] == self.target_dim

        expected_length = (
            target_length
            if self.max_context_length is None
            else self.max_context_length
        )

        if target_length != expected_length:
            # Fills values in the case where past_target_vec.shape[-1] <
            # target_length
            # as dataset.loader.BatchBuffer does not support varying shapes
            past_target_vec = CDFtoGaussianTransform._fill(
                past_target_vec, expected_length
            )

        # sorts along the time dimension to compute empirical CDF of each
        # dimension
        if is_train:
            past_target_vec = self._add_noise(past_target_vec)

        past_target_vec.sort(axis=0)

        assert past_target_vec.shape == (expected_length, self.target_dim)

        data[self.sort_target_field] = past_target_vec

    def _calc_pw_linear_params(self, data: DataEntry):
        """
        Calculates the piece-wise linear parameters to interpolate between
        the observed values in the empirical CDF.

        Once current limitation is that we use a zero slope line as the last
        piece. Thus, we cannot forecast anything higher than the highest
        observed value.

        Parameters
        ----------
        data
            Input data entry containing a sorted target field.

        Returns
        -------

        """
        sorted_target = data[self.sort_target_field]
        sorted_target_length, target_dim = sorted_target.shape

        quantiles = (
            np.stack(
                [np.arange(sorted_target_length) for _ in range(target_dim)],
                axis=1,
            )
            / float(sorted_target_length)
        )

        x_diff = np.diff(sorted_target, axis=0)
        y_diff = np.diff(quantiles, axis=0)

        # Calculate slopes of the pw-linear pieces.
        slopes = np.where(
            x_diff == 0.0, np.zeros_like(x_diff), y_diff / x_diff
        )

        zeroes = np.zeros_like(np.expand_dims(slopes[0, :], axis=0))
        slopes = np.append(slopes, zeroes, axis=0)

        # Calculate intercepts of the pw-linear pieces.
        intercepts = quantiles - slopes * sorted_target

        # Populate new fields with the piece-wise linear parameters.
        data[self.slopes_field] = slopes
        data[self.intercepts_field] = intercepts

    def _empirical_cdf_forward_transform(
        self,
        sorted_values: np.ndarray,
        values: np.ndarray,
        slopes: np.ndarray,
        intercepts: np.ndarray,
    ) -> np.ndarray:
        """
        Applies the empirical CDF forward transformation.

        Parameters
        ----------
        sorted_values
            Sorted target vector.
        values
            Values (real valued) that will be transformed to empirical CDF
            values.
        slopes
            Slopes of the piece-wise linear function.
        intercepts
            Intercepts of the piece-wise linear function.

        Returns
        -------
        quantiles
            Empirical CDF quantiles in [0, 1] interval with winzorized cutoff.

        """
        m = sorted_values.shape[0]
        quantiles = self._forward_transform(
            sorted_values, values, slopes, intercepts
        )

        quantiles = np.clip(
            quantiles, self.winsorized_cutoff(m), 1 - self.winsorized_cutoff(m)
        )
        return quantiles

    @staticmethod
    def _add_noise(x: np.array) -> np.array:
        scale_noise = 0.2
        std = np.sqrt(
            (np.square(x - x.mean(axis=1, keepdims=True))).mean(
                axis=1, keepdims=True
            )
        )
        noise = np.random.normal(
            loc=np.zeros_like(x), scale=np.ones_like(x) * std * scale_noise
        )
        x = x + noise
        return x

    @staticmethod
    def _search_sorted(
        sorted_vec: np.array, to_insert_vec: np.array
    ) -> np.array:
        """
        Finds the indices of the active piece-wise linear function.

        Parameters
        ----------
        sorted_vec
            Sorted target vector.
        to_insert_vec
            Vector for which the indicies of the active linear functions
            will be computed

        Returns
        -------
        indices
            Indices mapping to the active linear function.
        """
        indices_left = np.searchsorted(sorted_vec, to_insert_vec, side="left")
        indices_right = np.searchsorted(
            sorted_vec, to_insert_vec, side="right"
        )

        indices = indices_left + (indices_right - indices_left) // 2
        indices = indices - 1
        indices = np.minimum(indices, len(sorted_vec) - 1)
        indices[indices < 0] = 0
        return indices

    def _forward_transform(
        self,
        sorted_vec: np.array,
        target: np.array,
        slopes: np.array,
        intercepts: np.array,
    ) -> np.array:
        """
        Applies the forward transformation to the marginals of the multivariate
        target. Target (real valued) -> empirical cdf [0, 1]

        Parameters
        ----------
        sorted_vec
            Sorted (past) target vector.
        target
            Target that will be transformed.
        slopes
            Slopes of the piece-wise linear function.
        intercepts
            Intercepts of the piece-wise linear function

        Returns
        -------
        transformed_target
            Transformed target vector.
        """
        transformed = list()
        for sorted, t, slope, intercept in zip(
            sorted_vec.transpose(),
            target.transpose(),
            slopes.transpose(),
            intercepts.transpose(),
        ):
            indices = self._search_sorted(sorted, t)
            transformed_value = slope[indices] * t + intercept[indices]
            transformed.append(transformed_value)
        return np.array(transformed).transpose()

    @staticmethod
    def standard_gaussian_cdf(x: np.array) -> np.array:
        u = x / (np.sqrt(2.0))
        return (erf(u) + 1.0) / 2.0

    @staticmethod
    def standard_gaussian_ppf(y: np.array) -> np.array:
        y_clipped = np.clip(y, a_min=1.0e-6, a_max=1.0 - 1.0e-6)
        return np.sqrt(2.0) * erfinv(2.0 * y_clipped - 1.0)

    @staticmethod
    def winsorized_cutoff(m: np.array) -> np.array:
        """
        Apply truncation to the empirical CDF estimator to reduce variance as
        described here: https://arxiv.org/abs/0903.0649

        Parameters
        ----------
        m
            Input array with empirical CDF values.

        Returns
        -------
        res
            Truncated empirical CDf values.
        """
        res = 1 / (4 * m ** 0.25 * np.sqrt(3.14 * np.log(m)))
        assert 0 < res < 1
        return res

    @staticmethod
    def _fill(target: np.ndarray, expected_length: int) -> np.ndarray:
        """
        Makes sure target has at least expected_length time-units by repeating
        it or using zeros.

        Parameters
        ----------
        target : shape (seq_len, dim)
        expected_length

        Returns
        -------
            array of shape (target_length, dim)
        """

        current_length, target_dim = target.shape
        if current_length == 0:
            # todo handle the case with no observation better,
            # we could use dataset statistics but for now we use zeros
            filled_target = np.zeros((expected_length, target_dim))
        elif current_length < expected_length:
            filled_target = np.vstack(
                [target for _ in range(expected_length // current_length + 1)]
            )
            filled_target = filled_target[:expected_length]
        elif current_length > expected_length:
            filled_target = target[-expected_length:]
        else:
            filled_target = target

        assert filled_target.shape == (expected_length, target_dim)

        return filled_target

def cdf_to_gaussian_forward_transform(
    input_batch: DataEntry, outputs: torch.Tensor
) -> np.ndarray:
    """
    Forward transformation of the CDFtoGaussianTransform.

    Parameters
    ----------
    input_batch
        Input data to the predictor.
    outputs
        Predictor outputs.
    Returns
    -------
    outputs
        Forward transformed outputs.

    """

    def _empirical_cdf_inverse_transform(
        batch_target_sorted: torch.Tensor,
        batch_predictions: torch.Tensor,
        slopes: torch.Tensor,
        intercepts: torch.Tensor,
    ) -> np.ndarray:
        """
        Apply forward transformation of the empirical CDF.

        Parameters
        ----------
        batch_target_sorted
            Sorted targets of the input batch.
        batch_predictions
            Predictions of the underlying probability distribution
        slopes
            Slopes of the piece-wise linear function.
        intercepts
            Intercepts of the piece-wise linear function.

        Returns
        -------
        outputs
            Forward transformed outputs.

        """
        slopes = slopes.cpu().numpy()
        intercepts = intercepts.cpu().numpy()

        batch_target_sorted = batch_target_sorted.cpu().numpy()
        _, num_timesteps, _ = batch_target_sorted.shape
        indices = np.floor(batch_predictions * num_timesteps)
        # indices = indices - 1
        # for now project into [0, 1]
        indices = np.clip(indices, 0, num_timesteps - 1)
        indices = indices.astype(np.int)

        transformed = np.where(
            np.take_along_axis(slopes, indices, axis=1) != 0.0,
            (
                batch_predictions
                - np.take_along_axis(intercepts, indices, axis=1)
            )
            / np.take_along_axis(slopes, indices, axis=1),
            np.take_along_axis(batch_target_sorted, indices, axis=1),
        )
        return transformed

    # applies inverse cdf to all outputs
    _, samples, _, _ = outputs.shape
    for sample_index in range(0, samples):
        outputs[:, sample_index, :, :] = _empirical_cdf_inverse_transform(
            input_batch["past_target_sorted"],
            CDFtoGaussianTransform.standard_gaussian_cdf(
                outputs[:, sample_index, :, :]
            ),
            input_batch["slopes"],
            input_batch["intercepts"],
        )
    return outputs

max_stars_repo_name: daobook/gluon-ts

max_issues_repo_name: daobook/gluon-ts

max_forks_repo_name: daobook/gluon-ts

reason: import numpy,from scipy


text:
import pytest
import numpy as np
import tensorflow as tf

import zfit
from zfit.core.sample import extract_extended_pdfs, extended_sampling

obs1 = 'obs1'

@pytest.mark.flaky(reruns=3)  # poissonian sampling
def test_extract_extended_pdfs():
    gauss1 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)
    gauss2 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)
    gauss3 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)
    gauss4 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)
    gauss5 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)
    gauss6 = zfit.pdf.Gauss(obs=obs1, mu=1.3, sigma=5.4)

    yield1 = zfit.Parameter('yield123' + str(np.random.random()), 200.)

    # sum1 = 0.3 * gauss1 + gauss2
    gauss3_ext = 45. * gauss3
    gauss4_ext = 100. * gauss4
    sum2_ext_daughters = gauss3_ext + gauss4_ext
    sum3 = 0.4 * gauss5 + gauss6
    sum3_ext = sum3.create_extended(yield1)

    sum_all = zfit.pdf.SumPDF(pdfs=[sum2_ext_daughters, sum3_ext])
    sum_all.set_norm_range((-5, 5))

    extracted_pdfs = extract_extended_pdfs(pdfs=sum_all)
    assert frozenset(extracted_pdfs) == {gauss3_ext, gauss4_ext, sum3_ext}

    limits = zfit.Space(obs=obs1, limits=(-4, 5))
    limits = limits.with_autofill_axes()
    extended_sample = extended_sampling(pdfs=sum_all, limits=limits)
    extended_sample_np = zfit.run(extended_sample)
    assert np.shape(extended_sample_np)[0] == pytest.approx(expected=(45 + 100 + 200), rel=0.1)
    samples_from_pdf = sum_all.sample(n='extended', limits=limits)
    samples_from_pdf_np = zfit.run(samples_from_pdf)
    assert np.shape(samples_from_pdf_np)[0] == pytest.approx(expected=(45 + 100 + 200), rel=0.1)

max_stars_repo_name: olantwin/zfit

max_issues_repo_name: olantwin/zfit

max_forks_repo_name: olantwin/zfit


text:
import unittest
import phaseplot
from phaseplot import phase_portrait
import matplotlib
from matplotlib import pyplot as plt

class TestBasics(unittest.TestCase):
    """A collection of basic tests with no particular theme"""

    def test_retval(self):
        """phase_portrait returns an AxesImage instance"""
        def somefun(z): return z*z + 1
        retval = phase_portrait(somefun)
        self.assertIsInstance(retval, matplotlib.image.AxesImage)

    def test_extent(self):
        """Test that the 'box' argument matches extent"""

        # See also: issue #1

        ai = phase_portrait(lambda(z) : z, box = (-1,1,-1,1))
        # extent = (left, right, bottom, top)
        extent = ai.get_extent()
        self.assertEqual( extent, (-1, 1, -1, 1) )

        ai = phase_portrait(lambda(z) : z, box = (-1,2,-3,4))
        # extent = (left, right, bottom, top)
        extent = ai.get_extent()
        self.assertEqual( extent, (-1, 2, -3, 4) )

    def test_box_argument(self):
        """Test various ways to specify a plot box"""

        from numpy import sin

        # Method 1: Explicitly specify bounds for real(z) and imag(z)
        box = (-2,2,-2,2)
        ai = phase_portrait(sin, box=box)
        self.assertEqual(ai.get_extent(), box)

        # Method 2: Midpoint, re-part-radius, im-part-radius (infinity norm)
        box = (0, 3, 2)
        fullbox = (-3, 3, -2, 2)
        ai = phase_portrait(sin, box=box)
        self.assertEqual(ai.get_extent(), fullbox)

        # Method 3: lower-left corner & upper-right corner
        box = (-1-1j, 1+1j)
        fullbox = (-1,1,-1,1)
        ai = phase_portrait(sin, box=box)
        self.assertEqual(ai.get_extent(), fullbox)

        # Method 4: midpoint & real radius (infinity norm)
        box = (2 + 2j, 2)
        fullbox = (0, 4, 0, 4)
        ai = phase_portrait(sin, box=box)
        self.assertEqual(ai.get_extent(), fullbox)

        # Method 5: midpoint only (assumes radius == 1.0)
        box = (-1-1j)
        fullbox = (-2, 0, -2, 0)
        ai = phase_portrait(sin, box=box)
        self.assertEqual(ai.get_extent(), fullbox)

    def test_box_normalization(self):
        """Test that input is converted to float"""
        from phaseplot.plotfuns import interpret_box
        result = interpret_box( (1,2,3,4) )
        for item in result:
            self.assertIsInstance(item, float)

max_stars_repo_name: rluce/python-phaseplot

max_issues_repo_name: rluce/python-phaseplot

max_forks_repo_name: rluce/python-phaseplot

reason: from numpy


text:
import logging

import numpy as np

from braindecode.datautil.iterators import get_balanced_batches
from braindecode.util import wrap_reshape_apply_fn, corr

log = logging.getLogger(__name__)

def phase_perturbation(amps, phases, rng=None):
    """Takes amps and phases of BxCxF with B input, C channels, F frequencies
    Shifts spectral phases randomly U(-pi,pi) for input and frequencies, but same for all channels

    Parameters
    ----------
    amps : numpy array
        Spectral amplitude (not used)
    phases : numpy array
        Spectral phases
    rng : object
        Random Generator

    Returns
    -------
    amps : numpy array
        Input amps (not modified)
    phases_pert : numpy array
        Shifted phases
    pert_vals : numpy array
        Absolute phase shifts
    """
    if rng is None:
        rng = np.random.RandomState()
    noise_shape = list(phases.shape)
    noise_shape[1] = 1  # Do not sample noise for channels individually

    # Sample phase perturbation noise
    phase_noise = rng.uniform(-np.pi, np.pi, noise_shape).astype(np.float32)
    phase_noise_rep = phase_noise.repeat(phases.shape[1], axis=1)
    # Apply noise to inputs
    phases_pert = phases + phase_noise_rep
    phases_pert[phases_pert < -np.pi] += 2 * np.pi
    phases_pert[phases_pert > np.pi] -= 2 * np.pi

    pert_vals = np.abs(phase_noise)
    return amps, phases_pert, pert_vals

def amp_perturbation_additive(amps, phases, rng=None):
    """Takes amplitudes and phases of BxCxF with B input, C channels, F frequencies
    Adds additive noise N(0,0.02) to amplitudes

    Parameters
    ----------
    amps : numpy array
        Spectral amplitude
    phases : numpy array
        Spectral phases (not used)
    rng : object
        Random Seed

    Returns
    -------
    amps_pert : numpy array
        Scaled amplitudes
    phases_pert : numpy array
        Input phases (not modified)
    pert_vals : numpy array
        Amplitude noise
    """
    if rng is None:
        rng = np.random.RandomState()
    amp_noise = rng.normal(0, 1, amps.shape).astype(np.float32)
    amps_pert = amps + amp_noise
    amps_pert[amps_pert < 0] = 0
    amp_noise = amps_pert - amps
    return amps_pert, phases, amp_noise

def amp_perturbation_multiplicative(amps, phases, rng=None):
    """Takes amplitude and phases of BxCxF with B input, C channels, F frequencies
    Adds multiplicative noise N(1,0.02) to amplitudes

    Parameters
    ----------
    amps : numpy array
        Spectral amplitude
    phases : numpy array
        Spectral phases (not used)
    rng : object
        Random Seed

    Returns
    -------
    amps_pert : numpy array
        Scaled amplitudes
    phases_pert : numpy array
        Input phases (not modified)
    pert_vals : numpy array
        Amplitude scaling factor
    """
    if rng is None:
        rng = np.random.RandomState()
    amp_noise = rng.normal(1, 0.02, amps.shape).astype(np.float32)
    amps_pert = amps * amp_noise
    amps_pert[amps_pert < 0] = 0
    return amps_pert, phases, amp_noise

def correlate_feature_maps(x, y):
    """Takes two activation matrices of the form Bx[F]xT where B is batch size, F number of filters (optional) and T time points
    Returns correlations of the corresponding activations over T

    Parameters
    ----------
    x : numpy array
        Activations Bx[F]xT
    y : numpy array
        Activations Bx[F]xT

    Returns
    correlations : numpy array
        Correlations of `x` and `y` Bx[F]
    """
    shape_x = x.shape
    shape_y = y.shape
    assert np.array_equal(shape_x, shape_y)
    assert len(shape_x) < 4
    x = x.reshape((-1, shape_x[-1]))
    y = y.reshape((-1, shape_y[-1]))

    x = (x - x.mean(axis=1, keepdims=True)) / x.std(axis=1, keepdims=True)
    y = (y - y.mean(axis=1, keepdims=True)) / y.std(axis=1, keepdims=True)

    tmp_corr = x * y
    corr_ = tmp_corr.sum(axis=1)
    # corr_ = np.zeros((x.shape[0]))
    # for i in range(x.shape[0]):
    #    # Correlation of standardized variables
    #    corr_[i] = np.correlate((x[i]-x[i].mean())/x[i].std(),(y[i]-y[i].mean())/y[i].std())

    correlations = corr_.reshape(*shape_x[:-1])
    return correlations

def mean_diff_feature_maps(x, y):
    """Takes two activation matrices of the form BxFxT where B is batch size, F number of filters and T time points
    Returns mean difference between feature map activations

    Parameters
    ----------
    x : numpy array
        Activations Bx[F]xT
    y : numpy array
        Activations Bx[F]xT

    Returns
    mean_diff : numpy array
        Mean difference between `x` and `y` Bx[F]
    """
    mean_diff = np.mean(x - y, axis=2)
    return mean_diff

def spectral_perturbation_correlation(
    pert_fn,
    diff_fn,
    pred_fn,
    n_layers,
    inputs,
    n_iterations,
    batch_size=30,
    seed=((2017, 7, 10)),
):
    """Calculates perturbation correlations for layers in network by perturbing either amplitudes or phases

    Parameters
    ----------
    pert_fn : function
        Function that perturbs spectral phase and amplitudes of inputs
    diff_fn : function
        Function that calculates difference between original and perturbed activations
    pred_fn : function
        Function that returns a list of activations.
        Each entry_bak in the list corresponds to the output of 1 layer in a network
    n_layers : int
        Number of layers pred_fn returns activations for.
    inputs : numpy array
        Original inputs that are used for perturbation [B,X,T,1]
        Phase perturbations are sampled for each input individually, but applied to all X of that input
    n_iterations : int
        Number of iterations of correlation computation. The higher the better
    batch_size : int
        Number of inputs that are used for one forward pass. (Concatenated for all inputs)

    Returns
    -------
    pert_corrs : numpy array
        List of length n_layers containing average perturbation correlations over iterations
        L  x  CxFrxFi (Channels,Frequencies,Filters)
    """
    rng = np.random.RandomState(seed)

    # Get batch indeces
    batch_inds = get_balanced_batches(
        n_trials=len(inputs), rng=rng, shuffle=False, batch_size=batch_size
    )
    # Calculate layer activations and reshape
    log.info("Compute original predictions...")
    orig_preds = [pred_fn(inputs[inds]) for inds in batch_inds]
    use_shape = []
    for l in range(n_layers):
        tmp = list(orig_preds[0][l].shape)
        tmp.extend([1] * (4 - len(tmp)))
        tmp[0] = len(inputs)
        use_shape.append(tmp)
    orig_preds_layers = [
        np.concatenate(
            [orig_preds[o][l] for o in range(len(orig_preds))]
        ).reshape(use_shape[l])
        for l in range(n_layers)
    ]

    # Compute FFT of inputs
    fft_input = np.fft.rfft(inputs, n=inputs.shape[2], axis=2)
    amps = np.abs(fft_input)
    phases = np.angle(fft_input)

    pert_corrs = [0] * n_layers
    for i in range(n_iterations):
        log.info("Iteration {:d}...".format(i))
        log.info("Sample perturbation...")
        amps_pert, phases_pert, pert_vals = pert_fn(amps, phases, rng=rng)

        # Compute perturbed inputs
        log.info("Compute perturbed complex inputs...")
        fft_pert = amps_pert * np.exp(1j * phases_pert)
        log.info("Compute perturbed real inputs...")
        inputs_pert = np.fft.irfft(fft_pert, n=inputs.shape[2], axis=2).astype(
            np.float32
        )

        # Calculate layer activations for perturbed inputs
        log.info("Compute new predictions...")
        new_preds = [pred_fn(inputs_pert[inds]) for inds in batch_inds]
        new_preds_layers = [
            np.concatenate(
                [new_preds[o][l] for o in range(len(new_preds))]
            ).reshape(use_shape[l])
            for l in range(n_layers)
        ]

        for l in range(n_layers):
            log.info("Layer {:d}...".format(l))
            # Calculate difference of original and perturbed feature map activations
            log.info("Compute activation difference...")
            preds_diff = diff_fn(
                new_preds_layers[l][:, :, :, 0],
                orig_preds_layers[l][:, :, :, 0],
            )

            # Calculate feature map differences with perturbations
            log.info("Compute correlation...")
            pert_corrs_tmp = wrap_reshape_apply_fn(
                corr, pert_vals[:, :, :, 0], preds_diff, axis_a=(0,), axis_b=(0)
            )
            pert_corrs[l] += pert_corrs_tmp

    pert_corrs = [
        pert_corrs[l] / n_iterations for l in range(n_layers)
    ]  # mean over iterations
    return pert_corrs

def compute_amplitude_prediction_correlations(
    pred_fn,
    examples,
    n_iterations,
    perturb_fn=amp_perturbation_additive,
    batch_size=30,
    seed=((2017, 7, 10)),
):
    """
    Perturb input amplitudes and compute correlation between amplitude
    perturbations and prediction changes when pushing perturbed input through
    the prediction function.

    For more details, see [EEGDeepLearning]_.

    Parameters
    ----------
    pred_fn: function
        Function accepting an numpy input and returning prediction.
    examples: ndarray
        Numpy examples, first axis should be example axis.
    n_iterations: int
        Number of iterations to compute.
    perturb_fn: function, optional
        Function accepting amplitude array and random generator and returning
        perturbation. Default is Gaussian perturbation.
    batch_size: int, optional
        Batch size for computing predictions.
    seed: int, optional
        Random generator seed

    Returns
    -------
    amplitude_pred_corrs: ndarray
        Correlations between amplitude perturbations and prediction changes
        for all sensors and frequency bins.

    References
    ----------

    .. [EEGDeepLearning] Schirrmeister, R. T., Springenberg, J. T., Fiederer, L. D. J.,
       Glasstetter, M., Eggensperger, K., Tangermann, M., Hutter, F. & Ball, T. (2017).
       Deep learning with convolutional neural networks for EEG decoding and
       visualization.
       Human Brain Mapping , Aug. 2017. Online: http://dx.doi.org/10.1002/hbm.23730
    """
    pred_fn_new = lambda x: [pred_fn(x)]
    pred_corrs = spectral_perturbation_correlation(
        perturb_fn,
        mean_diff_feature_maps,
        pred_fn_new,
        1,
        examples,
        n_iterations,
        batch_size=batch_size,
        seed=seed,
    )

    return pred_corrs[0]


text:
import gc
import random

import torch
import torch.backends.cudnn
from torch.utils.data import DataLoader
from tqdm import tqdm
import numpy as np

import constants
from constants import DUMP_DIR, DEVICE, TEST_BATCH_SIZE, TOT_IMG_NUM, TEST_SUB_BATCH_SIZE, TEST_DATA_LOAD_WORKERS
from datasets.eval_dataset import EvalDatasetGraphDGL
from eval import eval
from knowledge_embed import KnowledgeData
import models
from raw_data_fix import RawData
from utils import collate_fn_eval

QueryEncoder = {
    'simple': models.QueryEncoder,
    'expand': models.QueryEncoderExpand,
    'weighted': models.QueryEncoderExpandWeighted,
    'ex_vgg': models.QueryEncoderExpandVGG
}[constants.QUERY_TYPE]

KnowledgeEncoder = {
    'bi': models.KnowledgeEncoderBidirectional,
    'bi_g': models.KnowledgeEncoderBidirectionalGate,
    'bi_g_wo_img': models.KnowledgeEncoderBidirectionalGateWoImg,
    'bi_g_wo_que': models.KnowledgeEncoderBidirectionalGateWoQuery,
    'bi_g_vgg': models.KnowledgeEncoderBidirectionalGateVGG
}[constants.KNOWLEDGE_TYPE]

print(f'Query encoder type: {QueryEncoder}')
print(f'Knowledge encoder type: {KnowledgeEncoder}')

def set_bn_eval(m):
    classname = m.__class__.__name__
    if classname.find('BatchNorm') != -1:
        m.eval()

class Evaluator:
    def __init__(self):
        random.seed(constants.SEED)
        np.random.seed(constants.SEED)
        torch.manual_seed(constants.SEED)
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False
        self.raw_data = RawData()
        self.knowledge_data = KnowledgeData(self.raw_data)
        self.query_encoder = QueryEncoder(self.raw_data).to(DEVICE)
        self.knowledge_encoder = KnowledgeEncoder(self.knowledge_data).to(DEVICE)

        self.test_dataset = EvalDatasetGraphDGL(self.raw_data, self.knowledge_data, 'test')
        # self.test_dataset = EvalDatasetGraphDGL(self.raw_data, self.knowledge_data, 'valid')
        self.test_data_loader = DataLoader(self.test_dataset, batch_size=TEST_BATCH_SIZE,
                                           shuffle=False, collate_fn=collate_fn_eval,
                                           num_workers=TEST_DATA_LOAD_WORKERS)

    def fold_tensor(self, x):
        return x.reshape(x.size(0) * x.size(1), *x.size()[2:])

    def eval(self):

        # model_file = DUMP_DIR / 'check_points.tar_kd2.0_att_only'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.5'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.7'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.12'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.12_all'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.12_all_dis_att'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_att_only1.12_all_dis_cel'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.1'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.3'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-dis-cel'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-att-only'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-dis-att'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-image-only'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-dis-sty'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_full_reverse1.2-image-only-rerun'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_reverse_fuse1.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_reverse_sep1.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_reverse_act1.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bidirectional1.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bidirectional1.0_dis_sty'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bidirectional1.0_att_only'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bidirectional1.0_dis_cel'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.0_dis_sty'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.1'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_img_only'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_dis_sty'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_dis_cel'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_dis_att'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bidirectional2.0'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_wo_img'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_wo_que'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_4gat'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_3gat'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_6gat'
        # model_file = DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_vgg'
        model_file = DUMP_DIR / 'check_points.tar_q_weighted_dgl_bid_gate1.0'
        print(f'Load model from {model_file}')
        state = torch.load(model_file)
        self.query_encoder.load_state_dict(state['query_encoder'])
        self.knowledge_encoder.load_state_dict(state['knowledge_encoder'])
        self.query_encoder.eval()
        self.knowledge_encoder.eval()
        self.query_encoder.apply(set_bn_eval)
        self.knowledge_encoder.apply(set_bn_eval)
        # log_writer = open(DUMP_DIR / 'check_points_kd2.0_att_only_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only1.7_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only1.12_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only1.12_all_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only1.12_all_dis_att_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_att_only1.12_all_dis_cel_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.1_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.3_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_dis_cel_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_att_only_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_dis_att_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_image_only_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_dis_sty_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_image_only_rerun_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_reverse_fuse1.0_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_reverse_sep1.0_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_reverse_act1.0_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_val_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_full_reverse1.2_image_only_rerun_val_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bidirectional1.0_r_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bidirectional1.0_dis_sty_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bidirectional1.0_att_only_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bidirectional1.0_dis_cel_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.0_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.0_dis_sty_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.1_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_img_only_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_dis_sty_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_dis_cel_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_dis_att_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_rm_att_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid2.0_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_wo_img_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_wo_que_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_4gat_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_3gat_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points_dgl_bid_gate1.2_6gat_full_neg_img1000.log', 'w')
        # log_writer = open(DUMP_DIR / 'check_points.tar_dgl_bid_gate1.2_vgg_full_neg_img1000.log', 'w')
        log_writer = open(DUMP_DIR / 'check_points.tar_q_weighted_dgl_bid_gate1.0_full_neg_img_1000.log', 'w')
        print(f'Load model from {model_file}', file=log_writer, flush=True)

        with torch.no_grad():
            p_5, p_10, p_20, recall_5, recall_10, recall_20, ndcg_5, ndcg_10, ndcg_20 = 0, 0, 0, 0, 0, 0, 0, 0, 0

            print('start')

            for batch_id, data in enumerate(tqdm(self.test_data_loader)):

                # print('batch data loaded')
                graph_inputs, num_pos_products, products = data

                session_embeddings = self.query_encoder(*graph_inputs)

                all_scores = []
                _images, _style_tips, _celebrities, _attributes = products
                batch_size = _images.size(0)

                for index in range(0, TOT_IMG_NUM, TEST_SUB_BATCH_SIZE):
                    # print('start sub batch data copy')
                    images = _images[:, index:(index + TEST_SUB_BATCH_SIZE)].to(DEVICE)
                    style_tips = _style_tips[:, index:(index + TEST_SUB_BATCH_SIZE)].to(DEVICE)
                    celebrities = _celebrities[:, index:(index + TEST_SUB_BATCH_SIZE)].to(DEVICE)
                    attributes = _attributes[:, index:(index + TEST_SUB_BATCH_SIZE)].to(DEVICE)

                    _image_num = images.size(1)
                    images = self.fold_tensor(images)
                    style_tips = self.fold_tensor(style_tips)
                    celebrities = self.fold_tensor(celebrities)
                    attributes = self.fold_tensor(attributes)
                    _session_embeddings = self.fold_tensor(session_embeddings.unsqueeze(1).expand(-1, _image_num, -1))

                    desired_images = self.knowledge_encoder(_session_embeddings,
                                                            images, style_tips, celebrities, attributes)
                    if isinstance(desired_images, tuple):
                        images_scores = torch.cosine_similarity(_session_embeddings, desired_images[0])
                        knowledge_scores = torch.cosine_similarity(_session_embeddings, desired_images[1])
                        scores = constants.MIX_SCALAR * images_scores + (1 - constants.MIX_SCALAR) * knowledge_scores
                    else:
                        scores = torch.cosine_similarity(_session_embeddings, desired_images)

                    all_scores.append(scores.detach().reshape(batch_size, _image_num))

                    del images
                    del style_tips
                    del celebrities
                    del attributes
                    del desired_images
                    del scores

                all_scores = torch.cat(all_scores, dim=1).cpu()

                del _images
                del _style_tips
                del _celebrities
                del _attributes
                del products

                # Remove pad
                num_pos_products, num_padding = zip(*num_pos_products)
                assert len(num_pos_products) == len(num_padding) == batch_size
                for _b, _pad_len in enumerate(num_padding):
                    all_scores[_b, -_pad_len:] = torch.zeros(_pad_len).fill_(-10000.0)

                _p_5, _p_10, _p_20, _recall_5, _recall_10, _recall_20, _ndcg_5, _ndcg_10, _ndcg_20 = eval(all_scores, [
                    list(range(num)) for num in num_pos_products])

                del all_scores

                N = batch_id + 1
                p_5 += _p_5
                p_10 += _p_10
                p_20 += _p_20
                recall_5 += _recall_5
                recall_10 += _recall_10
                recall_20 += _recall_20
                ndcg_5 += _ndcg_5
                ndcg_10 += _ndcg_10
                ndcg_20 += _ndcg_20

                if batch_id % 10 == 0:
                    gc.collect()
                    print('--p_5:', str(p_5 / N), '--p_10:', str(p_10 / N), '--p_20:', str(p_20 / N),
                          '--recall_5:', str(recall_5 / N), '--recall_10:', str(recall_10 / N), '--recall_20:',
                          str(recall_20 / N),
                          '--ndcg_5:', str(ndcg_5 / N), '--ndcg_10:', str(ndcg_10 / N), '--ndcg_20:', str(ndcg_20 / N))

                    print(f'N = {N}', file=log_writer, flush=True)
                    print('--p_5:', str(p_5 / N), '--p_10:', str(p_10 / N), '--p_20:', str(p_20 / N),
                          '--recall_5:', str(recall_5 / N), '--recall_10:', str(recall_10 / N), '--recall_20:',
                          str(recall_20 / N),
                          '--ndcg_5:', str(ndcg_5 / N), '--ndcg_10:', str(ndcg_10 / N), '--ndcg_20:', str(ndcg_20 / N),
                          file=log_writer, flush=True)

            print('================== End ===========================')
            print('--p_5:', str(p_5 / N), '--p_10:', str(p_10 / N), '--p_20:', str(p_20 / N),
                  '--recall_5:', str(recall_5 / N), '--recall_10:', str(recall_10 / N), '--recall_20:',
                  str(recall_20 / N),
                  '--ndcg_5:', str(ndcg_5 / N), '--ndcg_10:', str(ndcg_10 / N), '--ndcg_20:', str(ndcg_20 / N))

            print('================== End ===========================', file=log_writer, flush=True)
            print('--p_5:', str(p_5 / N), '--p_10:', str(p_10 / N), '--p_20:', str(p_20 / N),
                  '--recall_5:', str(recall_5 / N), '--recall_10:', str(recall_10 / N), '--recall_20:',
                  str(recall_20 / N),
                  '--ndcg_5:', str(ndcg_5 / N), '--ndcg_10:', str(ndcg_10 / N), '--ndcg_20:', str(ndcg_20 / N),
                  file=log_writer, flush=True)

max_stars_repo_name: SparkJiao/LARCH

max_issues_repo_name: SparkJiao/LARCH

max_forks_repo_name: SparkJiao/LARCH


text:
from PIL import Image
from collections import defaultdict
import math
import cv2
import multiprocessing
import socket
import numpy as np

def blockproc(block, num):
    for hor in range(8):
        for ver in range(8):
            frame[hor, ver] = 255,255,255
            # print num, block[0][0]

def blockelate(frame):
    # print frame[0][0]
    piycc = {}
    mulp = []
    # frame[100:116][100:116]
    print frame[0][0]
    piycc = {}

    for pixy in range(2):
        piycc[pixy] = {}
        for pixx in range(2):
            piycc[pixy][pixx] = {}
            for py8 in range(8):
                piycc[pixy][pixx][py8] = {}
                for px8 in range(8):
                    piycc[pixy][pixx][py8][px8] = frame[(pixx*8) + 100][(pixy*8) + 100]
    mulproc1 = multiprocessing.Process(target=blockproc(piycc[0][0], 1))
    mulproc2 = multiprocessing.Process(target=blockproc(piycc[0][1], 2))
    mulproc3 = multiprocessing.Process(target=blockproc(piycc[1][0], 3))
    mulproc4 = multiprocessing.Process(target=blockproc(piycc[1][1], 4))
    mulproc1.start()
    mulproc2.start()
    mulproc3.start()
    mulproc4.start()
    for blocky in range(4):
        piycc[blocky] = {}
        for blockx in range(4):
           piycc[blocky][blockx] = {}
           mulproc = multiprocessing.Process(target=blockproc(blocky, blockx))
           mulp.append(mulproc)
           mulproc.start()
           piycc[blocky][blockx] = blockproc(blocky, blockx)

           # for py8 in range(8):
           #     piycc[pixy][pixx][py8] = {}
           #     for px8 in range(8):
           #         piycc[pixy][pixx][py8][px8] = {}
                   # frame[pixy][pixx] = 255, 255, 255
                   # print 'a'
#                    piycc[pixy][pixx][py8][px8] = toYCbCr(frame[(pixx * 8) + px8][(pixy * 8) + py8])
#             dct88y[pixy][pixx] = dct(piycc[pixy][pixx])
#             print piycc[pixy][pixx]
#            if pixx % 4 == 0 and pixx != 0:
#                ymail = blockenc(dct88y[pixy][pixx - 4], dct88y[pixy][pixx - 3], dct88y[pixy][pixx - 2], dct88y[pixy][pixx - 1])
#                 cbmail = blockenc(dct88cb[pixx][pixy], dct88cb[pixx + 1][pixy], dct88cb[pixx + 2][pixy], dct88cb[pixx + 3][pixy])
#                 crmail = blockenc(dct88cr[pixx][pixy], dct88cr[pixx + 1][pixy], dct88cr[pixx + 2][pixy], dct88cr[pixx + 3][pixy])
#                 print ymail
                #print "========================444444444========================"
            #     print cbmail
            #     print crmail

def send(mail):
    print mail

def recv(mail):
    print mail

def blockenc(block0, block1, block2, block3):
    dc = str(block0[0][0]) + 'z' + str(block1[0][0]) + 'z' + str(block2[0][0]) + 'z' + str(block3[0][0]) + 'z'
    #dcenc = denc(dc)
    #print dcenc
    ac4, encac4 = defaultdict(dict), defaultdict(dict)
    ac4[0] = str(quan(block0))
    ac4[1] = str(quan(block1))
    ac4[2] = str(quan(block2))
    ac4[3] = str(quan(block3))
    #for y in range(4):
    #    for x in range(3):
    #        encac4[y][x] = aenc(ac4[y][x])
    #mail = str(dcenc) + str(encac4)
    ac4 = ac4[0] + ac4[1] + ac4[2] + ac4[3]
    mail = str(dc) + str(ac4)
    # print mail
    return mail

def quan(block):
    ac = defaultdict(dict)
    ac[0] = (str(block[1][0]) + 'x' + str(block[0][1]) + 'x' + str(block[0][2]) + 'x' + str(block[1][1])) + 'y'
    ac[1] = (str(block[2][0]) + 'x' + str(block[3][0]) + 'x' + str(block[2][1]) + 'x' + str(block[1][2])) + 'y'
    ac[2] = (str(block[0][3]) + 'x' + str(block[0][4]) + 'x' + str(block[1][3]) + 'x' + str(block[2][2])) + 'yy'
    ac = ac[0] + ac[1] + ac[2]
    #print ac
    return ac

def toYCbCr((r, g, b)):
    y = (0.299 * r) + (0.587 * g) + (0.114 * b)
    cb = -(0.1687 * r) - (0.3313 * g) + (0.5 * b) + 128
    cr = (0.5 * r) - (0.4187 * g) - (0.0813 * b) + 128
    #print y, cb, cr
    return (y, cb, cr)

def toRGB(y, cb, cr):
    r = y + (1.402 * (cr - 128))
    g = y - (0.34414 * (cb - 128)) - (0.71414 * (cr-128))
    b = y + (1.772 * (cb - 128))
    print r, g, b
    return r, g, b

def dct(block):
    dcty = defaultdict(dict)
    for dy8 in range(8):
        for dx8 in range(8):
            sumc = 0
            for y8 in range(8):
                for x8 in range(8):
                    sumct = block[x8][y8] * math.cos(((2 * x8 + 1) * dx8 * math.pi) / 16) * math.cos(((2 * y8 + 1) * dy8 * math.pi) / 16)
                    #print "block: " + str(block[x8][y8])
                    #print "cosx: " + str(math.cos(((2 * x8 + 1) * dx8 * math.pi) / 16))
                    #print "cosy: " + str(math.cos(((2 * y8 + 1) * dy8 * math.pi) / 16))
                    #print "temp: " + str(sumct)
                    sumc = sumc + sumct
                    #print "sumtemp: " + str(sumc)
            i, j = 1, 1
            if dx8 == 0:
                i = (2 ** -0.5)
            if dy8 == 0:
                j = (2 ** -0.5)
            #print i, j
            dcty[dx8][dy8] = 0.25 * i * j * sumc
            #print "dct: " + str(dcty[dx8][dy8])
            #print "round: " + str(dcty[dx8][dy8] % 1)
            if dcty[dx8][dy8] > 0:
                if dcty[dx8][dy8] % 1 > 0.5 and dcty[dx8][dy8] % 1 < 1:
                    dcty[dx8][dy8] = int(dcty[dx8][dy8]) + 1
                if dcty[dx8][dy8] % 1 <= 0.5 and dcty[dx8][dy8] % 1 > 0:
                    dcty[dx8][dy8] = int(dcty[dx8][dy8])
            elif dcty[dx8][dy8] < 0:
                if dcty[dx8][dy8] % -1 >= -0.5 and dcty[dx8][dy8] % -1 < 0:
                    dcty[dx8][dy8] = int(dcty[dx8][dy8])
                if dcty[dx8][dy8] % -1 < -0.5 and dcty[dx8][dy8] % -1 > -1:
                    dcty[dx8][dy8] = int(dcty[dx8][dy8]) - 1
            dcty[dx8][dy8] = int(dcty[dx8][dy8])
            #print str(dx8), str(dy8), str(dcty[dx8][dy8])
    return dcty
    #idct(dcty)

def idct(block):
    dcta = defaultdict(dict)
    for y8 in range(8):
        for x8 in range(8):
            sumc = 0
            for dy8 in range(8):
                for dx8 in range(8):
                    i, j = 1, 1
                    if dx8 == 0:
                        i = (2 ** -0.5)
                    if dy8 == 0:
                        j = (2 ** -0.5)
                    sumct = i * j * block[dx8][dy8] * math.cos(((2 * x8 + 1) * dx8 * math.pi) / 16) * math.cos(((2 * y8 + 1) * dy8 * math.pi) / 16)
                    #print "block: " + str(block[x8][y8])
                    #print "cosx: " + str(math.cos(((2 * x8 + 1) * dx8 * 180) / 16))
                    #print "cosy: " + str(math.cos(((2 * y8 + 1) * dy8 * 180) / 16))
                    #print "temp: " + str(sumct)
                    sumc = sumc + sumct
                    #print "sumtemp: " + str(sumc)
            dcta[x8][y8] = 0.25 * sumc
            dcta[x8][y8] = int(dcta[x8][y8])
            print str(x8), str(y8), str(dcta[x8][y8])
            #print "============================================================="
    return dcta

def preproc():
    print "stream"

def rc5():
    print "stream"

def send():
    print "stream"

def enc():
    print "stream"

def dec():
    print "stream"

cap = cv2.VideoCapture(0)
cuk = defaultdict(dict)
cap.set(3, 320)
cap.set(4, 400)
#cap.set(3, 1600)
#cap.set(4, 900)
cap.set(11, 45)

while(True):
    ret, frame = cap.read()
    blockelate(frame)
    cv2.imshow('frame', frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

max_stars_repo_name: numburanggata/sloth

max_issues_repo_name: numburanggata/sloth

max_forks_repo_name: numburanggata/sloth


text:
from __future__ import print_function

import inspect
import os
# this is written to retrive airnow data concatenate and add to pandas array for usage
from builtins import object
from datetime import datetime, timedelta

import pandas as pd
from numpy import array

class AirNow(object):
    def __init__(self):
        self.datadir = '.'
        self.cwd = os.getcwd()
        self.url = None
        self.dates = [datetime.strptime('2016-06-06 12:00:00', '%Y-%m-%d %H:%M:%S'),
                      datetime.strptime('2016-06-06 13:00:00', '%Y-%m-%d %H:%M:%S')]
        self.datestr = []
        self.df = None
        self.se_states = array(
            ['AL', 'FL', 'GA', 'MS', 'NC', 'SC', 'TN',
             'VA', 'WV'], dtype='|S14')
        self.ne_states = array(['CT', 'DE', 'DC', 'ME', 'MD', 'MA',
                                'NH', 'NJ', 'NY', 'PA', 'RI', 'VT'],
                               dtype='|S20')
        self.nc_states = array(
            ['IL', 'IN', 'IA', 'KY', 'MI', 'MN', 'MO', 'OH', 'WI'],
            dtype='|S9')
        self.sc_states = array(['AR', 'LA', 'OK', 'TX'], dtype='|S9')
        self.r_states = array(['AZ', 'CO', 'ID', 'KS', 'MT', 'NE', 'NV', 'NM',
                               'ND', 'SD', 'UT', 'WY'], dtype='|S12')
        self.p_states = array(['CA', 'OR', 'WA'], dtype='|S10')
        self.objtype = 'AirNow'
        self.filelist = None
        self.monitor_file = inspect.getfile(
            self.__class__)[:-16] + '/data/monitoring_site_locations.dat'
        self.monitor_df = None
        self.savecols = ['datetime', 'SCS', 'Site', 'utcoffset', 'Species', 'Units', 'Obs', 'datetime_local',
                         'Site_Name', 'Latitude', 'Longitude', 'CMSA_Name', 'MSA_Code', 'MSA_Name', 'State_Name',
                         'EPA_region']

    def convert_dates_tofnames(self):
        self.datestr = []
        for i in self.dates:
            self.datestr.append(i.strftime('%Y%m%d%H.dat'))

    def change_path(self):
        os.chdir(self.datadir)

    def change_back(self):
        os.chdir(self.cwd)

    def build_urls(self):
        from numpy import empty, where
        import wget
        import requests
        from glob import glob

        furls = []

        print('Retrieving AIRNOW files...')
        # 2017/20170131/HourlyData_2017012408.dat
        url = 'https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/'
        for i in self.dates:
            f = url + i.strftime('%Y/%Y%m%d/HourlyData_%Y%m%d%H.dat')
            furls.append(f)

        # files needed for comparison
        self.url = pd.Series(furls, index=None)

    def read_csv(self, fn):
        try:
            dft = pd.read_csv(fn, delimiter='|', header=None,
                              error_bad_lines=False)
            cols = ['date', 'time', 'SCS', 'Site', 'utcoffset',
                    'Species', 'Units', 'Obs', 'Source']
            dft.columns = cols
        except:
            cols = ['date', 'time', 'SCS', 'Site', 'utcoffset',
                    'Species', 'Units', 'Obs', 'Source']
            dft = pd.DataFrame(columns=cols)
        return dft

    def aggragate_files(self):
        import dask
        import dask.dataframe as dd

        print('Aggregating AIRNOW files...')
        self.build_urls()
        dfs = [dask.delayed(self.read_csv)(f) for f in self.url]
        dff = dd.from_delayed(dfs)
        df = dff.compute()
        df['datetime'] = pd.to_datetime(
            df.date + ' ' + df.time, format='%m/%d/%y %H:%M', exact=True, box=False)
        df.drop(['date', 'time'], axis=1, inplace=True)
        df['datetime_local'] = df.datetime + \
                               pd.to_timedelta(df.utcoffset, unit='H')
        self.df = df
        print('    Adding in Meta-data')
        self.get_station_locations()
        self.df = self.df[self.savecols]
        self.df.drop_duplicates(inplace=True)

    def calc_datetime(self):
        # takes in an array of string dates and converts to numpy array of datetime objects
        dt = []
        for i in self.df.utcoffset.values:
            dt.append(timedelta(hours=i))
        self.df['datetime_local'] = self.df.datetime + dt

    def set_daterange(self, begin='', end=''):
        dates = pd.date_range(start=begin, end=end,
                              freq='H').values.astype('M8[s]').astype('O')
        self.dates = dates

    def get_station_locations(self):
        self.read_monitor_file()
        self.df = pd.merge(self.df, self.monitor_df, on='SCS', how='left')

    def get_station_locations_remerge(self, df):
        df = pd.merge(df, self.monitor_df.drop(
            ['Latitude', 'Longitude'], axis=1), on='SCS', how='left')
        return df

    def read_monitor_file(self):
        try:
            print('    Monitor Station Meta-Data Found: Compiling Dataset')
            monitor_url = 'https://s3-us-west-1.amazonaws.com//files.airnowtech.org/airnow/today/monitoring_site_locations.dat'
            colsinuse = [0, 2, 3, 4, 5, 6, 7, 8, 9, 10,
                         11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21]
            f = pd.read_csv(monitor_url, delimiter='|',
                            header=None, usecols=colsinuse)
            f.columns = ['SCS', 'Site_Code', 'Site_Name', 'Status', 'Agency', 'Agency_Name', 'EPA_region', 'Latitude',
                         'Longitude', 'Elevation', 'GMT_Offset', 'Country_Code', 'CMSA_Code', 'CMSA_Name', 'MSA_Code',
                         'MSA_Name', 'State_Code', 'State_Name', 'County_Code', 'County_Name', 'City_Code']
        except:
            from glob import glob
            if os.path.isfile(self.monitor_file) == True:
                print('    Monitor Station Meta-Data Found: Compiling Dataset')
                fname = self.monitor_file
            else:
                self.openftp()
                self.ftp.cwd('Locations')
                self.download_single_rawfile(
                    fname='monitoring_site_locations.dat')
            fname = glob('monitoring_site_locations.dat')[0]
            colsinuse = [0, 2, 3, 4, 5, 6, 7, 8, 9, 10,
                         11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21]
            f = pd.read_csv(fname, delimiter='|',
                            header=None, usecols=colsinuse)
            f.columns = ['SCS', 'Site_Code', 'Site_Name', 'Status', 'Agency', 'Agency_Name', 'EPA_region', 'Latitude',
                         'Longitude', 'Elevation', 'GMT_Offset', 'Country_Code', 'CMSA_Code', 'CMSA_Name', 'MSA_Code',
                         'MSA_Name', 'State_Code', 'State_Name', 'County_Code', 'County_Name', 'City_Code']

        self.monitor_df = f.copy()

    def get_region(self):
        sr = self.df.State_Name.copy().values
        for i in self.se_states:
            con = sr == i
            sr[con] = 'Southeast'
        for i in self.ne_states:
            con = sr == i
            sr[con] = 'Northeast'
        for i in self.nc_states:
            con = sr == i
            sr[con] = 'North Central'
        for i in self.sc_states:
            con = sr == i
            sr[con] = 'South Central'
        for i in self.p_states:
            con = sr == i
            sr[con] = 'Pacific'
        for i in self.r_states:
            con = sr == i
            sr[con] = 'Rockies'
        sr[sr == 'CC'] = 'Canada'
        sr[sr == 'MX'] = 'Mexico'

        self.df['Region'] = array(sr)

max_stars_repo_name: barronh/MONET

max_issues_repo_name: barronh/MONET

max_forks_repo_name: barronh/MONET


text:
import timeit
import pickle
import numpy as np
import scipy.stats as st
import networkx as nx
import matplotlib.pyplot as plt

from simulation.helpers import update_progress, csv_export, create_random_graph_distances
from simulation.plotting import print_graph, print_tips_over_time, \
print_tips_over_time_multiple_agents, print_tips_over_time_multiple_agents_with_tangle, \
print_attachment_probabilities_alone, print_attachment_probabilities_all_agents
from simulation.simulation import Single_Agent_Simulation
from simulation.simulation_multi_agent import Multi_Agent_Simulation

#############################################################################
# SIMULATION: SINGLE AGENT
#############################################################################

#Parameters: no_of_transactions, lambda, no_of_agents, alpha, latency (h), tip_selection_algo
#Tip selection algorithms: Choose among "random", "weighted", "unweighted" as input

# simu = Single_Agent_Simulation(100, 50, 1, 0.005, 1, "weighted")
# simu.setup()
# simu.run()
# print_tips_over_time(simu)

#############################################################################
# SIMULATION: MULTI AGENT
#############################################################################

#Parameters: no_of_transactions, lambda, no_of_agents, alpha, distance, tip_selection_algo
# latency (default value 1), agent_choice (default vlaue uniform distribution, printing)
#Tip selection algorithms: Choose among "random", "weighted", "unweighted" as input

start_time = timeit.default_timer()
runs = 1

for i in range(runs):

    simu2 = Multi_Agent_Simulation(800, 50, 2, 0.1, 1, "weighted", _printing=True)
    simu2.setup()
    simu2.run()
    # csv_export(simu2)

print("TOTAL simulation time: " + str(np.round(timeit.default_timer() - start_time, 3)) + " seconds\n")

#############################################################################
# PLOTTING
#############################################################################

# print_graph(simu2)
# print_tips_over_time(simu2)
# print_tips_over_time_multiple_agents(simu2, simu2.no_of_transactions)
print_tips_over_time_multiple_agents_with_tangle(simu2, simu2.no_of_transactions)
# print_attachment_probabilities_all_agents(simu2)

reason: import numpy,import scipy,import networkx


text:
#!/usr/bin/env python3
import numpy as np
import math
import os
import sys

# Read a 2D text file from disk into a NumPy array
def read_file(fn):
    data = np.fromfile(fn, dtype=np.int, sep=" ")
    data = data[2:].reshape(tuple(data[0:2]))
    return data

# Save a block from a 2D NumPy array to disk
def save_piece(fn, array, offset, n_blocks, block_index, size):
    with open(fn, 'w') as f:
        f.write('#GLOBAL_EXTENTS ' + ' '.join(map(str, array.shape)) + '\n')
        f.write('#OFFSET ' + ' '.join(map(str, offset))+'\n')
        f.write('#BLOCKS_PER_DIM ' + ' '.join(map(str, n_blocks))+'\n')
        f.write('#BLOCK_INDEX ' + ' '.join(map(str, block_index))+'\n')
        f.write(' '.join(map(str, size)) + '\n')
        np.savetxt(f, array[offset[0]:offset[0]+size[0],offset[1]:offset[1]+size[1]], fmt='%.16g')

# Compute split points for splitting into n blocks
def split_points(shape, nblocks):
    dx = float(shape-1) / nblocks
    return [ math.floor(i*dx) for i in range(nblocks)] + [ shape - 1 ]

if len(sys.argv) < 2:
    print("Error: Usage split_data_2d.py <filename> [<n_blocks_per_axis>|<n_blocks_x> <n_blocks_y>]", file=sys.stderr)
    sys.exit(1)

# Parse parameters
in_filename = sys.argv[1]
n_blocks = (2, 2)
if len(sys.argv) > 2:
    if len(sys.argv) >= 4:
        n_blocks = (int(sys.argv[2]), int(sys.argv[3]))
    else:
        n_blocks = (int(sys.argv[2]), int(sys.argv[2]))

name, ext = os.path.splitext(in_filename)
out_filename_pattern = name + '_part_%d_of_' + str(n_blocks[0]*n_blocks[1]) + ext

# Read data
data = read_file(in_filename)

# Compute split points
split_points_x = split_points(data.shape[0], n_blocks[0])
split_points_y = split_points(data.shape[1], n_blocks[1])

# Save blocks
block_no = 0
for block_index_x, (x_start, x_stop) in enumerate(zip(split_points_x, split_points_x[1:])):
    for block_index_y, (y_start, y_stop) in enumerate(zip(split_points_y, split_points_y[1:])):
        n_x = x_stop - x_start + 1
        n_y = y_stop - y_start + 1
        save_piece(out_filename_pattern % block_no, data, (x_start, y_start), n_blocks, (block_index_x, block_index_y), (n_x, n_y))
#         print("Wrote block %d, origin %d %d, size %d %d" % (block_no, x_start, y_start, n_x, n_y))
        block_no += 1


text:
# -----------------------------------------------------------------------
# Copyright (c) 2020, NVIDIA Corporation. All rights reserved.
#
# This work is made available
# under the Nvidia Source Code License (1-way Commercial).
#
# Official Implementation of the CVPR2020 Paper
# Two-shot Spatially-varying BRDF and Shape Estimation
# Mark Boss, Varun Jampani, Kihwan Kim, Hendrik P. A. Lensch, Jan Kautz
# -----------------------------------------------------------------------

import numpy as np
import tensorflow as tf
from tensorpack.tfutils.summary import add_moving_summary

import utils.tb_utils as tbutil
from utils.common_layers import uncompressDepth

def l1_loss(gt: tf.Tensor, pred: tf.Tensor):
    return tf.abs(gt - pred)

def l2_loss(gt: tf.Tensor, pred: tf.Tensor):
    loss = gt - pred
    return loss * loss

def cosine_distance(l, p):
    ps = p
    ls = l

    lNrm = tf.nn.l2_normalize(ls, 1)
    pNrm = tf.nn.l2_normalize(ps, 1)
    return tf.reduce_sum(lNrm * pNrm, axis=-1, keepdims=True)

def angular_distance(l, p, negative_vals=False):
    cosAng = tf.clip_by_value(cosine_distance(l, p), -1.0 + EPS, 1.0 - EPS)
    mult = 2.0 if negative_vals else 1.0
    return (mult * tf.acos(cosAng)) / np.pi

def masked_loss(loss: tf.Tensor, mask: tf.Tensor):
    return tf.where(tf.less_equal(mask, 1e-5), tf.zeros_like(loss), loss)

def height_normal_consistency_loss(
    depth, normal, mask3, consistency_loss_factor: float, imgSize: int, max_outputs: int
):
    near = uncompressDepth(1)
    far = uncompressDepth(0)
    d = uncompressDepth(depth)
    h = tf.div_no_nan(d - near, far - near)

    sobel = tf.image.sobel_edges(h)  # b,h,w,1,[dy,dx] - 1 because height has 1 channel
    dx = sobel[:, :, :, :, 1]  # b,h,w,1
    dy = -sobel[:, :, :, :, 0]
    # We're using a depth map instead of a height. Which means bright
    # values are at a greater depth. Thus, we need to invert the gradient
    texelSize = 1 / imgSize
    dz = tf.ones_like(dx) * texelSize * 2

    cn = tf.nn.l2_normalize(tf.concat([dx, dy, dz], axis=-1), axis=-1)

    cn = cn * 0.5 + 0.5
    cl = masked_loss(l1_loss(cn, normal), mask3)

    clr = tf.reduce_mean(cl, name="consistency_loss") * consistency_loss_factor
    add_moving_summary(clr)
    tf.losses.add_loss(clr, tf.GraphKeys.LOSSES)

    repeat = [1 for _ in range(len(depth.shape))]
    repeat[-1] = 3
    tbutil.four_side_by_side(
        tf.tile(depth, repeat), cn, normal, cl, "consistency", max_outputs
    )

max_stars_repo_name: CSuppan/two-shot-brdf-shape

max_issues_repo_name: CSuppan/two-shot-brdf-shape

max_forks_repo_name: CSuppan/two-shot-brdf-shape


text:
# -*- mode: python; coding: utf-8 -*-
# Copyright 2017 the HERA Collaboration
# Licensed under the 2-clause BSD license.

import numpy as np
import numpy.testing as npt
import pytest

from astropy.time import Time
from astropy.units import Quantity

from .. import utils

def test_LSTScheduler_lstbinsize():
    """
    test that two bins have the right time separation
    """
    LSTbin_size = 10
    # pick a date far in the past just in case IERS is down
    starttime1 = Time('2015-9-19T05:05:05.0', format='isot', scale='utc')
    scheduletime1, hour1 = utils.LSTScheduler(starttime1, LSTbin_size)
    starttime2 = Time('2015-9-19T05:05:15.0', format='isot', scale='utc')
    scheduletime2, hour2 = utils.LSTScheduler(starttime2, LSTbin_size)
    assert np.isclose((hour2 - hour1).hour * 3600, LSTbin_size)
    # length of sidereal second in SI seconds.
    sidesec = Quantity(1, 'sday').to('day').value
    npt.assert_almost_equal((scheduletime2 - scheduletime1).value * 24 * 3600,
                            LSTbin_size * sidesec, decimal=5)

    with pytest.raises(
        TypeError,
        match="starttime is not a valid Astropy Time object"
    ):
        utils.LSTScheduler("foo", LSTbin_size)

def test_LSTScheduler_multiday():
    """
    test that two bins are at the same LST on different days
    """
    LSTbin_size = 10
    # pick a date far in the past just in case IERS is down
    starttime1 = Time('2015-9-19T05:04:09.0', format='isot', scale='utc')
    scheduletime1, hour1 = utils.LSTScheduler(starttime1, LSTbin_size)
    # lst is 4 minutes earlier every day
    starttime2 = Time('2015-9-20T05:00:09.0', format='isot', scale='utc')
    scheduletime2, hour2 = utils.LSTScheduler(starttime2, LSTbin_size)
    assert np.isclose((hour2.hour - hour1.hour) * 3600, 0)

def test_get_iterable():
    assert utils.get_iterable(2) == (2,)
    assert utils.get_iterable([2]) == [2]
    assert utils.get_iterable([2, 3]) == [2, 3]
    assert utils.get_iterable("foo") == ("foo",)
    assert utils.get_iterable(["foo"]) == ["foo"]
    assert utils.get_iterable(["foo", "bar"]) == ["foo", "bar"]

def test_get_obsid_from_file(tmp_path):
    """
    Test the get_obsid_from_file function; requires h5py.
    """
    h5py = pytest.importorskip("h5py")

    # make a fake time array
    jd0 = 2457000
    t0 = Time(jd0, format="jd", scale="utc")
    times = np.linspace(jd0, jd0 + 0.5, num=10)
    output_file = str(tmp_path / "test.uvh5")
    with h5py.File(output_file, "w") as h5f:
        header = h5f.create_group("Header")
        times = header.create_dataset("time_array", data=times)

    # test function
    obsid = utils.get_obsid_from_file(output_file)
    assert obsid == int(np.floor(t0.gps))

    return

reason: import numpy,from astropy