text:
import unittest

from flow.core.params import SumoParams, EnvParams, InitialConfig, \
    NetParams, SumoCarFollowingParams
from flow.core.vehicles import Vehicles

from flow.controllers.routing_controllers import ContinuousRouter
from flow.controllers.car_following_models import IDMController
from flow.envs.loop.loop_accel import ADDITIONAL_ENV_PARAMS

from tests.setup_scripts import ring_road_exp_setup
import os
import numpy as np

os.environ["TEST_FLAG"] = "True"

class TestStartingPositionShuffle(unittest.TestCase):
    """
    Tests that, at resets, the starting position of vehicles changes while
    keeping the ordering and relative spacing between vehicles.
    """

    def setUp(self):
        # turn on starting position shuffle
        env_params = EnvParams(starting_position_shuffle=True,
                               additional_params=ADDITIONAL_ENV_PARAMS)

        # place 5 vehicles in the network (we need at least more than 1)
        vehicles = Vehicles()
        vehicles.add(veh_id="test",
                     acceleration_controller=(IDMController, {}),
                     routing_controller=(ContinuousRouter, {}),
                     num_vehicles=5)

        initial_config = InitialConfig(x0=5)

        # create the environment and scenario classes for a ring road
        self.env, scenario = ring_road_exp_setup(env_params=env_params,
                                                 initial_config=initial_config,
                                                 vehicles=vehicles)

    def tearDown(self):
        # terminate the traci instance
        self.env.terminate()

        # free data used by the class
        self.env = None

    def runTest(self):
        ids = self.env.vehicles.get_ids()

        # position of vehicles before reset
        before_reset = \
            np.array([self.env.get_x_by_id(veh_id) for veh_id in ids])

        # reset the environment
        self.env.reset()

        # position of vehicles after reset
        after_reset = \
            np.array([self.env.get_x_by_id(veh_id) for veh_id in ids])

        offset = after_reset[0] - before_reset[0]

        # remove the new offset from the original positions after reset
        after_reset = np.mod(after_reset - offset, self.env.scenario.length)

        np.testing.assert_array_almost_equal(before_reset, after_reset)

class TestVehicleArrangementShuffle(unittest.TestCase):
    """
    Tests that, at resets, the ordering of vehicles changes while the starting
    position values stay the same.
    """

    def setUp(self):
        # turn on vehicle arrangement shuffle
        env_params = EnvParams(vehicle_arrangement_shuffle=True,
                               additional_params=ADDITIONAL_ENV_PARAMS)

        # place 5 vehicles in the network (we need at least more than 1)
        vehicles = Vehicles()
        vehicles.add(veh_id="test",
                     acceleration_controller=(IDMController, {}),
                     routing_controller=(ContinuousRouter, {}),
                     num_vehicles=5)

        initial_config = InitialConfig(x0=5)

        # create the environment and scenario classes for a ring road
        self.env, scenario = ring_road_exp_setup(env_params=env_params,
                                                 initial_config=initial_config,
                                                 vehicles=vehicles)

    def tearDown(self):
        # terminate the traci instance
        self.env.terminate()

        # free data used by the class
        self.env = None

    def runTest(self):
        ids = self.env.vehicles.get_ids()

        # position of vehicles before reset
        before_reset = [self.env.get_x_by_id(veh_id) for veh_id in ids]

        # reset the environment
        self.env.reset()

        # position of vehicles after reset
        after_reset = [self.env.get_x_by_id(veh_id) for veh_id in ids]

        self.assertCountEqual(before_reset, after_reset)

class TestEmissionPath(unittest.TestCase):
    """
    Tests that the default emission path of an environment is set to None.
    If it is not None, then sumo starts accumulating memory.
    """

    def setUp(self):
        # set sumo_params to default
        sumo_params = SumoParams()

        # create the environment and scenario classes for a ring road
        self.env, scenario = ring_road_exp_setup(sumo_params=sumo_params)

    def tearDown(self):
        # terminate the traci instance
        self.env.terminate()

        # free data used by the class
        self.env = None

    def runTest(self):
        self.assertIsNone(self.env.sumo_params.emission_path)

class TestApplyingActionsWithSumo(unittest.TestCase):
    """
    Tests the apply_acceleration, apply_lane_change, and choose_routes
    functions in base_env.py
    """
    def setUp(self):
        # create a 2-lane ring road network
        additional_net_params = {"length": 230, "lanes": 3, "speed_limit": 30,
                                 "resolution": 40}
        net_params = NetParams(additional_params=additional_net_params)

        # turn on starting position shuffle
        env_params = EnvParams(starting_position_shuffle=True,
                               additional_params=ADDITIONAL_ENV_PARAMS)

        # place 5 vehicles in the network (we need at least more than 1)
        vehicles = Vehicles()
        vehicles.add(veh_id="test",
                     acceleration_controller=(IDMController, {}),
                     routing_controller=(ContinuousRouter, {}),
                     sumo_car_following_params=SumoCarFollowingParams(
                         accel=1000, decel=1000),
                     num_vehicles=5)

        # create the environment and scenario classes for a ring road
        self.env, scenario = ring_road_exp_setup(net_params=net_params,
                                                 env_params=env_params,
                                                 vehicles=vehicles)

    def tearDown(self):
        # terminate the traci instance
        self.env.terminate()

        # free data used by the class
        self.env = None

    def test_apply_acceleration(self):
        """
        Tests that, in the absence of all failsafes, the acceleration requested
        from sumo is equal to the acceleration witnessed in between steps. Also
        ensures that vehicles can never have velocities below zero given any
        acceleration.
        """
        ids = self.env.vehicles.get_ids()

        vel0 = np.array([self.env.vehicles.get_speed(veh_id)
                         for veh_id in ids])

        # apply a certain set of accelerations to the vehicles in the network
        accel_step0 = np.array([0, 1, 4, 9, 16])
        self.env.apply_acceleration(veh_ids=ids, acc=accel_step0)
        self.env.traci_connection.simulationStep()

        # compare the new velocity of the vehicles to the expected velocity
        # given the accelerations
        vel1 = np.array([self.env.traci_connection.vehicle.getSpeed(veh_id)
                         for veh_id in ids])
        expected_vel1 = (vel0 + accel_step0 * 0.1).clip(min=0)

        np.testing.assert_array_almost_equal(vel1, expected_vel1, 1)

        # collect information on the vehicle in the network from sumo
        veh_obs = self.env.traci_connection.vehicle.getSubscriptionResults()

        # get vehicle ids for the entering, exiting, and colliding vehicles
        id_list = self.env.traci_connection.simulation.getSubscriptionResults()

        # store the network observations in the vehicles class
        self.env.vehicles.update(veh_obs, id_list, self.env)

        # apply a set of decelerations
        accel_step1 = np.array([-16, -9, -4, -1, 0])
        self.env.apply_acceleration(veh_ids=ids, acc=accel_step1)
        self.env.traci_connection.simulationStep()

        # this time, some vehicles should be at 0 velocity (NOT less), and sum
        # are a result of the accelerations that took place
        vel2 = np.array([self.env.traci_connection.vehicle.getSpeed(veh_id)
                         for veh_id in ids])
        expected_vel2 = (vel1 + accel_step1 * 0.1).clip(min=0)

        np.testing.assert_array_almost_equal(vel2, expected_vel2, 1)

    def test_apply_lane_change_errors(self):
        """
        Ensures that apply_lane_change raises ValueErrors when it should
        """
        self.env.reset()
        ids = self.env.vehicles.get_ids()

        # make sure that running apply lane change with a invalid direction
        # values leads to a ValueError
        bad_directions = np.array([-1, 0, 1, 2, 3])

        self.assertRaises(
            ValueError,
            self.env.apply_lane_change, veh_ids=ids, direction=bad_directions)

    def test_apply_lane_change_direction(self):
        """
        Tests the direction method for apply_lane_change. Ensures that the lane
        change action requested from sumo is the same as the lane change that
        occurs, and that vehicles attempting do not issue lane changes in there
        is no lane in te requested direction.
        """
        self.env.reset()
        ids = self.env.vehicles.get_ids()
        lane0 = np.array([self.env.vehicles.get_lane(veh_id)
                          for veh_id in ids])

        # perform lane-changing actions using the direction method
        direction0 = np.array([0, 1, 0, 1, -1])
        self.env.apply_lane_change(ids, direction=direction0)
        self.env.traci_connection.simulationStep()

        # check that the lane vehicle lane changes to the correct direction
        # without skipping lanes
        lane1 = np.array(
            [self.env.traci_connection.vehicle.getLaneIndex(veh_id)
             for veh_id in ids])
        expected_lane1 = (lane0 + np.sign(direction0)).clip(
            min=0, max=self.env.scenario.lanes - 1)

        np.testing.assert_array_almost_equal(lane1, expected_lane1, 1)

        # collect information on the vehicle in the network from sumo
        veh_obs = self.env.traci_connection.vehicle.getSubscriptionResults()

        # get vehicle ids for the entering, exiting, and colliding vehicles
        id_list = self.env.traci_connection.simulation.getSubscriptionResults()

        # store the network observations in the vehicles class
        self.env.vehicles.update(veh_obs, id_list, self.env)

        # perform lane-changing actions using the direction method one more
        # time to test lane changes to the right
        direction1 = np.array([-1, -1, -1, -1, -1])
        self.env.apply_lane_change(ids, direction=direction1)
        self.env.traci_connection.simulationStep()

        # check that the lane vehicle lane changes to the correct direction
        # without skipping lanes
        lane2 = np.array(
            [self.env.traci_connection.vehicle.getLaneIndex(veh_id)
             for veh_id in ids])
        expected_lane2 = (lane1 + np.sign(direction1)).clip(
            min=0, max=self.env.scenario.lanes - 1)

        np.testing.assert_array_almost_equal(lane2, expected_lane2, 1)

class TestSorting(unittest.TestCase):
    """
    Tests that the sorting method returns a list of ids sorted by the
    get_absolute_position() method when sorting is requested, and does nothing
    if it is not requested
    """

    def test_sorting(self):
        # setup a environment with the "sort_vehicles" attribute set to True
        additional_env_params = ADDITIONAL_ENV_PARAMS
        env_params = EnvParams(additional_params=additional_env_params,
                               sort_vehicles=True)
        initial_config = InitialConfig(shuffle=True)
        vehicles = Vehicles()
        vehicles.add(veh_id="test", num_vehicles=5)
        self.env, scenario = ring_road_exp_setup(env_params=env_params,
                                                 initial_config=initial_config,
                                                 vehicles=vehicles)

        self.env.reset()

        sorted_ids = self.env.sorted_ids
        positions = self.env.vehicles.get_absolute_position(sorted_ids)

        # ensure vehicles ids are in sorted order by positions
        self.assertTrue(all(positions[i] <= positions[i + 1]
                            for i in range(len(positions) - 1)))

    def test_no_sorting(self):
        # setup a environment with the "sort_vehicles" attribute set to False,
        # and shuffling so that the vehicles are not sorted by their ids
        additional_env_params = ADDITIONAL_ENV_PARAMS
        env_params = EnvParams(additional_params=additional_env_params,
                               sort_vehicles=True)
        initial_config = InitialConfig(shuffle=True)
        vehicles = Vehicles()
        vehicles.add(veh_id="test", num_vehicles=5)
        self.env, scenario = ring_road_exp_setup(env_params=env_params,
                                                 initial_config=initial_config,
                                                 vehicles=vehicles)

        self.env.reset()

        sorted_ids = list(self.env.sorted_ids)
        ids = self.env.vehicles.get_ids()

        # ensure that the list of ids did not change
        self.assertListEqual(sorted_ids, ids)

class TestWarmUpSteps(unittest.TestCase):

    """Ensures that the appropriate number of warmup steps are run when using
    flow.core.params.EnvParams.warmup_steps"""

    def test_it_works(self):
        warmup_step = 5  # some value

        # start an environment with a number of simulations per step greater
        # than one
        env_params = EnvParams(warmup_steps=warmup_step,
                               additional_params=ADDITIONAL_ENV_PARAMS)
        env, scenario = ring_road_exp_setup(env_params=env_params)

        # time before running a reset
        t1 = env.time_counter
        # perform a reset
        env.reset()
        # time after a reset
        t2 = env.time_counter

        # ensure that the difference in time is equal to sims_per_step
        self.assertEqual(t2 - t1, warmup_step)

class TestSimsPerStep(unittest.TestCase):

    """Ensures that the appropriate number of simultaions are run at any given
    steps when using flow.core.params.EnvParams.sims_per_step"""

    def test_it_works(self):
        sims_per_step = 5  # some value

        # start an environment with a number of simulations per step greater
        # than one
        env_params = EnvParams(sims_per_step=sims_per_step,
                               additional_params=ADDITIONAL_ENV_PARAMS)
        env, scenario = ring_road_exp_setup(env_params=env_params)

        env.reset()
        # time before running a step
        t1 = env.time_counter
        # perform a step
        env.step(rl_actions=[])
        # time after a step
        t2 = env.time_counter

        # ensure that the difference in time is equal to sims_per_step
        self.assertEqual(t2 - t1, sims_per_step)

if __name__ == '__main__':
    unittest.main()

max_stars_repo_name: berkeleyflow/flow

max_issues_repo_name: berkeleyflow/flow

max_forks_repo_name: berkeleyflow/flow


text:
# Copyright (c) Facebook, Inc. and its affiliates.

import os, sys, shutil
import os.path as osp
import cv2
import numpy as np
from . import general_utils as gnu

def main():
    in_dir = "./sample_data/images/single_person"
    out_dir = "./sample_data/images/multi_person"
    gnu.renew_dir(out_dir)

    all_imgs = gnu.get_all_files(in_dir, (".jpg", ".png", ".jpeg"), "full")
    num_img = len(all_imgs)

    for i in range(num_img):
        for j in range(num_img):
            img1 = cv2.imread(all_imgs[i])
            img2 = cv2.imread(all_imgs[j])
            img2 = cv2.resize(img2, img1.shape[:2][::-1])
            res_img = np.concatenate((img1, img2), axis=1)
            res_img_path = osp.join(out_dir, f"{i:02d}_{j:02d}.jpg")
            cv2.imwrite(res_img_path, res_img)

if __name__ == '__main__':
    main()

ext: py

reason: import numpy


text:
"""
Imitate the parser representation.
"""
import inspect
import re
import sys
import os
from functools import partial

from jedi._compatibility import builtins as _builtins, unicode
from jedi import debug
from jedi.cache import underscore_memoization, memoize_method
from jedi.parser.tree import Param, Operator
from jedi.evaluate.helpers import FakeName
from jedi.evaluate.filters import AbstractFilter, AbstractNameDefinition, \
    ContextNameMixin
from jedi.evaluate.context import Context, LazyKnownContext
from . import fake

_sep = os.path.sep
if os.path.altsep is not None:
    _sep += os.path.altsep
_path_re = re.compile('(?:\.[^{0}]+|[{0}]__init__\.py)$'.format(re.escape(_sep)))
del _sep

class CheckAttribute(object):
    """Raises an AttributeError if the attribute X isn't available."""
    def __init__(self, func):
        self.func = func
        # Remove the py in front of e.g. py__call__.
        self.check_name = func.__name__[2:]

    def __get__(self, instance, owner):
        # This might raise an AttributeError. That's wanted.
        if self.check_name == '__iter__':
            # Python iterators are a bit strange, because there's no need for
            # the __iter__ function as long as __getitem__ is defined (it will
            # just start with __getitem__(0). This is especially true for
            # Python 2 strings, where `str.__iter__` is not even defined.
            try:
                iter(instance.obj)
            except TypeError:
                raise AttributeError
        else:
            getattr(instance.obj, self.check_name)
        return partial(self.func, instance)

class CompiledObject(Context):
    path = None  # modules have this attribute - set it to None.
    used_names = {}  # To be consistent with modules.

    def __init__(self, evaluator, obj, parent_context=None, faked_class=None):
        super(CompiledObject, self).__init__(evaluator, parent_context)
        self.obj = obj
        # This attribute will not be set for most classes, except for fakes.
        self.tree_node = faked_class

    def get_root_node(self):
        # To make things a bit easier with filters we add this method here.
        return self.get_root_context()

    @CheckAttribute
    def py__call__(self, params):
        if inspect.isclass(self.obj):
            from jedi.evaluate.instance import CompiledInstance
            return set([CompiledInstance(self.evaluator, self.parent_context, self, params)])
        else:
            return set(self._execute_function(params))

    @CheckAttribute
    def py__class__(self):
        return create(self.evaluator, self.obj.__class__)

    @CheckAttribute
    def py__mro__(self):
        return (self,) + tuple(create(self.evaluator, cls) for cls in self.obj.__mro__[1:])

    @CheckAttribute
    def py__bases__(self):
        return tuple(create(self.evaluator, cls) for cls in self.obj.__bases__)

    def py__bool__(self):
        return bool(self.obj)

    def py__file__(self):
        try:
            return self.obj.__file__
        except AttributeError:
            return None

    def is_class(self):
        return inspect.isclass(self.obj)

    @property
    def doc(self):
        return inspect.getdoc(self.obj) or ''

    @property
    def get_params(self):
        return []  # TODO Fix me.
        params_str, ret = self._parse_function_doc()
        tokens = params_str.split(',')
        if inspect.ismethoddescriptor(self.obj):
            tokens.insert(0, 'self')
        params = []
        for p in tokens:
            parts = [FakeName(part) for part in p.strip().split('=')]
            if len(parts) > 1:
                parts.insert(1, Operator('=', (0, 0)))
            params.append(Param(parts, self))
        return params

    def get_param_names(self):
        params_str, ret = self._parse_function_doc()
        tokens = params_str.split(',')
        if inspect.ismethoddescriptor(self.obj):
            tokens.insert(0, 'self')
        for p in tokens:
            parts = p.strip().split('=')
            if len(parts) > 1:
                parts.insert(1, Operator('=', (0, 0)))
            yield UnresolvableParamName(self, parts[0])

    def __repr__(self):
        return '<%s: %s>' % (self.__class__.__name__, repr(self.obj))

    @underscore_memoization
    def _parse_function_doc(self):
        if self.doc is None:
            return '', ''

        return _parse_function_doc(self.doc)

    @property
    def api_type(self):
        obj = self.obj
        if inspect.isclass(obj):
            return 'class'
        elif inspect.ismodule(obj):
            return 'module'
        elif inspect.isbuiltin(obj) or inspect.ismethod(obj) \
                or inspect.ismethoddescriptor(obj) or inspect.isfunction(obj):
            return 'function'
        # Everything else...
        return 'instance'

    @property
    def type(self):
        """Imitate the tree.Node.type values."""
        cls = self._get_class()
        if inspect.isclass(cls):
            return 'classdef'
        elif inspect.ismodule(cls):
            return 'file_input'
        elif inspect.isbuiltin(cls) or inspect.ismethod(cls) or \
                inspect.ismethoddescriptor(cls):
            return 'funcdef'

    @underscore_memoization
    def _cls(self):
        """
        We used to limit the lookups for instantiated objects like list(), but
        this is not the case anymore. Python itself
        """
        # Ensures that a CompiledObject is returned that is not an instance (like list)
        return self

    def _get_class(self):
        if not fake.is_class_instance(self.obj) or \
                inspect.ismethoddescriptor(self.obj):  # slots
            return self.obj

        try:
            return self.obj.__class__
        except AttributeError:
            # happens with numpy.core.umath._UFUNC_API (you get it
            # automatically by doing `import numpy`.
            return type

    def get_filters(self, search_global=False, is_instance=False,
                    until_position=None, origin_scope=None):
        yield self._ensure_one_filter(is_instance)

    @memoize_method
    def _ensure_one_filter(self, is_instance):
        """
        search_global shouldn't change the fact that there's one dict, this way
        there's only one `object`.
        """
        return CompiledObjectFilter(self.evaluator, self, is_instance)

    def get_subscope_by_name(self, name):
        if name in dir(self.obj):
            return CompiledName(self.evaluator, self, name).parent
        else:
            raise KeyError("CompiledObject doesn't have an attribute '%s'." % name)

    @CheckAttribute
    def py__getitem__(self, index):
        if type(self.obj) not in (str, list, tuple, unicode, bytes, bytearray, dict):
            # Get rid of side effects, we won't call custom `__getitem__`s.
            return set()

        return set([create(self.evaluator, self.obj[index])])

    @CheckAttribute
    def py__iter__(self):
        if type(self.obj) not in (str, list, tuple, unicode, bytes, bytearray, dict):
            # Get rid of side effects, we won't call custom `__getitem__`s.
            return

        for part in self.obj:
            yield LazyKnownContext(create(self.evaluator, part))

    def py__name__(self):
        try:
            return self._get_class().__name__
        except AttributeError:
            return None

    @property
    def name(self):
        try:
            name = self._get_class().__name__
        except AttributeError:
            name = repr(self.obj)
        return CompiledContextName(self, name)

    def _execute_function(self, params):
        if self.type != 'funcdef':
            return

        for name in self._parse_function_doc()[1].split():
            try:
                bltn_obj = getattr(_builtins, name)
            except AttributeError:
                continue
            else:
                if bltn_obj is None:
                    # We want to evaluate everything except None.
                    # TODO do we?
                    continue
                bltn_obj = create(self.evaluator, bltn_obj)
                for result in self.evaluator.execute(bltn_obj, params):
                    yield result

    def is_scope(self):
        return True

    def get_self_attributes(self):
        return []  # Instance compatibility

    def get_imports(self):
        return []  # Builtins don't have imports

class CompiledName(AbstractNameDefinition):
    def __init__(self, evaluator, parent_context, name):
        self._evaluator = evaluator
        self.parent_context = parent_context
        self.string_name = name

    def __repr__(self):
        try:
            name = self.parent_context.name  # __name__ is not defined all the time
        except AttributeError:
            name = None
        return '<%s: (%s).%s>' % (self.__class__.__name__, name, self.string_name)

    @property
    def api_type(self):
        return next(iter(self.infer())).api_type

    @underscore_memoization
    def infer(self):
        module = self.parent_context.get_root_context()
        return [_create_from_name(self._evaluator, module, self.parent_context, self.string_name)]

class UnresolvableParamName(AbstractNameDefinition):
    api_type = 'param'

    def __init__(self, compiled_obj, name):
        self.parent_context = compiled_obj.parent_context
        self.string_name = name

    def infer(self):
        return set()

class CompiledContextName(ContextNameMixin, AbstractNameDefinition):
    def __init__(self, context, name):
        self.string_name = name
        self._context = context
        self.parent_context = context.parent_context

class EmptyCompiledName(AbstractNameDefinition):
    """
    Accessing some names will raise an exception. To avoid not having any
    completions, just give Jedi the option to return this object. It infers to
    nothing.
    """
    def __init__(self, evaluator, name):
        self.parent_context = evaluator.BUILTINS
        self.string_name = name

    def infer(self):
        return []

class CompiledObjectFilter(AbstractFilter):
    name_class = CompiledName

    def __init__(self, evaluator, compiled_object, is_instance=False):
        self._evaluator = evaluator
        self._compiled_object = compiled_object
        self._is_instance = is_instance

    @memoize_method
    def get(self, name):
        name = str(name)
        obj = self._compiled_object.obj
        try:
            getattr(obj, name)
            if self._is_instance and name not in dir(obj):
                return []
        except AttributeError:
            return []
        except Exception:
            # This is a bit ugly. We're basically returning this to make
            # lookups possible without having the actual attribute. However
            # this makes proper completion possible.
            return [EmptyCompiledName(self._evaluator, name)]
        return [self._create_name(name)]

    def values(self):
        obj = self._compiled_object.obj

        names = []
        for name in dir(obj):
            names += self.get(name)

        is_instance = self._is_instance or fake.is_class_instance(obj)
        # ``dir`` doesn't include the type names.
        if not inspect.ismodule(obj) and obj != type and not is_instance:
            for filter in create(self._evaluator, type).get_filters():
                names += filter.values()
        return names

    def _create_name(self, name):
        return self.name_class(self._evaluator, self._compiled_object, name)

def dotted_from_fs_path(fs_path, sys_path):
    """
    Changes `/usr/lib/python3.4/email/utils.py` to `email.utils`.  I.e.
    compares the path with sys.path and then returns the dotted_path. If the
    path is not in the sys.path, just returns None.
    """
    if os.path.basename(fs_path).startswith('__init__.'):
        # We are calculating the path. __init__ files are not interesting.
        fs_path = os.path.dirname(fs_path)

    # prefer
    #   - UNIX
    #     /path/to/pythonX.Y/lib-dynload
    #     /path/to/pythonX.Y/site-packages
    #   - Windows
    #     C:\path\to\DLLs
    #     C:\path\to\Lib\site-packages
    # over
    #   - UNIX
    #     /path/to/pythonX.Y
    #   - Windows
    #     C:\path\to\Lib
    path = ''
    for s in sys_path:
        if (fs_path.startswith(s) and len(path) < len(s)):
            path = s

    # - Window
    # X:\path\to\lib-dynload/datetime.pyd => datetime
    module_path = fs_path[len(path):].lstrip(os.path.sep).lstrip('/')
    # - Window
    # Replace like X:\path\to\something/foo/bar.py
    return _path_re.sub('', module_path).replace(os.path.sep, '.').replace('/', '.')

def load_module(evaluator, path=None, name=None):
    sys_path = evaluator.sys_path
    if path is not None:
        dotted_path = dotted_from_fs_path(path, sys_path=sys_path)
    else:
        dotted_path = name

    if dotted_path is None:
        p, _, dotted_path = path.partition(os.path.sep)
        sys_path.insert(0, p)

    temp, sys.path = sys.path, sys_path
    try:
        __import__(dotted_path)
    except RuntimeError:
        if 'PySide' in dotted_path or 'PyQt' in dotted_path:
            # RuntimeError: the PyQt4.QtCore and PyQt5.QtCore modules both wrap
            # the QObject class.
            # See https://github.com/davidhalter/jedi/pull/483
            return None
        raise
    except ImportError:
        # If a module is "corrupt" or not really a Python module or whatever.
        debug.warning('Module %s not importable.', path)
        return None
    finally:
        sys.path = temp

    # Just access the cache after import, because of #59 as well as the very
    # complicated import structure of Python.
    module = sys.modules[dotted_path]

    return create(evaluator, module)

docstr_defaults = {
    'floating point number': 'float',
    'character': 'str',
    'integer': 'int',
    'dictionary': 'dict',
    'string': 'str',
}

def _parse_function_doc(doc):
    """
    Takes a function and returns the params and return value as a tuple.
    This is nothing more than a docstring parser.

    TODO docstrings like utime(path, (atime, mtime)) and a(b [, b]) -> None
    TODO docstrings like 'tuple of integers'
    """
    # parse round parentheses: def func(a, (b,c))
    try:
        count = 0
        start = doc.index('(')
        for i, s in enumerate(doc[start:]):
            if s == '(':
                count += 1
            elif s == ')':
                count -= 1
            if count == 0:
                end = start + i
                break
        param_str = doc[start + 1:end]
    except (ValueError, UnboundLocalError):
        # ValueError for doc.index
        # UnboundLocalError for undefined end in last line
        debug.dbg('no brackets found - no param')
        end = 0
        param_str = ''
    else:
        # remove square brackets, that show an optional param ( = None)
        def change_options(m):
            args = m.group(1).split(',')
            for i, a in enumerate(args):
                if a and '=' not in a:
                    args[i] += '=None'
            return ','.join(args)

        while True:
            param_str, changes = re.subn(r' ?\[([^\[\]]+)\]',
                                         change_options, param_str)
            if changes == 0:
                break
    param_str = param_str.replace('-', '_')  # see: isinstance.__doc__

    # parse return value
    r = re.search('-[>-]* ', doc[end:end + 7])
    if r is None:
        ret = ''
    else:
        index = end + r.end()
        # get result type, which can contain newlines
        pattern = re.compile(r'(,\n|[^\n-])+')
        ret_str = pattern.match(doc, index).group(0).strip()
        # New object -> object()
        ret_str = re.sub(r'[nN]ew (.*)', r'\1()', ret_str)

        ret = docstr_defaults.get(ret_str, ret_str)

    return param_str, ret

def _create_from_name(evaluator, module, compiled_object, name):
    obj = compiled_object.obj
    faked = None
    try:
        faked = fake.get_faked(evaluator, module, obj, parent_context=compiled_object, name=name)
        if faked.type == 'funcdef':
            from jedi.evaluate.representation import FunctionContext
            return FunctionContext(evaluator, compiled_object, faked)
    except fake.FakeDoesNotExist:
        pass

    try:
        obj = getattr(obj, name)
    except AttributeError:
        # Happens e.g. in properties of
        # PyQt4.QtGui.QStyleOptionComboBox.currentText
        # -> just set it to None
        obj = None
    return create(evaluator, obj, parent_context=compiled_object, faked=faked)

def builtin_from_name(evaluator, string):
    bltn_obj = getattr(_builtins, string)
    return create(evaluator, bltn_obj)

def _a_generator(foo):
    """Used to have an object to return for generators."""
    yield 42
    yield foo

_SPECIAL_OBJECTS = {
    'FUNCTION_CLASS': type(load_module),
    'METHOD_CLASS': type(CompiledObject.is_class),
    'MODULE_CLASS': type(os),
    'GENERATOR_OBJECT': _a_generator(1.0),
    'BUILTINS': _builtins,
}

def get_special_object(evaluator, identifier):
    obj = _SPECIAL_OBJECTS[identifier]
    return create(evaluator, obj, parent_context=create(evaluator, _builtins))

def compiled_objects_cache(attribute_name):
    def decorator(func):
        """
        This decorator caches just the ids, oopposed to caching the object itself.
        Caching the id has the advantage that an object doesn't need to be
        hashable.
        """
        def wrapper(evaluator, obj, parent_context=None, module=None, faked=None):
            cache = getattr(evaluator, attribute_name)
            # Do a very cheap form of caching here.
            key = id(obj), id(parent_context)
            try:
                return cache[key][0]
            except KeyError:
                # TODO this whole decorator is way too ugly
                result = func(evaluator, obj, parent_context, module, faked)
                # Need to cache all of them, otherwise the id could be overwritten.
                cache[key] = result, obj, parent_context, module, faked
                return result
        return wrapper

    return decorator

@compiled_objects_cache('compiled_cache')
def create(evaluator, obj, parent_context=None, module=None, faked=None):
    """
    A very weird interface class to this module. The more options provided the
    more acurate loading compiled objects is.
    """
    if inspect.ismodule(obj):
        if parent_context is not None:
            # Modules don't have parents, be careful with caching: recurse.
            return create(evaluator, obj)
    else:
        if parent_context is None and obj != _builtins:
            return create(evaluator, obj, create(evaluator, _builtins))

        try:
            faked = fake.get_faked(evaluator, module, obj, parent_context=parent_context)
            if faked.type == 'funcdef':
                from jedi.evaluate.representation import FunctionContext
                return FunctionContext(evaluator, parent_context, faked)
        except fake.FakeDoesNotExist:
            pass

    return CompiledObject(evaluator, obj, parent_context, faked)

max_stars_repo_name: Andrewnetwork/pythonVSCode

max_issues_repo_name: Andrewnetwork/pythonVSCode

max_forks_repo_name: Andrewnetwork/pythonVSCode


text:
#!/usr/bin/env python
# coding: utf-8
# # Text provided under a Creative Commons Attribution license, CC-BY.  All code is made available under the FSF-approved BSD-3 license.  (c) Lorena A. Barba, Gilbert F. Forsyth 2017. Thanks to NSF for support via CAREER award #1149784.
# [@LorenaABarba](https://twitter.com/LorenaABarba)

# 12 steps to Navier–Stokes
# =====
# ***

# For a moment, recall the Navier–Stokes equations for an incompressible fluid, where $\vec{v}$ represents the velocity field:
#
# $$
# \begin{eqnarray*}
# \nabla \cdot\vec{v} &=& 0 \\
# \frac{\partial \vec{v}}{\partial t}+(\vec{v}\cdot\nabla)\vec{v} &=& -\frac{1}{\rho}\nabla p + \nu \nabla^2\vec{v}
# \end{eqnarray*}
# $$
#
# The first equation represents mass conservation at constant density. The second equation is the conservation of momentum. But a problem appears: the continuity equation for incompressble flow does not have a dominant variable and there is no obvious way to couple the velocity and the pressure. In the case of compressible flow, in contrast, mass continuity would provide an evolution equation for the density $\rho$, which is coupled with an equation of state relating $\rho$ and $p$.
#
# In incompressible flow, the continuity equation $\nabla \cdot\vec{v}=0$ provides a *kinematic constraint* that requires the pressure field to evolve so that the rate of expansion $\nabla \cdot\vec{v}$ should vanish everywhere. A way out of this difficulty is to *construct* a pressure field that guarantees continuity is satisfied; such a relation can be obtained by taking the divergence of the momentum equation. In that process, a Poisson equation for the pressure shows up!

# Step 10: 2D Poisson Equation
# ----
# ***

# Poisson's equation is obtained from adding a source term to the right-hand-side of Laplace's equation:
#
# $$\frac{\partial ^2 p}{\partial x^2} + \frac{\partial ^2 p}{\partial y^2} = b$$
#
# So, unlinke the Laplace equation, there is some finite value inside the field that affects the solution. Poisson's equation acts to "relax" the initial sources in the field.
#
# In discretized form, this looks almost the same as [Step 9](./12_Step_9.ipynb), except for the source term:
#
# $$\frac{p_{i+1,j}^{n}-2p_{i,j}^{n}+p_{i-1,j}^{n}}{\Delta x^2}+\frac{p_{i,j+1}^{n}-2 p_{i,j}^{n}+p_{i,j-1}^{n}}{\Delta y^2}=b_{i,j}^{n}$$
#
# As before, we rearrange this so that we obtain an equation for $p$ at point $i,j$. Thus, we obtain:
#
# $$p_{i,j}^{n}=\frac{(p_{i+1,j}^{n}+p_{i-1,j}^{n})\Delta y^2+(p_{i,j+1}^{n}+p_{i,j-1}^{n})\Delta x^2-b_{i,j}^{n}\Delta x^2\Delta y^2}{2(\Delta x^2+\Delta y^2)}$$
#

# We will solve this equation by assuming an initial state of $p=0$ everywhere, and applying boundary conditions as follows:
#
# $p=0$ at $x=0, \ 2$ and $y=0, \ 1$
#
# and the source term consists of two initial spikes inside the domain, as follows:
#
# $b_{i,j}=100$ at $i=\frac{1}{4}nx, j=\frac{1}{4}ny$
#
# $b_{i,j}=-100$ at $i=\frac{3}{4}nx, j=\frac{3}{4}ny$
#
# $b_{i,j}=0$ everywhere else.
#
# The iterations will advance in pseudo-time to relax the initial spikes. The relaxation under Poisson's equation gets slower and slower as they progress. *Why?*

# Let's look at one possible way to write the code for Poisson's equation. As always, we load our favorite Python libraries. We also want to make some lovely plots in 3D. Let's get our parameters defined and the initialization out of the way. What do you notice of the approach below?

# In[1]:

import numpy
from matplotlib import pyplot, cm
from mpl_toolkits.mplot3d import Axes3D
## get_ipython().run_line_magic('matplotlib', 'inline')

# In[2]:

# Parameters
nx = 50
ny = 50
nt = 100
xmin = 0
xmax = 2
ymin = 0
ymax = 1

dx = (xmax - xmin) / (nx - 1)
dy = (ymax - ymin) / (ny - 1)

# Initialization
p = numpy.zeros((ny, nx))
pd = numpy.zeros((ny, nx))
b = numpy.zeros((ny, nx))
x = numpy.linspace(xmin, xmax, nx)
y = numpy.linspace(xmin, xmax, ny)

# Source
b[int(ny / 4), int(nx / 4)] = 100
b[int(3 * ny / 4), int(3 * nx / 4)] = -100

# With that, we are ready to advance the initial guess in pseudo-time. How is the code below different from the function used in [Step 9](./12_Step_9.ipynb) to solve Laplace's equation?

# In[3]:

for it in range(nt):

    pd = p.copy()

    p[1:-1, 1:-1] = (((pd[1:-1, 2:] + pd[1:-1, :-2]) * dy**2 +
                      (pd[2:, 1:-1] + pd[:-2, 1:-1]) * dx**2 -
                      b[1:-1, 1:-1] * dx**2 * dy**2) /
                     (2 * (dx**2 + dy**2)))

    p[0, :] = 0
    p[ny - 1, :] = 0
    p[:, 0] = 0
    p[:, nx - 1] = 0

# Maybe we could reuse our plotting function from [Step 9](./12_Step_9.ipynb), don't you think?

# In[4]:

def plot2D(x, y, p):
    fig = pyplot.figure(figsize=(11, 7), dpi=100)
    ax = fig.gca(projection='3d')
    X, Y = numpy.meshgrid(x, y)
    surf = ax.plot_surface(X, Y, p[:], rstride=1, cstride=1, cmap=cm.viridis,
                           linewidth=0, antialiased=False)
    ax.view_init(30, 225)
    ax.set_xlabel('$x$')
    ax.set_ylabel('$y$')

# In[5]:

plot2D(x, y, p)
pyplot.show()

# Ah! The wonders of code reuse! Now, you probably think: "Well, if I've written this neat little function that does something so useful, I want to use it over and over again. How can I do this without copying and pasting it each time? —If you are very curious about this, you'll have to learn about *packaging*. But this goes beyond the scope of our CFD lessons. You'll just have to Google it if you really want to know.

# ***

# ## Learn More

# To learn more about the role of the Poisson equation in CFD, watch **Video Lesson 11** on You Tube:

# In[6]:

#from IPython.display import YouTubeVideo
# YouTubeVideo('ZjfxA3qq2Lg')

# In[7]:

#from IPython.core.display import HTML
# def css_styling():
#    styles = open("../styles/custom.css", "r").read()
#    return HTML(styles)
# css_styling()

# > (The cell above executes the style for this notebook.)

max_stars_repo_name: tnakaicode/python-cfd

max_issues_repo_name: tnakaicode/python-cfd

max_forks_repo_name: tnakaicode/python-cfd


text:
#!/usr/bin/env python

import argparse
import datetime
import functools
import os
import os.path as osp
import socket

os.environ['MPLBACKEND'] = 'Agg'  # NOQA

import chainer
from chainer.training import extensions
import chainer_mask_rcnn as cmr
import fcn
import numpy as np

import instance_occlsegm_lib
from instance_occlsegm_lib.contrib import instance_occlsegm

here = osp.dirname(osp.abspath(__file__))

def transform_size(in_data):
    img, lbl, lbl_occ = in_data
    H, W, C = img.shape
    assert lbl.shape == (H, W)
    assert lbl_occ.shape == (H, W, lbl_occ.shape[2])

    H_dst = int(np.ceil(H / 16) * 16)
    W_dst = int(np.ceil(W / 16) * 16)

    img = np.pad(
        img,
        [(0, H_dst - H), (0, W_dst - W), (0, 0)],
        mode='constant',
        constant_values=0,
    )
    assert img.shape == (H_dst, W_dst, C)
    assert (img[H:H_dst, :, :] == 0).all()
    assert (img[:, W:W_dst, :] == 0).all()

    lbl = np.pad(
        lbl,
        [(0, H_dst - H), (0, W_dst - W)],
        mode='constant',
        constant_values=-1,
    )
    assert lbl.shape == (H_dst, W_dst)
    assert (lbl[H:H_dst, :] == -1).all()
    assert (lbl[:, W:W_dst] == -1).all()

    lbl_occ = np.pad(
        lbl_occ,
        [(0, H_dst - H), (0, W_dst - W), (0, 0)],
        mode='constant',
        constant_values=-1,
    )
    assert lbl_occ.shape == (H_dst, W_dst, lbl_occ.shape[2])
    assert (lbl_occ[H:H_dst, :, :] == -1).all()
    assert (lbl_occ[:, W:W_dst, :] == -1).all()
    return img, lbl, lbl_occ

def transform(in_data, extractor='res'):
    img, lbl, lbl_occ = transform_size(in_data)

    if extractor == 'res':
        MEAN_RGB = (123.152, 115.903, 103.063)  # Same as MaskRCNNResNet
        img = img.astype(np.float32)
        img -= MEAN_RGB
        img = img.transpose(2, 0, 1)  # HWC -> CHW
    else:
        assert extractor == 'vgg'
        img, = fcn.datasets.transform_lsvrc2012_vgg16((img,))

    lbl_occ = lbl_occ.transpose(2, 0, 1)  # HWC -> CHW

    return img, lbl, lbl_occ

def get_data(batch_size=1, comm=None, extractor='res'):
    dataset_train = instance_occlsegm.datasets.OcclusionSegmentationDataset(
        split='train'
    )
    dataset_valid = instance_occlsegm.datasets.OcclusionSegmentationDataset(
        split='test'
    )
    # dataset_train = fcn.datasets.SBDClassSeg(split='train')
    # dataset_valid = fcn.datasets.VOC2011ClassSeg(split='seg11valid')
    class_names = dataset_train.class_names

    if comm:
        import chainermn
        if comm.rank != 0:
            dataset_train = None
        dataset_train = chainermn.scatter_dataset(
            dataset_train, comm, shuffle=True
        )

    iter_train = chainer.iterators.SerialIterator(
        chainer.datasets.TransformDataset(
            dataset_train,
            lambda x: transform(x, extractor=extractor),
        ),
        batch_size=batch_size,
    )
    iter_valid_raw = chainer.iterators.SerialIterator(
        chainer.datasets.TransformDataset(dataset_valid, transform_size),
        batch_size=1,
        repeat=False,
        shuffle=False,
    )
    iter_valid = chainer.iterators.SerialIterator(
        chainer.datasets.TransformDataset(
            dataset_valid,
            lambda x: transform(x, extractor=extractor),
        ),
        batch_size=1,
        repeat=False,
        shuffle=False,
    )

    return class_names, iter_train, iter_valid, iter_valid_raw

def get_trainer(
    optimizer,
    iter_train,
    iter_valid,
    iter_valid_raw,
    class_names,
    args,
    comm=None,
    device=None,
    extractor='res',
):
    if device is None:
        device = args.gpu

    model = optimizer.target

    converter = functools.partial(
        chainer.dataset.concat_examples, padding=(0, -1, -1),
    )
    updater = chainer.training.StandardUpdater(
        iter_train, optimizer, converter=converter, device=device,
    )

    trainer = chainer.training.Trainer(
        updater, (args.max_epoch, 'epoch'), out=args.out)

    if comm and comm.rank != 0:
        return trainer

    trainer.extend(fcn.extensions.ParamsReport(args.__dict__))

    trainer.extend(extensions.ProgressBar(update_interval=10))

    trainer.extend(extensions.LogReport(
        trigger=(args.interval_print, 'iteration')))
    trainer.extend(extensions.PrintReport([
        'epoch',
        'iteration',
        'elapsed_time',
        'main/loss_vis',
        'main/loss_occ',
        'main/loss',
        'validation/main/miou/vis'
        'validation/main/miou/occ'
        'validation/main/miou',
    ]))

    trainer.extend(
        instance_occlsegm.extensions.OcclusionSegmentationVisReport(
            iter_valid_raw,
            model.predictor,
            transform=lambda x: transform(x, extractor=extractor),
            class_names=class_names,
            device=device,
            shape=(4, 2),
        ),
        trigger=(args.interval_eval, 'epoch'))

    trainer.extend(
        instance_occlsegm.extensions.OcclusionSegmentationEvaluator(
            iter_valid, model.predictor, label_names=class_names),
        trigger=(args.interval_eval, 'epoch'))

    trainer.extend(extensions.snapshot_object(
        target=model.predictor, filename='model_best.npz'),
        trigger=chainer.training.triggers.MaxValueTrigger(
            key='validation/main/miou',
            trigger=(args.interval_eval, 'epoch')))

    assert extensions.PlotReport.available()
    trainer.extend(extensions.PlotReport(
        y_keys=['main/loss', 'main/loss_vis', 'main/loss_occ'],
        x_key='iteration',
        file_name='loss.png', trigger=(args.interval_print, 'iteration')))
    trainer.extend(extensions.PlotReport(
        y_keys=[
            'validation/main/miou',
            'validation/main/miou/vis',
            'validation/main/miou/occ',
        ],
        x_key='iteration',
        file_name='miou.png', trigger=(args.interval_print, 'iteration')))

    return trainer

def main():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument('-g', '--gpu', type=int, help='gpu id')
    parser.add_argument(
        '--multi-node', action='store_true', help='use multi node'
    )
    parser.add_argument(
        '--notrain-occlusion', action='store_true', help='not train occlusion'
    )
    parser.add_argument(
        '--batch-size-per-gpu', type=int, default=1, help='batch size per gpu'
    )
    parser.add_argument(
        '--lr-base', type=float, default=1e-4, help='learning rate base'
    )
    parser.add_argument(
        '--weight-decay', type=float, default=0.0001, help='weight decay'
    )
    parser.add_argument('--max-epoch', type=float, default=36, help='epoch')
    choices_model = ['fcn16s_resnet50']
    parser.add_argument(
        '--model',
        choices=choices_model,
        default=choices_model[0],
        help='model',
    )
    args = parser.parse_args()

    if args.multi_node:
        import chainermn
        comm = chainermn.create_communicator('hierarchical')
        device = comm.intra_rank

        args.n_node = comm.inter_size
        args.n_gpu = comm.size
        chainer.cuda.get_device_from_id(device).use()
    else:
        comm = None
        args.n_node = 1
        args.n_gpu = 1
        chainer.cuda.get_device_from_id(args.gpu).use()
        device = args.gpu

    now = datetime.datetime.now()
    args.timestamp = now.isoformat()

    if comm is None or comm.rank == 0:
        out = osp.join(here, 'logs', now.strftime('%Y%m%d_%H%M%S.%f'))
    else:
        out = None
    if comm:
        args.out = comm.bcast_obj(out)
    else:
        args.out = out
    del out

    args.dataset = 'occlusion'
    args.hostname = socket.gethostname()
    args.git_hash = instance_occlsegm_lib.utils.git_hash(__file__)

    args.batch_size = args.batch_size_per_gpu * args.n_gpu
    args.lr = args.lr_base * args.batch_size
    args.momentum = 0.99

    args.interval_print = 20
    args.interval_eval = 0.5

    # -------------------------------------------------------------------------

    if 'resnet' in args.model:
        extractor = 'res'
    else:
        assert 'vgg' in args.model
        extractor = 'vgg'

    # data
    class_names, iter_train, iter_valid, iter_valid_raw = get_data(
        batch_size=args.batch_size_per_gpu, comm=comm, extractor=extractor,
    )
    n_class = len(class_names)

    # model
    if args.model == 'fcn16s_resnet50':
        predictor = instance_occlsegm.models.FCN16sResNetOcclusion(
            n_class=n_class
        )
        model = instance_occlsegm.models.OcclusionSegmentationTrainChain(
            predictor, train_occlusion=not args.notrain_occlusion
        )
    else:
        raise ValueError

    if device >= 0:
        chainer.cuda.get_device(device).use()
        model.to_gpu()

    # optimizer
    optimizer = chainer.optimizers.MomentumSGD(
        lr=args.lr, momentum=args.momentum)
    if comm:
        optimizer = chainermn.create_multi_node_optimizer(optimizer, comm)
    optimizer.setup(model)
    optimizer.add_hook(chainer.optimizer.WeightDecay(rate=args.weight_decay))
    for p in model.params():
        if p.name == 'b':
            p.update_rule = chainer.optimizers.momentum_sgd.MomentumSGDRule(
                lr=optimizer.lr * 2, momentum=0)

    if extractor == 'res':
        model.predictor.extractor.conv1.disable_update()
        model.predictor.extractor.bn1.disable_update()
        model.predictor.extractor.res2.disable_update()
        for link in model.links():
            if isinstance(link, cmr.links.AffineChannel2D):
                link.disable_update()
    else:
        raise ValueError
        # model.upscore.disable_update()

    # trainer
    trainer = get_trainer(
        optimizer,
        iter_train,
        iter_valid,
        iter_valid_raw,
        class_names=class_names,
        args=args,
        comm=comm,
        device=device,
        extractor=extractor,
    )
    trainer.run()

if __name__ == '__main__':
    main()


text:
import datetime
import time
import math
from functools import lru_cache
from collections.abc import Sequence

import numpy as np
from numpy.linalg import pinv

import ephem

from shapely.geometry import shape
from affine import Affine
import pyproj

from gbdxtools.rda.graph import VIRTUAL_RDA_URL
import gbdxtools.rda.constants as constants

from gbdxtools.deprecate import GBDXDeprecation, deprecation

RDA_TO_DTYPE = {
    "BINARY": "bool",
    "BYTE": "uint8",
    "SHORT": "short",
    "UNSIGNED_SHORT": "ushort",
    "INTEGER": "int32",
    "UNSIGNED_INTEGER": "uint32",
    "LONG": "int64",
    "UNSIGNED_LONG": "uint64",
    "FLOAT": "float32",
    "DOUBLE": "float64"
}

CUSTOM_PRJ = {
    "EPSG:54008": "+proj=sinu +lon_0=0 +x_0=0 +y_0=0 +ellps=WGS84 +datum=WGS84 +units=m +no_defs"
}

@lru_cache()
def get_proj(prj_code):
    """
      Helper method for handling projection codes that are unknown to pyproj

      Args:
          prj_code (str): an epsg proj code

      Returns:
          projection: a pyproj projection
    """
    if prj_code in CUSTOM_PRJ:
        proj = pyproj.CRS(CUSTOM_PRJ[prj_code])
    else:
        proj = pyproj.CRS(prj_code)
    return proj

def reproject_params(proj):
    _params = {}
    if proj is not None:
        _params["SourceSRSCode"] = "EPSG:4326"
        _params["Source pixel-to-world transform"] = None
        _params["DestSRSCode"] = proj
        _params["Dest pixel-to-world transform"] = None
    return _params

def ortho_params(proj, gsd=None):
    params = {}
    if gsd is not None:
        params["Requested GSD"] = str(gsd)
        params["Resampling Kernel"] = "INTERP_BILINEAR"
        params["Grid Size"] = 10
    if proj is not None:
        params["Output Coordinate Reference System"] = proj
        params["Sensor Model"] = None
        params["Elevation Source"] = ""
        params["Output Pixel to World Transform"] = ""
    return params

def calc_toa_gain_offset(meta):
    """
    Compute (gain, offset) tuples for each band of the specified image metadata
    """
    # Set satellite index to look up cal factors
    sat_index = meta['satid'].upper() + "_" + meta['bandid'].upper()

    # Set scale for at sensor radiance
    # Eq is:
    # L = GAIN * DN * (ACF/EBW) + Offset
    # ACF abscal factor from meta data
    # EBW effectiveBandwidth from meta data
    # Gain provided by abscal from const
    # Offset provided by abscal from const
    acf = np.asarray(meta['abscalfactor'])  # Should be nbands length
    ebw = np.asarray(meta['effbandwidth'])  # Should be nbands length
    gain = np.asarray(constants.DG_ABSCAL_GAIN[sat_index])
    scale = (acf / ebw) * gain
    offset = np.asarray(constants.DG_ABSCAL_OFFSET[sat_index])

    e_sun_index = meta['satid'].upper() + "_" + meta['bandid'].upper()
    e_sun = np.asarray(constants.DG_ESUN[e_sun_index])
    sun = ephem.Sun()
    img_obs = ephem.Observer()
    img_obs.lon = meta['latlonhae'][1]
    img_obs.lat = meta['latlonhae'][0]
    img_obs.elevation = meta['latlonhae'][2]
    img_obs.date = datetime.datetime.fromtimestamp(meta['img_datetime_obj_utc']['$date'] / 1000.0).strftime(
        '%Y-%m-%d %H:%M:%S.%f')
    sun.compute(img_obs)
    d_es = sun.earth_distance

    # Pull sun elevation from the image metadata
    # theta_s can be zenith or elevation - the calc below will us either
    # a cos or s in respectively
    # theta_s = float(self.meta_dg.IMD.IMAGE.MEANSUNEL)
    theta_s = 90 - float(meta['mean_sun_el'])
    scale2 = (d_es ** 2 * np.pi) / (e_sun * np.cos(np.deg2rad(theta_s)))

    # Return scaled data
    # Radiance = Scale * Image + offset, Reflectance = Radiance * Scale2
    return zip(scale, scale2, offset)

#from skimage.transform._geometric import GeometricTransform
class GeometricTransform():
    ''' Tranforms could inherit from skimage's GeometricTransform base class
        but the interface doesn't provide enough value to justify installing skimage

        Check it out if you need more information about image transforms'''

    pass

class RatPolyTransform(GeometricTransform):
    def __init__(self, A, B, offset, scale, px_offset, px_scale, gsd=None, proj=None, default_z=0):
        self.proj = proj
        self._A = A
        self._B = B
        self._offset = offset
        self._scale = scale
        self._px_offset = px_offset
        self._px_scale = px_scale
        self._gsd = gsd
        self._offscl = np.vstack([offset, scale])
        self._offscl_rev = np.vstack([-offset/scale, 1.0/scale])
        self._px_offscl_rev = np.vstack([px_offset, px_scale])
        self._px_offscl = np.vstack([-px_offset/px_scale, 1.0/px_scale])

        self._default_z = default_z

        self._A_rev = np.dot(pinv(np.dot(np.transpose(A), A)), np.transpose(A))
        # only using the numerator (more dynamic range for the fit?)
        # self._B_rev = np.dot(pinv(np.dot(np.transpose(B), B)), np.transpose(B))

    @property
    def gsd(self):
        return self._gsd

    def rev(self, lng, lat, z=None, _type=np.int32):
        if z is None:
            z = self._default_z

        if all(isinstance(var, (int, float, tuple)) for var in [lng, lat]):
            lng, lat = (np.array([lng]), np.array([lat]))
        if not all(isinstance(var, np.ndarray) for var in [lng, lat]):
            raise ValueError("lng, lat inputs must be of type int, float, tuple or numpy.ndarray")
        if not isinstance(z, np.ndarray):
            z = np.zeros_like(lng) + z
        coord = np.dstack([lng, lat, z])
        offset, scale = np.vsplit(self._offscl, 2)
        normed = coord * scale + offset
        X = self._rpc(normed)
        result = np.rollaxis(np.inner(self._A, X) / np.inner(self._B, X), 0, 3)
        rev_offset, rev_scale = np.vsplit(self._px_offscl_rev, 2)
        # needs to return x/y
        return  np.rint(np.rollaxis(result * rev_scale + rev_offset, 2)).squeeze().astype(_type)[::-1]

    def fwd(self, x, y, z=None):
        if isinstance(x, (Sequence, np.ndarray)):
            if z is None:
                z = [None]*len(x)
            return  np.transpose(np.asarray([self.fwd(x_i, y_i, z_i) for x_i, y_i, z_i in zip(x, y, z)]))
        coord = np.asarray([x, y])
        normed = np.sum(self._px_offscl * np.vstack([np.ones(coord.shape), coord]), axis=0)
        coord = np.dot(self._A_rev, normed)[[1,2,3]] # likely unstable
        return np.sum(self._offscl_rev * np.vstack([np.ones(coord.shape), coord]), axis=0)

    def __call__(self, coords):
        assert isinstance(coords, np.ndarray)
        try:
            d0, d1 = coords.shape
            assert d1 ==2
        except (ValueError, AssertionError):
            raise NotImplementedError("input coords must be [N x 2] dimension numpy array")
        if d1 != 2:
            raise NotImplementedError("input coords must be [N x 2] dimension numpy array")

        xarr, yarr = np.hsplit(coords, 2)
        res = self.fwd(xarr, yarr)
        return res

    def inverse(self, coords):
        pass

    def residuals(self, src, dst):
        pass

    def _rpc(self, x):
        L, P, H = np.dsplit(x, 3)
        return np.dstack([np.ones((x.shape[0], x.shape[1]), dtype=np.float32), L, P, H, L*P, L*H, P*H, L**2, P**2, H**2,
                           L*P*H, L**3, L*(P**2), L*(H**2), (L**2)*P, P**3, P*(H**2),
                           (L**2)*H, (P**2)*H, H**3])

    def __add__(self, other):
        if isinstance(other, Sequence) and len(other) == 2:
            shift = np.asarray(other)
            # shift is an x/y px_offset needs to be y/x
            return RatPolyTransform(self._A, self._B, self._offset, self._scale,
                                    self._px_offset - shift[::-1], self._px_scale,
                                    self.gsd, self.proj, self._default_z)
        else:
            raise NotImplemented

    def __sub__(self, other):
        try:
            return self.__add__(-other)
        except:
            return self.__add__([-e for e in other])

    @classmethod
    def from_rpcs(cls, rpcs):
        P = np.vstack([np.asarray(rpcs["lineNumCoefs"]),
                       np.asarray(rpcs["sampleNumCoefs"])])
        Q = np.vstack([np.asarray(rpcs["lineDenCoefs"]),
                       np.asarray(rpcs["sampleDenCoefs"])])

        scale = np.asarray([1.0/rpcs["lonScale"],
                            1.0/rpcs["latScale"],
                            1.0/rpcs["heightScale"]])
        offset = np.asarray([-rpcs["lonOffset"],
                             -rpcs["latOffset"],
                             -rpcs["heightOffset"]])*scale

        px_scale = np.asarray([rpcs["lineScale"], rpcs["sampleScale"]])
        px_offset = np.asarray([rpcs["lineOffset"], rpcs["sampleOffset"]])

        return cls(P, Q, offset, scale, px_offset, px_scale, rpcs["gsd"], rpcs["spatialReferenceSystem"],
                   rpcs["heightOffset"])

    @classmethod
    def from_affine(cls, affine):
        pass

class AffineTransform(GeometricTransform):
    def __init__(self, affine, proj=None):
        self._affine = affine
        self._iaffine = None
        self.proj = proj

    def rev(self, lng, lat, z=0):
        if self._iaffine is None:
            self._iaffine = ~self._affine
        px, py = (self._iaffine * (lng, lat))
        if type(px).__name__ == 'ndarray' and type(py).__name__ == 'ndarray':
            return np.rint(np.asarray(px)), np.rint(np.asarray(py))
        else:
            return int(round(px)), int(round(py))

    def fwd(self, x, y, z=0):
        return self._affine * (x, y)

    def __call__(self, coords):
        assert isinstance(coords, np.ndarray) and len(coords.shape) == 2 and coords.shape[1] == 2
        _coords = np.copy(coords)
        self._affine.itransform(_coords)
        return _coords

    def inverse(self, coords):
        assert isinstance(coords, np.ndarray) and len(coords.shape) == 2 and coords.shape[1] == 2
        if self._iaffine is None:
            self._iaffine = ~self._affine
        _coords = np.copy(coords)
        self._iaffine.itransform(_coords)
        return _coords

    def residuals(self, src, dst):
        return super(AffineTransform, self).residuals(src, dst)

    def __add__(self, other):
        if isinstance(other, Sequence) and len(other) == 2:
            shift = np.asarray(other)
            return AffineTransform(self._affine * Affine.translation(shift[0], shift[1]), proj=self.proj)
        else:
            raise NotImplemented

    def __sub__(self, other):
        try:
            return self.__add__(-other)
        except:
            return self.__add__([-e for e in other])

    @classmethod
    def from_georef(cls, georef):
        tfm = Affine.from_gdal(georef["translateX"], georef["scaleX"], georef["shearX"],
                               georef["translateY"], georef["shearY"], georef["scaleY"])
        return cls(tfm, proj=georef["spatialReferenceSystemCode"])

def pad_safe_negative(padsize=2, transpix=None, ref_im=None, ind=0):
    trans = transpix[ind,:,:].min() - padsize
    if trans < 0.0:
        trans = transpix[ind,:,:].min()
    return int(math.floor(trans))

def pad_safe_positive(padsize=2, transpix=None, ref_im=None, ind=0):
    trans = transpix[ind,:,:].max() + padsize
    if len(ref_im.shape) == 3:
        critical = ref_im.shape[ind + 1]
    elif len(ref_im.shape) == 2:
        critical = ref_im.shape[ind]
    else:
        raise NotImplementedError("Padding supported only for reference images of shape (L, W) or (Nbands, L, W)")
    if trans > critical:
        return critical
    return int(math.ceil(trans))

max_stars_repo_name: DigitalGlobe/gbdxtools

max_issues_repo_name: DigitalGlobe/gbdxtools

max_forks_repo_name: DigitalGlobe/gbdxtools

reason: import numpy,from numpy


text:
from typing import Any
from typing import Tuple
from typing import List
from typing import Union
from typing import Sequence
from typing import Optional
from typing_extensions import Annotated
from nptyping import NDArray

from dataclasses import dataclass
from dataclasses import astuple

from numpy import asarray
from numpy import float64
from numpy import zeros_like
from scipy.sparse import diags

from compas.numerical import connectivity_matrix

from .result import Result

@dataclass
class FDNumericalData:
    """Stores numerical data used by the force density algorithms."""
    free: int
    fixed: int
    xyz: NDArray[(Any, 3), float64]
    C: NDArray[(Any, Any), int]
    q: NDArray[(Any, 1), float64]
    Q: NDArray[(Any, Any), float64]
    p: NDArray[(Any, 1), float64]
    A: NDArray[(Any, Any), float64]
    Ai: NDArray[(Any, Any), float64]
    Af: NDArray[(Any, Any), float64]
    forces: NDArray[(Any, 1), float64] = None
    lengths: NDArray[(Any, 1), float64] = None
    residuals: NDArray[(Any, 3), float64] = None
    tangent_residuals: NDArray[(Any, 3), float64] = None
    normal_residuals: NDArray[(Any, 1), float64] = None

    def __iter__(self):
        return iter(astuple(self))

    @classmethod
    def from_params(cls,
                    vertices: Union[Sequence[Annotated[List[float], 3]], NDArray[(Any, 3), float64]],
                    fixed: List[int],
                    edges: List[Tuple[int, int]],
                    forcedensities: List[float],
                    loads: Optional[Union[Sequence[Annotated[List[float], 3]], NDArray[(Any, 3), float64]]] = None):
        """Construct numerical arrays from force density solver input parameters."""
        free = list(set(range(len(vertices))) - set(fixed))
        xyz = asarray(vertices, dtype=float64).reshape((-1, 3))
        C = connectivity_matrix(edges, 'csr')
        Ci = C[:, free]
        Cf = C[:, fixed]
        q = asarray(forcedensities, dtype=float64).reshape((-1, 1))
        Q = diags([q.flatten()], [0])
        p = (zeros_like(xyz) if loads is None else
             asarray(loads, dtype=float64).reshape((-1, 3)))
        A = C.T.dot(Q).dot(C)
        Ai = Ci.T.dot(Q).dot(Ci)
        Af = Ci.T.dot(Q).dot(Cf)
        return cls(free, fixed, xyz, C, q, Q, p, A, Ai, Af)

    @classmethod
    def from_mesh(cls, mesh):
        """Construct numerical arrays from input mesh."""
        raise NotImplementedError

    def to_result(self) -> Result:
        """Parse relevant numerical data into a Result object."""
        return Result(self.xyz, self.residuals, self.forces, self.lengths)

reason: from numpy,from scipy