text:
from unittest import TestCase

import numpy as np

from teafacto.blocks.seq.enc import SimpleSeq2Vec, SimpleSeq2Sca, SimpleSeq2MultiVec
from teafacto.blocks.match import MatchScore

class TestSimpleSeq2Vec(TestCase):
    def setUp(self):
        enc = SimpleSeq2Vec(indim=100, inpembdim=10, innerdim=20)
        x = np.random.randint(0, 100, (33, 5))
        o = enc.autobuild(x)
        self.o = o[1][0]
        m = MatchScore(enc, enc)
        mo = m.autobuild(x, x)
        self.mo = mo[1][0]

    def test_output(self):
        print self.mo, self.mo.ndim
        #theano.printing.pydotprint(self.mo.d, "debug.png")

    def test_mask(self):
        np.random.seed(1337)
        enc = SimpleSeq2Vec(indim=100, inpembdim=10, innerdim=4, maskid=-1, layers=2).all_outputs()
        x = np.random.randint(0, 100, (33, 5))
        maskr = np.random.randint(1, x.shape[1], (x.shape[0],))
        for i in range(x.shape[0]):
            x[i, maskr[i]:] = -1
        pred = enc.predict(x)
        print maskr
        print x
        print pred
        print pred.shape

class TestSimpleSeq2Sca(TestCase):
    def test_shape(self):
        enc = SimpleSeq2Sca(indim=100, inpembdim=10, innerdim=20)
        x = np.random.randint(0, 10, (33, 5))
        prd, mask = enc.predict(x)
        self.assertEqual(prd.shape, (33, 5))
        self.assertEqual(mask.shape, (33, 5))

class TestSimpleSeq2MultiVec(TestCase):
    def test_shape(self):
        enc = SimpleSeq2MultiVec(indim=100, inpembdim=10,
                                 innerdim=20, numouts=2, mode="seq")
        x = np.random.randint(0, 100, (33, 5))
        prd = enc.predict(x)
        self.assertEqual(prd.shape, (33, 2, 20))

    def test_bidir_shape(self):
        enc = SimpleSeq2MultiVec(indim=100, inpembdim=10,
                                 innerdim=20, numouts=2, mode="seq",
                                 bidir=True)
        x = np.random.randint(0, 100, (33, 5))
        prd = enc.predict(x)
        self.assertEqual(prd.shape, (33, 2, 40))

max_stars_repo_name: lukovnikov/teafacto

max_issues_repo_name: lukovnikov/teafacto

max_forks_repo_name: lukovnikov/teafacto


text:
import yaml
from pathlib import Path

import cv2
import numpy as np

from nurses_2.colors import AColor

_THIS_DIR = Path(__file__).parent
_INPUTS = _THIS_DIR.parent.parent.parent / "aoc_helper" / "inputs.yaml"
_RAW = yaml.full_load(_INPUTS.read_text())["1"]

DEPTHS = list(map(int, _RAW.splitlines()))
FLOOR_COLOR = AColor.from_hex("050423ff")
SCALE = 10  # Stretch texture horizontally by this scale.

def create_floor_texture():
    """
    Create a texture array from puzzle input using cv2's `fillPoly`.
    """
    min_depth = min(DEPTHS)
    height = max(DEPTHS) - min_depth
    width = len(DEPTHS) * SCALE

    texture = np.zeros((height, width, 4), dtype=np.uint8)
    points = np.array([[SCALE * i, depth - min_depth] for i, depth in enumerate(DEPTHS)] + [[0, height]])
    cv2.fillPoly(texture, [points], FLOOR_COLOR)

    return texture

ext: py

max_stars_repo_name: salt-die/Advent-of-Code

max_issues_repo_name: mpcjanssen/Advent-of-Code

max_forks_repo_name: mpcjanssen/Advent-of-Code

reason: import numpy


text:
#   Copyright (c) 2019 PaddlePaddle Authors. 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.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import print_function

import unittest
import numpy as np
from op_test import OpTest
import paddle
import paddle.fluid as fluid
from paddle.fluid.framework import program_guard, Program

class TestGatherTreeOp(OpTest):
    def setUp(self):
        self.op_type = "gather_tree"
        self.python_api = paddle.nn.functional.gather_tree
        max_length, batch_size, beam_size = 5, 2, 2
        ids = np.random.randint(
            0, high=10, size=(max_length, batch_size, beam_size))
        parents = np.random.randint(
            0, high=beam_size, size=(max_length, batch_size, beam_size))
        self.inputs = {"Ids": ids, "Parents": parents}
        self.outputs = {'Out': self.backtrace(ids, parents)}

    def test_check_output(self):
        self.check_output(check_eager=True)

    @staticmethod
    def backtrace(ids, parents):
        out = np.zeros_like(ids)
        (max_length, batch_size, beam_size) = ids.shape
        for batch in range(batch_size):
            for beam in range(beam_size):
                out[max_length - 1, batch, beam] = ids[max_length - 1, batch,
                                                       beam]
                parent = parents[max_length - 1, batch, beam]
                for step in range(max_length - 2, -1, -1):
                    out[step, batch, beam] = ids[step, batch, parent]
                    parent = parents[step, batch, parent]
        return out

class TestGatherTreeOpAPI(unittest.TestCase):
    def test_case(self):
        paddle.enable_static()
        ids = fluid.layers.data(
            name='ids', shape=[5, 2, 2], dtype='int64', append_batch_size=False)
        parents = fluid.layers.data(
            name='parents',
            shape=[5, 2, 2],
            dtype='int64',
            append_batch_size=False)
        final_sequences = fluid.layers.gather_tree(ids, parents)
        paddle.disable_static()

    def test_case2(self):
        ids = paddle.to_tensor(
            [[[2, 2], [6, 1]], [[3, 9], [6, 1]], [[0, 1], [9, 0]]])
        parents = paddle.to_tensor(
            [[[0, 0], [1, 1]], [[1, 0], [1, 0]], [[0, 0], [0, 1]]])
        final_sequences = paddle.nn.functional.gather_tree(ids, parents)

class TestGatherTreeOpError(unittest.TestCase):
    def test_errors(self):
        paddle.enable_static()
        with program_guard(Program(), Program()):
            ids = fluid.layers.data(
                name='ids',
                shape=[5, 2, 2],
                dtype='int64',
                append_batch_size=False)
            parents = fluid.layers.data(
                name='parents',
                shape=[5, 2, 2],
                dtype='int64',
                append_batch_size=False)

            def test_Variable_ids():
                # the input type must be Variable
                np_ids = np.random.random((5, 2, 2), dtype='int64')
                fluid.layers.gather_tree(np_ids, parents)

            self.assertRaises(TypeError, test_Variable_ids)

            def test_Variable_parents():
                # the input type must be Variable
                np_parents = np.random.random((5, 2, 2), dtype='int64')
                fluid.layers.gather_tree(ids, np_parents)

            self.assertRaises(TypeError, test_Variable_parents)

            def test_type_ids():
                # dtype must be int32 or int64
                bad_ids = fluid.layers.data(
                    name='bad_ids',
                    shape=[5, 2, 2],
                    dtype='float32',
                    append_batch_size=False)
                fluid.layers.gather_tree(bad_ids, parents)

            self.assertRaises(TypeError, test_type_ids)

            def test_type_parents():
                # dtype must be int32 or int64
                bad_parents = fluid.layers.data(
                    name='bad_parents',
                    shape=[5, 2, 2],
                    dtype='float32',
                    append_batch_size=False)
                fluid.layers.gather_tree(ids, bad_parents)

            self.assertRaises(TypeError, test_type_parents)
        paddle.disable_static()

if __name__ == "__main__":
    paddle.enable_static()
    unittest.main()

max_stars_repo_name: DevilCarp/Paddle

max_issues_repo_name: DevilCarp/Paddle

max_forks_repo_name: DevilCarp/Paddle


text:
# Copyright (c) Yuta Saito, Yusuke Narita, and ZOZO Technologies, Inc. All rights reserved.
# Licensed under the Apache 2.0 License.

"""Useful Tools."""
from typing import Dict
from typing import Optional
from typing import Union

import numpy as np
import pandas as pd
from sklearn.utils import check_random_state
from sklearn.utils import check_scalar
import torch

def check_confidence_interval_arguments(
    alpha: float = 0.05,
    n_bootstrap_samples: int = 10000,
    random_state: Optional[int] = None,
) -> Optional[ValueError]:
    """Check confidence interval arguments.

    Parameters
    ----------
    alpha: float, default=0.05
        Significance level.

    n_bootstrap_samples: int, default=10000
        Number of resampling performed in bootstrap sampling.

    random_state: int, default=None
        Controls the random seed in bootstrap sampling.

    Returns
    ----------
    estimated_confidence_interval: Dict[str, float]
        Dictionary storing the estimated mean and upper-lower confidence bounds.

    """
    check_random_state(random_state)
    check_scalar(alpha, "alpha", float, min_val=0.0, max_val=1.0)
    check_scalar(n_bootstrap_samples, "n_bootstrap_samples", int, min_val=1)

def estimate_confidence_interval_by_bootstrap(
    samples: np.ndarray,
    alpha: float = 0.05,
    n_bootstrap_samples: int = 10000,
    random_state: Optional[int] = None,
) -> Dict[str, float]:
    """Estimate confidence interval using bootstrap.

    Parameters
    ----------
    samples: array-like
        Empirical observed samples to be used to estimate cumulative distribution function.

    alpha: float, default=0.05
        Significance level.

    n_bootstrap_samples: int, default=10000
        Number of resampling performed in bootstrap sampling.

    random_state: int, default=None
        Controls the random seed in bootstrap sampling.

    Returns
    ----------
    estimated_confidence_interval: Dict[str, float]
        Dictionary storing the estimated mean and upper-lower confidence bounds.

    """
    check_confidence_interval_arguments(
        alpha=alpha, n_bootstrap_samples=n_bootstrap_samples, random_state=random_state
    )

    boot_samples = list()
    random_ = check_random_state(random_state)
    for _ in np.arange(n_bootstrap_samples):
        boot_samples.append(np.mean(random_.choice(samples, size=samples.shape[0])))
    lower_bound = np.percentile(boot_samples, 100 * (alpha / 2))
    upper_bound = np.percentile(boot_samples, 100 * (1.0 - alpha / 2))
    return {
        "mean": np.mean(boot_samples),
        f"{100 * (1. - alpha)}% CI (lower)": lower_bound,
        f"{100 * (1. - alpha)}% CI (upper)": upper_bound,
    }

def sample_action_fast(
    action_dist: np.ndarray, random_state: Optional[int] = None
) -> np.ndarray:
    """Sample actions faster based on a given action distribution.

    Parameters
    ----------
    action_dist: array-like, shape (n_rounds, n_actions)
        Distribution over actions.

    random_state: Optional[int], default=None
        Controls the random seed in sampling synthetic bandit dataset.

    Returns
    ---------
    sampled_action: array-like, shape (n_rounds,)
        Actions sampled based on `action_dist`.

    """
    random_ = check_random_state(random_state)
    uniform_rvs = random_.uniform(size=action_dist.shape[0])[:, np.newaxis]
    cum_action_dist = action_dist.cumsum(axis=1)
    flg = cum_action_dist > uniform_rvs
    sampled_action = flg.argmax(axis=1)
    return sampled_action

def convert_to_action_dist(
    n_actions: int,
    selected_actions: np.ndarray,
) -> np.ndarray:
    """Convert selected actions (output of `run_bandit_simulation`) to distribution over actions.

    Parameters
    ----------
    n_actions: int
        Number of actions.

    selected_actions: array-like, shape (n_rounds, len_list)
        Sequence of actions selected by evaluation policy
        at each round in offline bandit simulation.

    Returns
    ----------
    action_dist: array-like, shape (n_rounds, n_actions, len_list)
        Action choice probabilities (can be deterministic).

    """
    n_rounds, len_list = selected_actions.shape
    action_dist = np.zeros((n_rounds, n_actions, len_list))
    for pos in np.arange(len_list):
        selected_actions_ = selected_actions[:, pos]
        action_dist[
            np.arange(n_rounds),
            selected_actions_,
            pos * np.ones(n_rounds, int),
        ] = 1
    return action_dist

def check_array(
    array: np.ndarray,
    name: str,
    expected_dim: int = 1,
) -> ValueError:
    """Input validation on an array.

    Parameters
    -------------
    array: object
        Input object to check.

    name: str
        Name of the input array.

    expected_dim: int, default=1
        Expected dimension of the input array.

    """
    if not isinstance(array, np.ndarray):
        raise ValueError(
            f"`{name}` must be {expected_dim}D array, but got {type(array)}"
        )
    if array.ndim != expected_dim:
        raise ValueError(
            f"`{name}` must be {expected_dim}D array, but got {array.ndim}D array"
        )

def check_tensor(
    tensor: torch.tensor,
    name: str,
    expected_dim: int = 1,
) -> ValueError:
    """Input validation on a tensor.

    Parameters
    -------------
    array: object
        Input object to check.

    name: str
        Name of the input array.

    expected_dim: int, default=1
        Expected dimension of the input array.

    """
    if not isinstance(tensor, torch.Tensor):
        raise ValueError(
            f"`{name}` must be {expected_dim}D tensor, but got {type(tensor)}"
        )
    if tensor.ndim != expected_dim:
        raise ValueError(
            f"`{name}` must be {expected_dim}D tensor, but got {tensor.ndim}D tensor"
        )

def check_bandit_feedback_inputs(
    context: np.ndarray,
    action: np.ndarray,
    reward: np.ndarray,
    expected_reward: Optional[np.ndarray] = None,
    position: Optional[np.ndarray] = None,
    pscore: Optional[np.ndarray] = None,
    action_context: Optional[np.ndarray] = None,
) -> Optional[ValueError]:
    """Check inputs for bandit learning or simulation.

    Parameters
    -----------
    context: array-like, shape (n_rounds, dim_context)
        Context vectors observed for each data, i.e., :math:`x_i`.

    action: array-like, shape (n_rounds,)
        Actions sampled by the logging/behavior policy for each data in logged bandit data, i.e., :math:`a_i`.

    reward: array-like, shape (n_rounds,)
        Rewards observed for each data in logged bandit data, i.e., :math:`r_i`.

    expected_reward: array-like, shape (n_rounds, n_actions), default=None
        Expected reward of each data, i.e., :math:`\\mathbb{E}[r_i|x_i,a_i]`.

    position: array-like, shape (n_rounds,), default=None
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    pscore: array-like, shape (n_rounds,)
        Action choice probabilities of the logging/behavior policy (propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    action_context: array-like, shape (n_actions, dim_action_context)
        Context vectors characterizing each action.

    """
    check_array(array=context, name="context", expected_dim=2)
    check_array(array=action, name="action", expected_dim=1)
    check_array(array=reward, name="reward", expected_dim=1)
    if expected_reward is not None:
        check_array(array=expected_reward, name="expected_reward", expected_dim=2)
        if not (
            context.shape[0]
            == action.shape[0]
            == reward.shape[0]
            == expected_reward.shape[0]
        ):
            raise ValueError(
                "Expected `context.shape[0] == action.shape[0] == reward.shape[0] == expected_reward.shape[0]`"
                ", but found it False"
            )
        if not (
            np.issubdtype(action.dtype, np.integer)
            and action.min() >= 0
            and action.max() < expected_reward.shape[1]
        ):
            raise ValueError(
                "`action` elements must be integers in the range of [0, `expected_reward.shape[1]`)"
            )
    else:
        if not (np.issubdtype(action.dtype, np.integer) and action.min() >= 0):
            raise ValueError("`action` elements must be non-negative integers")
    if pscore is not None:
        check_array(array=pscore, name="pscore", expected_dim=1)
        if not (
            context.shape[0] == action.shape[0] == reward.shape[0] == pscore.shape[0]
        ):
            raise ValueError(
                "Expected `context.shape[0] == action.shape[0] == reward.shape[0] == pscore.shape[0]`"
                ", but found it False"
            )
        if np.any(pscore <= 0):
            raise ValueError("`pscore` must be positive")

    if position is not None:
        check_array(array=position, name="position", expected_dim=1)
        if not (
            context.shape[0] == action.shape[0] == reward.shape[0] == position.shape[0]
        ):
            raise ValueError(
                "Expected `context.shape[0] == action.shape[0] == reward.shape[0] == position.shape[0]`"
                ", but found it False"
            )
        if not (np.issubdtype(position.dtype, np.integer) and position.min() >= 0):
            raise ValueError("`position` elements must be non-negative integers")
    else:
        if not (context.shape[0] == action.shape[0] == reward.shape[0]):
            raise ValueError(
                "Expected `context.shape[0] == action.shape[0] == reward.shape[0]`"
                ", but found it False"
            )
    if action_context is not None:
        check_array(array=action_context, name="action_context", expected_dim=2)
        if not (
            np.issubdtype(action.dtype, np.integer)
            and action.min() >= 0
            and action.max() < action_context.shape[0]
        ):
            raise ValueError(
                "`action` elements must be integers in the range of [0, `action_context.shape[0]`)"
            )
    else:
        if not (np.issubdtype(action.dtype, np.integer) and action.min() >= 0):
            raise ValueError("`action` elements must be non-negative integers")

def check_ope_inputs(
    action_dist: np.ndarray,
    position: Optional[np.ndarray] = None,
    action: Optional[np.ndarray] = None,
    reward: Optional[np.ndarray] = None,
    pscore: Optional[np.ndarray] = None,
    estimated_rewards_by_reg_model: Optional[np.ndarray] = None,
    estimated_importance_weights: Optional[np.ndarray] = None,
) -> Optional[ValueError]:
    """Check inputs for ope.

    Parameters
    -----------
    action_dist: array-like, shape (n_rounds, n_actions, len_list)
        Action choice probabilities of the evaluation policy (can be deterministic), i.e., :math:`\\pi_e(a_i|x_i)`.

    position: array-like, shape (n_rounds,), default=None
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    action: array-like, shape (n_rounds,), default=None
        Actions sampled by the logging/behavior policy for each data in logged bandit data, i.e., :math:`a_i`.

    reward: array-like, shape (n_rounds,), default=None
        Rewards observed for each data in logged bandit data, i.e., :math:`r_i`.

    pscore: array-like, shape (n_rounds,)
        Action choice probabilities of the logging/behavior policy (propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    estimated_rewards_by_reg_model: array-like, shape (n_rounds, n_actions, len_list), default=None
        Estimated expected rewards given context, action, and position, i.e., :math:`\\hat{q}(x_i,a_i)`.

    estimated_importance_weights: array-like, shape (n_rounds,), default=None
        Importance weights estimated via supervised classification, i.e., :math:`\\hat{w}(x_t, a_t)`.

    """
    # action_dist
    check_array(array=action_dist, name="action_dist", expected_dim=3)
    if not np.allclose(action_dist.sum(axis=1), 1):
        raise ValueError("`action_dist` must be a probability distribution")

    # position
    if position is not None:
        check_array(array=position, name="position", expected_dim=1)
        if not (position.shape[0] == action_dist.shape[0]):
            raise ValueError(
                "Expected `position.shape[0] == action_dist.shape[0]`, but found it False"
            )
        if not (np.issubdtype(position.dtype, np.integer) and position.min() >= 0):
            raise ValueError("`position` elements must be non-negative integers")
        if position.max() >= action_dist.shape[2]:
            raise ValueError(
                "`position` elements must be smaller than `action_dist.shape[2]`"
            )
    elif action_dist.shape[2] > 1:
        raise ValueError(
            "`position` elements must be given when `action_dist.shape[2] > 1`"
        )

    # estimated_rewards_by_reg_model
    if estimated_rewards_by_reg_model is not None:
        if estimated_rewards_by_reg_model.shape != action_dist.shape:
            raise ValueError(
                "Expected `estimated_rewards_by_reg_model.shape == action_dist.shape`, but found it False"
            )

    if estimated_importance_weights is not None:
        if not (action.shape[0] == estimated_importance_weights.shape[0]):
            raise ValueError(
                "Expected `action.shape[0] == estimated_importance_weights.shape[0]`, but found it False"
            )
        if np.any(estimated_importance_weights < 0):
            raise ValueError("estimated_importance_weights must be non-negative")

    # action, reward
    if action is not None or reward is not None:
        check_array(array=action, name="action", expected_dim=1)
        check_array(array=reward, name="reward", expected_dim=1)
        if not (action.shape[0] == reward.shape[0]):
            raise ValueError(
                "Expected `action.shape[0] == reward.shape[0]`, but found it False"
            )
        if not (
            np.issubdtype(action.dtype, np.integer)
            and action.min() >= 0
            and action.max() < action_dist.shape[1]
        ):
            raise ValueError(
                "`action` elements must be integers in the range of [0, `action_dist.shape[1]`)"
            )

    # pscore
    if pscore is not None:
        if pscore.ndim != 1:
            raise ValueError("pscore must be 1-dimensional")
        if not (action.shape[0] == reward.shape[0] == pscore.shape[0]):
            raise ValueError(
                "Expected `action.shape[0] == reward.shape[0] == pscore.shape[0]`, but found it False"
            )
        if np.any(pscore <= 0):
            raise ValueError("`pscore` must be positive")

def check_multi_loggers_ope_inputs(
    action_dist: np.ndarray,
    position: Optional[np.ndarray] = None,
    action: Optional[np.ndarray] = None,
    reward: Optional[np.ndarray] = None,
    stratum_idx: Optional[np.ndarray] = None,
    pscore: Optional[np.ndarray] = None,
    estimated_rewards_by_reg_model: Optional[np.ndarray] = None,
) -> Optional[ValueError]:
    """Check inputs for ope with multiple loggers.

    Parameters
    -----------
    action_dist: array-like, shape (n_rounds, n_actions, len_list)
        Action choice probabilities of the evaluation policy (can be deterministic), i.e., :math:`\\pi_e(a_i|x_i)`.

    position: array-like, shape (n_rounds,), default=None
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    action: array-like, shape (n_rounds,), default=None
        Actions sampled by the logging/behavior policy for each data in logged bandit data, i.e., :math:`a_i`.

    reward: array-like, shape (n_rounds,), default=None
        Rewards observed for each data in logged bandit data, i.e., :math:`r_i`.

    stratum_idx: array-like, shape (n_rounds,)
        Indices to differentiate the logging/behavior policy that generate each data, i.e., :math:`k`.

    pscore: array-like, shape (n_rounds,)
        Action choice probabilities of the logging/behavior policy (propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    estimated_rewards_by_reg_model: array-like, shape (n_rounds, n_actions, len_list), default=None
        Estimated expected rewards given context, action, and position, i.e., :math:`\\hat{q}(x_i,a_i)`.

    """
    check_ope_inputs(
        action_dist=action_dist,
        position=position,
        action=action,
        reward=reward,
        pscore=pscore,
        estimated_rewards_by_reg_model=estimated_rewards_by_reg_model,
    )

    # stratum idx
    if stratum_idx is not None:
        check_array(array=stratum_idx, name="stratum_idx", expected_dim=1)
        if not (action_dist.shape[0] == stratum_idx.shape[0]):
            raise ValueError(
                "Expected `action_dist.shape[0] == stratum_idx.shape[0]`, but found it False"
            )
        if not (
            np.issubdtype(stratum_idx.dtype, np.integer) and stratum_idx.min() >= 0
        ):
            raise ValueError("`stratum_idx` elements must be non-negative integers")

def check_continuous_bandit_feedback_inputs(
    context: np.ndarray,
    action_by_behavior_policy: np.ndarray,
    reward: np.ndarray,
    expected_reward: Optional[np.ndarray] = None,
    pscore: Optional[np.ndarray] = None,
) -> Optional[ValueError]:
    """Check inputs for bandit learning or simulation with continuous actions.

    Parameters
    -----------
    context: array-like, shape (n_rounds, dim_context)
        Context vectors observed for each data, i.e., :math:`x_i`.

    action_by_behavior_policy: array-like, shape (n_rounds,)
        Continuous action values sampled by the logging/behavior policy for each data in logged bandit data, i.e., :math:`a_i`.

    reward: array-like, shape (n_rounds,)
        Rewards observed for each data in logged bandit data, i.e., :math:`r_i`.

    expected_reward: array-like, shape (n_rounds, n_actions), default=None
        Expected reward of each data, i.e., :math:`\\mathbb{E}[r_i|x_i,a_i]`.

    pscore: array-like, shape (n_rounds,), default=None
        Probability densities of the continuous action values sampled by the logging/behavior policy
        (generalized propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    """
    check_array(array=context, name="context", expected_dim=2)
    check_array(
        array=action_by_behavior_policy,
        name="action_by_behavior_policy",
        expected_dim=1,
    )
    check_array(array=reward, name="reward", expected_dim=1)

    if expected_reward is not None:
        check_array(array=expected_reward, name="expected_reward", expected_dim=1)
        if not (
            context.shape[0]
            == action_by_behavior_policy.shape[0]
            == reward.shape[0]
            == expected_reward.shape[0]
        ):
            raise ValueError(
                "Expected `context.shape[0] == action_by_behavior_policy.shape[0]"
                "== reward.shape[0] == expected_reward.shape[0]`, but found it False"
            )
    if pscore is not None:
        check_array(array=pscore, name="pscore", expected_dim=1)
        if not (
            context.shape[0]
            == action_by_behavior_policy.shape[0]
            == reward.shape[0]
            == pscore.shape[0]
        ):
            raise ValueError(
                "Expected `context.shape[0] == action_by_behavior_policy.shape[0]"
                "== reward.shape[0] == pscore.shape[0]`, but found it False"
            )
        if np.any(pscore <= 0):
            raise ValueError("`pscore` must be positive")

def check_continuous_ope_inputs(
    action_by_evaluation_policy: np.ndarray,
    action_by_behavior_policy: Optional[np.ndarray] = None,
    reward: Optional[np.ndarray] = None,
    pscore: Optional[np.ndarray] = None,
    estimated_rewards_by_reg_model: Optional[np.ndarray] = None,
) -> Optional[ValueError]:
    """Check inputs for OPE with continuous actions.

    Parameters
    -----------
    action_by_evaluation_policy: array-like, shape (n_rounds,)
        Continuous action values given by the evaluation policy (can be deterministic), i.e., :math:`\\pi_e(x_t)`.

    action_by_behavior_policy: array-like, shape (n_rounds,), default=None
        Continuous action values sampled by the logging/behavior policy for each data in logged bandit data, i.e., :math:`a_i`.

    reward: array-like, shape (n_rounds,), default=None
        Rewards observed for each data in logged bandit data, i.e., :math:`r_i`.

    pscore: array-like, shape (n_rounds,), default=None
        Probability densities of the continuous action values sampled by the logging/behavior policy
        (generalized propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    estimated_rewards_by_reg_model: array-like, shape (n_rounds,), default=None
        Expected rewards given context and action estimated by a regression model, i.e., :math:`\\hat{q}(x_i,a_i)`.

    """
    # action_by_evaluation_policy
    check_array(
        array=action_by_evaluation_policy,
        name="action_by_evaluation_policy",
        expected_dim=1,
    )

    # estimated_rewards_by_reg_model
    if estimated_rewards_by_reg_model is not None:
        check_array(
            array=estimated_rewards_by_reg_model,
            name="estimated_rewards_by_reg_model",
            expected_dim=1,
        )
        if (
            estimated_rewards_by_reg_model.shape[0]
            != action_by_evaluation_policy.shape[0]
        ):
            raise ValueError(
                "Expected `estimated_rewards_by_reg_model.shape[0] == action_by_evaluation_policy.shape[0]`"
                ", but found if False"
            )

    # action, reward
    if action_by_behavior_policy is not None or reward is not None:
        check_array(
            array=action_by_behavior_policy,
            name="action_by_behavior_policy",
            expected_dim=1,
        )
        check_array(array=reward, name="reward", expected_dim=1)
        if not (action_by_behavior_policy.shape[0] == reward.shape[0]):
            raise ValueError(
                "Expected `action_by_behavior_policy.shape[0] == reward.shape[0]`"
                ", but found it False"
            )
        if not (
            action_by_behavior_policy.shape[0] == action_by_evaluation_policy.shape[0]
        ):
            raise ValueError(
                "Expected `action_by_behavior_policy.shape[0] == action_by_evaluation_policy.shape[0]`"
                ", but found it False"
            )

    # pscore
    if pscore is not None:
        check_array(array=pscore, name="pscore", expected_dim=1)
        if not (
            action_by_behavior_policy.shape[0] == reward.shape[0] == pscore.shape[0]
        ):
            raise ValueError(
                "Expected `action_by_behavior_policy.shape[0] == reward.shape[0] == pscore.shape[0]`"
                ", but found it False"
            )
        if np.any(pscore <= 0):
            raise ValueError("`pscore` must be positive")

def _check_slate_ope_inputs(
    slate_id: np.ndarray,
    reward: np.ndarray,
    position: np.ndarray,
    pscore: np.ndarray,
    evaluation_policy_pscore: np.ndarray,
    pscore_type: str,
) -> Optional[ValueError]:
    """Check inputs of Slate OPE estimators.

    Parameters
    -----------
    slate_id: array-like, shape (<= n_rounds * len_list,)
        Slate id observed for each data in logged bandit data.

    reward: array-like, shape (<= n_rounds * len_list,)
        Slot-level rewards, i.e., :math:`r_{i}(l)`.

    position: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    pscore: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities of the logging/behavior policy (propensity scores).

    evaluation_policy_pscore: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities of the evaluation policy.

    pscore_type: str
        Either "pscore", "pscore_item_position", or "pscore_cascade".

    """
    # position
    check_array(array=position, name="position", expected_dim=1)
    if not (position.dtype == int and position.min() >= 0):
        raise ValueError("`position` elements must be non-negative integers")

    # reward
    check_array(array=reward, name="reward", expected_dim=1)

    # pscore
    check_array(array=pscore, name=f"{pscore_type}", expected_dim=1)
    if np.any(pscore <= 0) or np.any(pscore > 1):
        raise ValueError(f"`{pscore_type}` must be in the range of (0, 1]")

    # evaluation_policy_pscore
    check_array(
        array=evaluation_policy_pscore,
        name=f"evaluation_policy_{pscore_type}",
        expected_dim=1,
    )
    if np.any(evaluation_policy_pscore < 0) or np.any(evaluation_policy_pscore > 1):
        raise ValueError(
            f"`evaluation_policy_{pscore_type}` must be in the range of [0, 1]"
        )

    # slate id
    check_array(array=slate_id, name="slate_id", expected_dim=1)
    if not (slate_id.dtype == int and slate_id.min() >= 0):
        raise ValueError("slate_id elements must be non-negative integers")
    if not (
        slate_id.shape[0]
        == position.shape[0]
        == reward.shape[0]
        == pscore.shape[0]
        == evaluation_policy_pscore.shape[0]
    ):
        raise ValueError(
            f"`slate_id`, `position`, `reward`, `{pscore_type}`, and `evaluation_policy_{pscore_type}` "
            "must have the same number of samples."
        )

def check_sips_inputs(
    slate_id: np.ndarray,
    reward: np.ndarray,
    position: np.ndarray,
    pscore: np.ndarray,
    evaluation_policy_pscore: np.ndarray,
) -> Optional[ValueError]:
    """Check inputs of SlateStandardIPS.

    Parameters
    -----------
    slate_id: array-like, shape (<= n_rounds * len_list,)
        Slate id observed for each data in logged bandit data.

    reward: array-like, shape (<= n_rounds * len_list,)
        Slot-level rewards, i.e., :math:`r_{i}(l)`.

    position: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    pscore: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities of the logging/behavior policy (propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    evaluation_policy_pscore: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities of the evaluation policy, i.e., :math:`\\pi_e(a_i|x_i)`.

    """
    _check_slate_ope_inputs(
        slate_id=slate_id,
        reward=reward,
        position=position,
        pscore=pscore,
        evaluation_policy_pscore=evaluation_policy_pscore,
        pscore_type="pscore",
    )

    bandit_feedback_df = pd.DataFrame()
    bandit_feedback_df["slate_id"] = slate_id
    bandit_feedback_df["reward"] = reward
    bandit_feedback_df["position"] = position
    bandit_feedback_df["pscore"] = pscore
    bandit_feedback_df["evaluation_policy_pscore"] = evaluation_policy_pscore
    # check uniqueness
    if bandit_feedback_df.duplicated(["slate_id", "position"]).sum() > 0:
        raise ValueError("`position` must not be duplicated in each slate")
    # check pscore uniqueness
    distinct_count_pscore_in_slate = bandit_feedback_df.groupby("slate_id").apply(
        lambda x: x["pscore"].unique().shape[0]
    )
    if (distinct_count_pscore_in_slate != 1).sum() > 0:
        raise ValueError("`pscore` must be unique in each slate")
    # check pscore uniqueness of evaluation policy
    distinct_count_evaluation_policy_pscore_in_slate = bandit_feedback_df.groupby(
        "slate_id"
    ).apply(lambda x: x["evaluation_policy_pscore"].unique().shape[0])
    if (distinct_count_evaluation_policy_pscore_in_slate != 1).sum() > 0:
        raise ValueError("`evaluation_policy_pscore` must be unique in each slate")

def check_iips_inputs(
    slate_id: np.ndarray,
    reward: np.ndarray,
    position: np.ndarray,
    pscore_item_position: np.ndarray,
    evaluation_policy_pscore_item_position: np.ndarray,
) -> Optional[ValueError]:
    """Check inputs of SlateIndependentIPS.

    Parameters
    -----------
    slate_id: array-like, shape (<= n_rounds * len_list,)
        Slate id observed for each data in logged bandit data.

    reward: array-like, shape (<= n_rounds * len_list,)
        Slot-level rewards, i.e., :math:`r_{i}(l)`.

    position: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    pscore_item_position: array-like, shape (<= n_rounds * len_list,)
        Marginal action choice probabilities of the slot (:math:`l`) by a behavior policy (propensity scores), i.e., :math:`\\pi_b(a_{i}(l) |x_i)`.

    evaluation_policy_pscore_item_position: array-like, shape (<= n_rounds * len_list,)
        Marginal action choice probabilities of the slot (:math:`l`) by the evaluation policy, i.e., :math:`\\pi_e(a_{i}(l) |x_i)`.

    """
    _check_slate_ope_inputs(
        slate_id=slate_id,
        reward=reward,
        position=position,
        pscore=pscore_item_position,
        evaluation_policy_pscore=evaluation_policy_pscore_item_position,
        pscore_type="pscore_item_position",
    )

    bandit_feedback_df = pd.DataFrame()
    bandit_feedback_df["slate_id"] = slate_id
    bandit_feedback_df["position"] = position
    # check uniqueness
    if bandit_feedback_df.duplicated(["slate_id", "position"]).sum() > 0:
        raise ValueError("`position` must not be duplicated in each slate")

def check_rips_inputs(
    slate_id: np.ndarray,
    reward: np.ndarray,
    position: np.ndarray,
    pscore_cascade: np.ndarray,
    evaluation_policy_pscore_cascade: np.ndarray,
) -> Optional[ValueError]:
    """Check inputs of SlateRewardInteractionIPS.

    Parameters
    -----------
    slate_id: array-like, shape (<= n_rounds * len_list,)
        Slate id observed for each data in logged bandit data.

    reward: array-like, shape (<= n_rounds * len_list,)
        Slot-level rewards, i.e., :math:`r_{i}(l)`.

    position: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate positions in a recommendation interface where the actions are presented.

    pscore_cascade: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities of the logging/behavior policy (propensity scores), i.e., :math:`\\pi_b(a_i|x_i)`.

    evaluation_policy_pscore_cascade: array-like, shape (<= n_rounds * len_list,)
        Action choice probabilities above the slot (:math:`l`) by the evaluation policy, i.e., :math:`\\pi_e(\\{a_{t, j}\\}_{j \\le k}|x_t)`.

    """
    _check_slate_ope_inputs(
        slate_id=slate_id,
        reward=reward,
        position=position,
        pscore=pscore_cascade,
        evaluation_policy_pscore=evaluation_policy_pscore_cascade,
        pscore_type="pscore_cascade",
    )

    bandit_feedback_df = pd.DataFrame()
    bandit_feedback_df["slate_id"] = slate_id
    bandit_feedback_df["reward"] = reward
    bandit_feedback_df["position"] = position
    bandit_feedback_df["pscore_cascade"] = pscore_cascade
    bandit_feedback_df[
        "evaluation_policy_pscore_cascade"
    ] = evaluation_policy_pscore_cascade
    # sort dataframe
    bandit_feedback_df = (
        bandit_feedback_df.sort_values(["slate_id", "position"])
        .reset_index(drop=True)
        .copy()
    )
    # check uniqueness
    if bandit_feedback_df.duplicated(["slate_id", "position"]).sum() > 0:
        raise ValueError("`position` must not be duplicated in each slate")
    # check pscore_cascade structure
    previous_minimum_pscore_cascade = (
        bandit_feedback_df.groupby("slate_id")["pscore_cascade"]
        .expanding()
        .min()
        .values
    )
    if (
        previous_minimum_pscore_cascade < bandit_feedback_df["pscore_cascade"]
    ).sum() > 0:
        raise ValueError(
            "`pscore_cascade` must be non-increasing sequence in each slate"
        )
    # check pscore_cascade structure of evaluation policy
    previous_minimum_evaluation_policy_pscore_cascade = (
        bandit_feedback_df.groupby("slate_id")["evaluation_policy_pscore_cascade"]
        .expanding()
        .min()
        .values
    )
    if (
        previous_minimum_evaluation_policy_pscore_cascade
        < bandit_feedback_df["evaluation_policy_pscore_cascade"]
    ).sum() > 0:
        raise ValueError(
            "`evaluation_policy_pscore_cascade` must be non-increasing sequence in each slate"
        )

def check_cascade_dr_inputs(
    n_unique_action: int,
    slate_id: np.ndarray,
    action: np.ndarray,
    reward: np.ndarray,
    position: np.ndarray,
    pscore_cascade: np.ndarray,
    evaluation_policy_pscore_cascade: np.ndarray,
    q_hat: np.ndarray,
    evaluation_policy_action_dist: np.ndarray,
) -> Optional[ValueError]:
    """Check inputs of SlateCascadeDoublyRobust.

    Parameters
    -----------
    n_unique_action: int
        Number of unique actions.

    slate_id: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate slates (i.e., ranking or list of actions)

    action: array-like, (<= n_rounds * len_list,)
        Actions observed at each slot in a ranking/slate in logged bandit data, i.e., :math:`a_{i}(l)`,
        which is chosen by the behavior policy :math:`\\pi_b`.

    reward: array-like, shape (<= n_rounds * len_list,)
        Slot-level rewards observed for each data in logged bandit data, i.e., :math:`r_{i}(l)`.

    position: array-like, shape (<= n_rounds * len_list,)
        Indices to differentiate slots/positions in a slate/ranking.

    pscore_cascade: array-like, shape (<= n_rounds * len_list,)
        Probabilities of behavior policy selecting action :math:`a` at position (slot) `k` conditional on the previous actions (presented at position `1` to `k-1`)
        , i.e., :math:`\\pi_b(a_i(l) | x_i, a_i(1), \\ldots, a_i(l-1))`.

    evaluation_policy_pscore_cascade: array-like, shape (<= n_rounds * len_list,)
        Probabilities of evaluation policy selecting action :math:`a` at position (slot) `k` conditional on the previous actions (presented at position `1` to `k-1`)
        , i.e., :math:`\\pi_e(a_i(l) | x_i, a_i(1), \\ldots, a_i(l-1))`.

    q_hat: array-like (<= n_rounds * len_list * n_unique_actions, )
        :math:`\\hat{Q}_l` used in Cascade-DR.
        , i.e., :math:`\\hat{Q}_{i,l}(x_i, a_i(1), \\ldots, a_i(l-1), a_i(l)) \\forall a_i(l) \\in \\mathcal{A}`.

    evaluation_policy_action_dist: array-like (<= n_rounds * len_list * n_unique_actions, )
        Action choice probabilities of the evaluation policy for all possible actions
        , i.e., :math:`\\pi_e(a_i(l) | x_i, a_i(1), \\ldots, a_i(l-1)) \\forall a_i(l) \\in \\mathcal{A}`.

    """
    check_rips_inputs(
        slate_id=slate_id,
        reward=reward,
        position=position,
        pscore_cascade=pscore_cascade,
        evaluation_policy_pscore_cascade=evaluation_policy_pscore_cascade,
    )
    check_array(array=action, name="action", expected_dim=1)
    check_array(
        array=q_hat,
        name="q_hat",
        expected_dim=1,
    )
    check_array(
        array=evaluation_policy_action_dist,
        name="evaluation_policy_action_dist",
        expected_dim=1,
    )
    if not (
        np.issubdtype(action.dtype, np.integer)
        and action.min() >= 0
        and action.max() < n_unique_action
    ):
        raise ValueError(
            "`action` elements must be integers in the range of [0, n_unique_action)"
        )
    if not (
        slate_id.shape[0]
        == action.shape[0]
        == q_hat.shape[0] // n_unique_action
        == evaluation_policy_action_dist.shape[0] // n_unique_action
    ):
        raise ValueError(
            "Expected `slate_id.shape[0] == action.shape[0] == "
            "q_hat.shape[0] // n_unique_action == evaluation_policy_action_dist.shape[0] // n_unique_action`, "
            "but found it False"
        )
    evaluation_policy_action_dist_ = evaluation_policy_action_dist.reshape(
        (-1, n_unique_action)
    )
    if not np.allclose(
        np.ones(evaluation_policy_action_dist_.shape[0]),
        evaluation_policy_action_dist_.sum(axis=1),
    ):
        raise ValueError(
            "`evaluation_policy_action_dist[i * n_unique_action : (i+1) * n_unique_action]` "
            "must sum up to one for all i."
        )

def sigmoid(x: Union[float, np.ndarray]) -> Union[float, np.ndarray]:
    """Calculate sigmoid function."""
    return np.exp(np.minimum(x, 0)) / (1.0 + np.exp(-np.abs(x)))

def softmax(x: Union[float, np.ndarray]) -> Union[float, np.ndarray]:
    """Calculate softmax function."""
    b = np.max(x, axis=1)[:, np.newaxis]
    numerator = np.exp(x - b)
    denominator = np.sum(numerator, axis=1)[:, np.newaxis]
    return numerator / denominator


text:
"""Construct Map Machine nodes and ways."""
import logging
import sys
from datetime import datetime
from hashlib import sha256
from typing import Any, Iterator, Optional, Union

import numpy as np
from colour import Color

from map_machine.color import get_gradient_color
from map_machine.feature.building import Building, BUILDING_SCALE
from map_machine.feature.crater import Crater
from map_machine.feature.direction import DirectionSector
from map_machine.feature.road import Road, Roads
from map_machine.feature.tree import Tree
from map_machine.figure import StyledFigure
from map_machine.geometry.flinger import Flinger
from map_machine.map_configuration import DrawingMode, MapConfiguration
from map_machine.osm.osm_reader import (
    OSMData,
    OSMNode,
    OSMRelation,
    OSMWay,
    parse_levels,
    Tags,
)
from map_machine.pictogram.icon import (
    DEFAULT_SMALL_SHAPE_ID,
    Icon,
    IconSet,
    Shape,
    ShapeExtractor,
    ShapeSpecification,
)
from map_machine.pictogram.point import Point
from map_machine.scheme import LineStyle, RoadMatcher, Scheme
from map_machine.text import Label, TextConstructor
from map_machine.ui.cli import BuildingMode
from map_machine.util import MinMax

__author__ = "Sergey Vartanov"
__email__ = "me@enzet.ru"

DEBUG: bool = False
TIME_COLOR_SCALE: list[Color] = [
    Color("#581845"),
    Color("#900C3F"),
    Color("#C70039"),
    Color("#FF5733"),
    Color("#FFC300"),
    Color("#DAF7A6"),
]

def line_center(
    nodes: list[OSMNode], flinger: Flinger
) -> (np.ndarray, np.ndarray):
    """
    Get geometric center of nodes set.

    :param nodes: node list
    :param flinger: flinger that remap geo positions
    """
    boundary: list[MinMax] = [MinMax(), MinMax()]

    for node in nodes:
        boundary[0].update(node.coordinates[0])
        boundary[1].update(node.coordinates[1])
    center_coordinates: np.ndarray = np.array(
        (boundary[0].center(), boundary[1].center())
    )
    return flinger.fling(center_coordinates), center_coordinates

def get_user_color(text: str, seed: str) -> Color:
    """
    Generate random color based on text.
    """
    if text == "":
        return Color("black")
    return Color("#" + sha256((seed + text).encode("utf-8")).hexdigest()[-6:])

def get_time_color(time: Optional[datetime], boundaries: MinMax) -> Color:
    """
    Generate color based on time.

    :param time: current element creation time
    :param boundaries: minimum and maximum element creation time on the map
    """
    return get_gradient_color(
        time if time else boundaries.max_, boundaries, TIME_COLOR_SCALE
    )

def glue(ways: list[OSMWay]) -> list[list[OSMNode]]:
    """
    Try to glue ways that share nodes.

    :param ways: ways to glue
    """
    result: list[list[OSMNode]] = []
    to_process: set[tuple[OSMNode]] = set()

    for way in ways:
        if way.is_cycle():
            result.append(way.nodes)
        else:
            to_process.add(tuple(way.nodes))

    while to_process:
        nodes: list[OSMNode] = list(to_process.pop())
        glued: Optional[list[OSMNode]] = None
        other_nodes: Optional[tuple[OSMNode]] = None

        for other_nodes in to_process:
            glued = try_to_glue(nodes, list(other_nodes))
            if glued is not None:
                break

        if glued is not None:
            to_process.remove(other_nodes)
            if is_cycle(glued):
                result.append(glued)
            else:
                to_process.add(tuple(glued))
        else:
            result.append(nodes)

    return result

def is_cycle(nodes: list[OSMNode]) -> bool:
    """Is way a cycle way or an area boundary."""
    return nodes[0] == nodes[-1]

def try_to_glue(
    nodes: list[OSMNode], other: list[OSMNode]
) -> Optional[list[OSMNode]]:
    """Create new combined way if ways share endpoints."""
    if nodes[0] == other[0]:
        return list(reversed(other[1:])) + nodes
    if nodes[0] == other[-1]:
        return other[:-1] + nodes
    if nodes[-1] == other[-1]:
        return nodes + list(reversed(other[:-1]))
    if nodes[-1] == other[0]:
        return nodes + other[1:]
    return None

class Constructor:
    """Map Machine node and way constructor."""

    def __init__(
        self,
        osm_data: OSMData,
        flinger: Flinger,
        scheme: Scheme,
        extractor: ShapeExtractor,
        configuration: MapConfiguration,
    ) -> None:
        self.osm_data: OSMData = osm_data
        self.flinger: Flinger = flinger
        self.scheme: Scheme = scheme
        self.extractor: ShapeExtractor = extractor
        self.configuration: MapConfiguration = configuration
        self.text_constructor: TextConstructor = TextConstructor(self.scheme)

        if self.configuration.level == "all":
            self.check_level = lambda x: True
        elif self.configuration.level == "overground":
            self.check_level = check_level_overground
        elif self.configuration.level == "underground":
            self.check_level = lambda x: not check_level_overground(x)
        else:
            self.check_level = lambda x: check_level_number(
                x, float(self.configuration.level)
            )

        self.points: list[Point] = []
        self.figures: list[StyledFigure] = []
        self.buildings: list[Building] = []
        self.roads: Roads = Roads()
        self.trees: list[Tree] = []
        self.craters: list[Crater] = []
        self.direction_sectors: list[DirectionSector] = []

        self.heights: set[float] = {0.25 / BUILDING_SCALE, 0.5 / BUILDING_SCALE}

    def add_building(self, building: Building) -> None:
        """Add building and update levels."""
        self.buildings.append(building)
        self.heights.add(building.height)
        self.heights.add(building.min_height)

    def construct(self) -> None:
        """Construct nodes, ways, and relations."""
        self.construct_ways()
        self.construct_relations()
        self.construct_nodes()

    def construct_ways(self) -> None:
        """Construct Map Machine ways."""
        logging.info("Constructing ways...")
        for way_id in self.osm_data.ways:
            way: OSMWay = self.osm_data.ways[way_id]
            self.construct_line(way, [], [way.nodes])

    def construct_line(
        self,
        line: Union[OSMWay, OSMRelation],
        inners: list[list[OSMNode]],
        outers: list[list[OSMNode]],
    ) -> None:
        """
        Construct way or relation.

        :param line: OpenStreetMap way or relation
        :param inners: list of polygons that compose inner boundary
        :param outers: list of polygons that compose outer boundary
        """
        assert len(outers) >= 1

        if len(outers[0]) == 0:
            return

        if not self.check_level(line.tags):
            return

        center_point, _ = line_center(outers[0], self.flinger)
        if self.configuration.is_wireframe():
            color: Color
            if self.configuration.drawing_mode == DrawingMode.AUTHOR:
                color = get_user_color(
                    line.user if line.user else "", self.configuration.seed
                )
            elif self.configuration.drawing_mode == DrawingMode.TIME:
                color = get_time_color(line.timestamp, self.osm_data.time)
            elif self.configuration.drawing_mode == DrawingMode.WHITE:
                color = Color("#666666")
            elif self.configuration.drawing_mode == DrawingMode.BLACK:
                color = Color("#BBBBBB")
            elif self.configuration.drawing_mode != DrawingMode.NORMAL:
                logging.fatal(
                    f"Drawing mode {self.configuration.drawing_mode} is not "
                    f"supported."
                )
                sys.exit(1)
            self.draw_special_mode(line, inners, outers, color)
            return

        if not line.tags:
            return

        building_mode: BuildingMode = self.configuration.building_mode
        if "building" in line.tags or (
            building_mode == BuildingMode.ISOMETRIC
            and "building:part" in line.tags
        ):
            self.add_building(
                Building(line.tags, inners, outers, self.flinger, self.scheme)
            )

        road_matcher: RoadMatcher = self.scheme.get_road(line.tags)
        if road_matcher:
            road: Road = Road(
                line.tags, outers[0], road_matcher, self.flinger, self.scheme
            )
            self.roads.append(road)
            return

        processed: set[str] = set()

        recolor: Optional[Color] = None

        if line.tags.get("railway") == "subway":
            for color_tag_key in ["color", "colour"]:
                if color_tag_key in line.tags:
                    recolor = self.scheme.get_color(line.tags[color_tag_key])
                    processed.add(color_tag_key)

        line_styles: list[LineStyle] = self.scheme.get_style(line.tags)

        for line_style in line_styles:
            if recolor is not None:
                new_style: dict[str, Union[float, int, str]] = dict(
                    line_style.style
                )
                new_style["stroke"] = recolor.hex
                line_style = LineStyle(
                    new_style, line_style.parallel_offset, line_style.priority
                )

            self.figures.append(
                StyledFigure(line.tags, inners, outers, line_style)
            )
            if not (
                line.get_tag("area") == "yes"
                or line.get_tag("type") == "multipolygon"
                or is_cycle(outers[0])
                and line.get_tag("area") != "no"
                and self.scheme.is_area(line.tags)
            ):
                continue

            priority: int
            icon_set: IconSet
            icon_set, priority = self.scheme.get_icon(
                self.extractor, line.tags, processed, self.configuration
            )
            if icon_set is not None:
                labels: list[Label] = self.text_constructor.construct_text(
                    line.tags,
                    processed,
                    self.configuration.label_mode,
                )
                point: Point = Point(
                    icon_set,
                    labels,
                    line.tags,
                    processed,
                    center_point,
                    is_for_node=False,
                    priority=priority,
                    add_tooltips=self.configuration.show_tooltips,
                )
                self.points.append(point)

        if line_styles:
            return

        self.add_point_for_line(center_point, inners, line, outers)

    def add_point_for_line(self, center_point, inners, line, outers) -> None:
        """Add icon at the center point of the way or relation."""
        if DEBUG:
            style: dict[str, Any] = {
                "fill": "none",
                "stroke": Color("red").hex,
                "stroke-width": 1.0,
            }
            figure: StyledFigure = StyledFigure(
                line.tags, inners, outers, LineStyle(style, 0.0, 1000.0)
            )
            self.figures.append(figure)

        processed: set[str] = set()
        priority: int
        icon_set: IconSet
        icon_set, priority = self.scheme.get_icon(
            self.extractor, line.tags, processed, self.configuration
        )
        if icon_set is not None:
            labels: list[Label] = self.text_constructor.construct_text(
                line.tags, processed, self.configuration.label_mode
            )
            point: Point = Point(
                icon_set,
                labels,
                line.tags,
                processed,
                center_point,
                is_for_node=False,
                priority=priority,
                add_tooltips=self.configuration.show_tooltips,
            )
            self.points.append(point)

    def draw_special_mode(
        self,
        line: Union[OSMWay, OSMRelation],
        inners: list[list[OSMNode]],
        outers: list[list[OSMNode]],
        color: Color,
    ) -> None:
        """Add figure for special mode: time or author."""
        style: dict[str, Any] = {
            "fill": "none",
            "stroke": color.hex,
            "stroke-width": 1,
        }
        self.figures.append(
            StyledFigure(line.tags, inners, outers, LineStyle(style))
        )

    def construct_relations(self) -> None:
        """Construct Map Machine ways from OSM relations."""
        for relation_id in self.osm_data.relations:
            relation: OSMRelation = self.osm_data.relations[relation_id]
            tags: dict[str, str] = relation.tags
            if not self.check_level(tags):
                continue
            if "type" not in tags or tags["type"] != "multipolygon":
                continue
            inner_ways: list[OSMWay] = []
            outer_ways: list[OSMWay] = []
            for member in relation.members:
                if member.type_ == "way":
                    if member.role == "inner":
                        if member.ref in self.osm_data.ways:
                            inner_ways.append(self.osm_data.ways[member.ref])
                    elif member.role == "outer":
                        if member.ref in self.osm_data.ways:
                            outer_ways.append(self.osm_data.ways[member.ref])
                    else:
                        logging.warning(f'Unknown member role "{member.role}".')
            if outer_ways:
                inners_path: list[list[OSMNode]] = glue(inner_ways)
                outers_path: list[list[OSMNode]] = glue(outer_ways)
                self.construct_line(relation, inners_path, outers_path)

    def construct_nodes(self) -> None:
        """Draw nodes."""
        logging.info("Constructing nodes...")

        sorted_node_ids: Iterator[int] = sorted(
            self.osm_data.nodes.keys(),
            key=lambda x: -self.osm_data.nodes[x].coordinates[0],
        )
        for node_id in sorted_node_ids:
            self.construct_node(self.osm_data.nodes[node_id])

    def construct_node(self, node: OSMNode) -> None:
        """Draw one node."""
        tags: dict[str, str] = node.tags
        if not tags:
            return
        if not self.check_level(tags):
            return

        processed: set[str] = set()

        flung: np.ndarray = self.flinger.fling(node.coordinates)

        priority: int
        icon_set: IconSet
        draw_outline: bool = True

        if self.configuration.drawing_mode in (
            DrawingMode.AUTHOR,
            DrawingMode.TIME,
        ):
            color: Color = self.scheme.get_color("default")
            if self.configuration.drawing_mode == DrawingMode.AUTHOR:
                color = get_user_color(node.user, self.configuration.seed)
            if self.configuration.drawing_mode == DrawingMode.TIME:
                color = get_time_color(node.timestamp, self.osm_data.time)
            dot: Shape = self.extractor.get_shape(DEFAULT_SMALL_SHAPE_ID)
            icon_set: IconSet = IconSet(
                Icon([ShapeSpecification(dot, color)]),
                [],
                Icon([ShapeSpecification(dot, color)]),
                set(),
            )
            point: Point = Point(
                icon_set,
                [],
                tags,
                processed,
                flung,
                draw_outline=False,
                add_tooltips=self.configuration.show_tooltips,
            )
            self.points.append(point)
            return

        if self.configuration.drawing_mode in (
            DrawingMode.WHITE,
            DrawingMode.BLACK,
        ):
            if self.configuration.drawing_mode == DrawingMode.WHITE:
                color = Color("#CCCCCC")
            if self.configuration.drawing_mode == DrawingMode.BLACK:
                color = Color("#444444")
            icon_set, priority = self.scheme.get_icon(
                self.extractor, tags, processed, self.configuration
            )
            icon_set.main_icon.recolor(color)
            point: Point = Point(
                icon_set,
                [],
                tags,
                processed,
                flung,
                add_tooltips=self.configuration.show_tooltips,
            )
            self.points.append(point)
            return

        icon_set, priority = self.scheme.get_icon(
            self.extractor, tags, processed, self.configuration
        )
        if icon_set is None:
            return

        labels: list[Label] = self.text_constructor.construct_text(
            tags, processed, self.configuration.label_mode
        )
        self.scheme.process_ignored(tags, processed)

        if node.get_tag("natural") == "tree" and (
            "diameter_crown" in node.tags or "circumference" in node.tags
        ):
            self.trees.append(Tree(tags, node.coordinates, flung))
            return

        if node.get_tag("natural") == "crater" and "diameter" in node.tags:
            self.craters.append(Crater(tags, node.coordinates, flung))
            return

        if "direction" in node.tags or "camera:direction" in node.tags:
            self.direction_sectors.append(DirectionSector(tags, flung))
        point: Point = Point(
            icon_set,
            labels,
            tags,
            processed,
            flung,
            priority=priority,
            draw_outline=draw_outline,
            add_tooltips=self.configuration.show_tooltips,
        )
        self.points.append(point)

    def get_sorted_figures(self) -> list[StyledFigure]:
        """Get all figures sorted by priority."""
        return sorted(self.figures, key=lambda x: x.line_style.priority)

def check_level_number(tags: Tags, level: float) -> bool:
    """Check if element described by tags is no the specified level."""
    if "level" in tags:
        if level not in parse_levels(tags["level"]):
            return False
    else:
        return False
    return True

def check_level_overground(tags: Tags) -> bool:
    """Check if element described by tags is overground."""
    if "level" in tags:
        try:
            for level in map(float, tags["level"].replace(",", ".").split(";")):
                if level < 0.0:
                    return False
        except ValueError:
            pass

    return (
        tags.get("location") != "underground"
        and tags.get("parking") != "underground"
        and tags.get("tunnel") != "yes"
    )

max_stars_repo_name: enzet/Rontgen

max_issues_repo_name: enzet/Rontgen

max_forks_repo_name: enzet/Rontgen


text:
from fury import actor, window
from viz_shader_canvas import glyph_dot

import numpy as np
import vtk

scene = window.Scene()

colors = np.array([
    [.85, .07, .21], [.56, .14, .85], [.16, .65, .20], [.95, .73, .06],
    [.95, .55, .05], [.62, .42, .75], [.26, .58, .85], [.24, .82, .95],
    [.95, .78, .25], [.85, .58, .35], [1., 1., 1.]
])
n_points = len(colors)
np.random.seed(42)
centers = np.random.rand(n_points, 3)

canvas_actor = glyph_dot(centers, colors)
scene.add(actor.axes())
scene.add(canvas_actor)
#scene.background((1, 1, 1))
mapper = canvas_actor.GetMapper()

mapper.AddShaderReplacement(
    vtk.vtkShader.Fragment,
    "//VTK::Light::Dec",
    True,
    """
    //VTK::Light::Dec
    varying vec4 myVertexMC;

    float circle(vec2 uv, vec2 p, float r, float blur) {
        float d = length(uv - p);
        float c = smoothstep(r, r - blur, d);
        return c;
    }
    """,
    False
)

mapper.AddShaderReplacement(
    vtk.vtkShader.Fragment,
    "//VTK::Light::Impl",
    True,
    """
    vec3 color = vertexColorVSOutput.rgb;

    float xpos = 2 * gl_PointCoord.x - 1;
    float ypos = 1 - 2 * gl_PointCoord.y;
    vec2 p = vec2(xpos, ypos);

    // VTK Fake Spheres
    float p_len = length(p);
    if(p_len > 1) {
        discard;
    }

    // Smiley faces example
    float face = circle(p, vec2(0), 1, .05);
    face -= circle(p, vec2(-.35, .3), .2, .01);
    face -= circle(p, vec2(.35, .3), .2, .01);
    face -= circle(p, vec2(0.0, 0.0), .1, .01);

    float mouth = circle(p, vec2(0., 0.), .9, .02);
    mouth -= circle(p, vec2(0, .16), .9, .02);

    face -= mouth;

    fragOutput0 = vec4(color * face, 1);
    """,
    False
)

window.show(scene)

max_stars_repo_name: iamansoni/fury

max_issues_repo_name: iamansoni/fury

max_forks_repo_name: iamansoni/fury


text:
import numpy as np

from .wavefront import Wavefront, jones_to_mueller
from .optical_element import OpticalElement, make_agnostic_forward, make_agnostic_backward, AgnosticOpticalElement
from ..field import Field, field_dot, field_conjugate_transpose

def rotation_matrix(angle):
	'''Two dimensional rotation matrix.

	Parameters
	----------
	angle : scaler
		rotation angle in radians

	Returns
	-------
	ndarray
		The rotation matrix.
	'''
	return np.array([[np.cos(angle),  np.sin(angle)],[-np.sin(angle), np.cos(angle)]])

class JonesMatrixOpticalElement(AgnosticOpticalElement):
	'''A general Jones Matrix.

	Parameters
	----------
	jones_matrix : matrix or Field
		The Jones matrix describing the optical element.
	'''
	def __init__(self, jones_matrix):
		self.jones_matrix = jones_matrix

		AgnosticOpticalElement.__init__(self, True, True)

	def make_instance(self, instance_data, input_grid, output_grid, wavelength):
		instance_data.jones_matrix = self.evaluate_parameter(self.jones_matrix, input_grid, output_grid, wavelength)

	def get_input_grid(self, output_grid, wavelength):
		return output_grid

	def get_output_grid(self, input_grid, wavelength):
		return input_grid

	@make_agnostic_forward
	def forward(self, instance_data, wavefront):
		'''Propgate the wavefront through the Jones matrix.

		Parameters
		----------
		wavefront : Wavefront
			The wavefront to propagate.

		Returns
		-------
		Wavefront
			The propagated wavefront.
		'''
		if wavefront.is_scalar:
			# We generate an unpolarized wavefront
			if hasattr(instance_data.jones_matrix, 'grid'):
				electric_field = instance_data.jones_matrix * wavefront.electric_field
			else:
				electric_field = instance_data.jones_matrix[..., np.newaxis] * wavefront.electric_field

			return Wavefront(electric_field, wavelength=wavefront.wavelength, input_stokes_vector=[1, 0, 0, 0])
		else:
			wf = wavefront.copy()
			wf.electric_field = field_dot(instance_data.jones_matrix, wf.electric_field)

			return wf

	@make_agnostic_backward
	def backward(self, instance_data, wavefront):
		'''Propagate the wavefront backwards through the Jones matrix.

		Parameters
		----------
		wavefront : Wavefront
			The wavefront to propagate.

		Returns
		-------
		Wavefront
			The propagated wavefront.
		'''
		if wavefront.is_scalar:
			# We generate an unpolarized wavefront
			if hasattr(instance_data.jones_matrix, 'grid'):
				electric_field = field_conjugate_transpose(instance_data.jones_matrix) * wavefront.electric_field
			else:
				electric_field = field_conjugate_transpose(instance_data.jones_matrix)[..., np.newaxis] * wavefront.electric_field

			return Wavefront(electric_field, wavelength=wavefront.wavelength, input_stokes_vector=[1, 0, 0, 0])
		else:
			wf = wavefront.copy()
			wf.electric_field = field_dot(field_conjugate_transpose(instance_data.jones_matrix), wf.electric_field)

			return wf

	@property
	def mueller_matrix(self):
		'''Returns the Mueller matrix corresponding to the Jones matrix.
		'''
		return self.construct_function(jones_to_mueller, self.jones_matrix)

	def __mul__(self, b):
		'''Multiply the Jones matrix on the right side with the Jones matrix m.

		Parameters
		----------
		m : JonesMatrixOpticalElement, Field or matrix
			The Jones matrix with which to multiply.
		'''
		if hasattr(b, 'jones_matrix'):
			jones_matrix = self.construct_function(field_dot, self.jones_matrix, b.jones_matrix)
		else:
			jones_matrix = self.construct_function(field_dot, self.jones_matrix, b)

		return JonesMatrixOpticalElement(jones_matrix)

class PhaseRetarder(JonesMatrixOpticalElement):
	'''A general phase retarder.

	Parameters
	----------
	phase_retardation : scalar or Field
		The relative phase retardation induced between the fast and slow axis.
	fast_axis_orientation : scalar or Field
		The angle of the fast axis with respect to the x-axis in radians.
	circularity : scalar or Field
		The circularity of the phase retarder.
	'''
	def __init__(self, phase_retardation, fast_axis_orientation, circularity):
		self._phase_retardation = phase_retardation
		self._fast_axis_orientation = fast_axis_orientation
		self._circularity = circularity

		JonesMatrixOpticalElement.__init__(self, self.jones_matrix)

	@property
	def jones_matrix(self):
		def jones(phase_retardation, fast_axis_orientation, circularity):
			phi_plus = np.exp(1j * phase_retardation / 2)
			phi_minus = np.exp(-1j * phase_retardation / 2)

			# calculating the individual components
			j11 = phi_plus * np.cos(fast_axis_orientation)**2 + phi_minus * np.sin(fast_axis_orientation)**2
			j12 = (phi_plus - phi_minus) * np.exp(-1j * circularity) * np.cos(fast_axis_orientation) * np.sin(fast_axis_orientation)
			j21 = (phi_plus - phi_minus) * np.exp(1j * circularity) * np.cos(fast_axis_orientation) * np.sin(fast_axis_orientation)
			j22 = phi_plus * np.sin(fast_axis_orientation)**2 + phi_minus * np.cos(fast_axis_orientation)**2

			# constructing the Jones matrix.
			jones_matrix = np.array([[j11, j12], [j21, j22]])

			if hasattr(j11, 'grid'):
				jones_matrix = Field(jones_matrix, j11.grid)

			return jones_matrix

		return self.construct_function(jones, self.phase_retardation, self.fast_axis_orientation, self.circularity)

	@jones_matrix.setter
	def jones_matrix(self, b):
		pass

	@property
	def phase_retardation(self):
		return self._phase_retardation

	@phase_retardation.setter
	def phase_retardation(self, phase_retardation):
		self._phase_retardation = phase_retardation

		self.clear_cache()

	@property
	def fast_axis_orientation(self):
		return self._fast_axis_orientation

	@fast_axis_orientation.setter
	def fast_axis_orientation(self, fast_axis_orientation):
		self._fast_axis_orientation = fast_axis_orientation

		self.clear_cache()

	@property
	def circularity(self):
		return self._circularity

	@circularity.setter
	def circularity(self, circularity):
		self._circularity = circularity

		self.clear_cache()

class LinearRetarder(PhaseRetarder):
	'''A general linear retarder.

	Parameters
	----------
	phase_retardation : scalar or Field
		The relative phase retardation induced between the fast and slow axis.
	fast_axis_orientation : scalar or Field
		The angle of the fast axis with respect to the x-axis in radians.
	'''
	def __init__(self, phase_retardation, fast_axis_orientation):
		PhaseRetarder.__init__(self, phase_retardation, fast_axis_orientation, 0)

class CircularRetarder(PhaseRetarder):
	'''A general circular retarder.

	Parameters
	----------
	phase_retardation : scalar or Field
		The relative phase retardation induced between the fast and slow axis.
	fast_axis_orientation : scalar or Field
		The angle of the fast axis with respect to the x-axis in radians.
	'''
	def __init__(self, phase_retardation):
		PhaseRetarder.__init__(self, phase_retardation, np.pi / 4, np.pi / 2)

class QuarterWavePlate(LinearRetarder):
	'''A quarter-wave plate.

	Parameters
	----------
	fast_axis_orientation : scalar or Field
		The angle of the fast axis with respect to the x-axis in radians.
	'''
	def __init__(self, fast_axis_orientation):
		LinearRetarder.__init__(self, np.pi / 2, fast_axis_orientation)

class HalfWavePlate(LinearRetarder):
	'''A half-wave plate.

	Parameters
	----------
	fast_axis_orientation : scalar or Field
		The angle of the fast axis with respect to the x-axis in radians.
	'''
	def __init__(self, fast_axis_orientation):
		LinearRetarder.__init__(self, np.pi, fast_axis_orientation)

class GeometricPhaseElement(LinearRetarder):
	'''A general geometric phase element.

	Parameters
	----------
	phase_pattern : Field or array_like
		The phase pattern in radians.
	leakage : scalar
		The relative leakage strength (0 = no leakage, 1 = maximum leakage)
	retardance_offset : scalar
		The retardance offset from half wave in radians. This will result in leakage.
	'''
	def __init__(self, phase_pattern, leakage=None, retardance_offset=0):
		if leakage is not None:
			# Testing if the leakage is valid.
			if leakage < 0 or leakage > 1:
				raise ValueError('Leakage must be between 0 and 1.')

			# Calculating the required retardance offset to get the leakage.
			retardance_offset = 2 * np.arcsin(np.sqrt(leakage))

		LinearRetarder.__init__(self, np.pi - retardance_offset, phase_pattern / 2)

class LinearPolarizer(JonesMatrixOpticalElement):
	'''A linear polarizer.

	Parameters
	----------
	polarization_angle : scalar or Field
		The polarization angle of the polarizer. Light with this angle is transmitted.
	'''
	def __init__(self, polarization_angle):
		self.polarization_angle = polarization_angle

		JonesMatrixOpticalElement.__init__(self, self.jones_matrix)

	@property
	def jones_matrix(self):
		def jones(angle):
			c = np.cos(angle)
			s = np.sin(angle)

			return np.array([[c**2, c * s],[c * s, s**2]])

		return self.construct_function(jones, self.polarization_angle)

	@jones_matrix.setter
	def jones_matrix(self, b):
		pass

	@property
	def polarization_angle(self):
		'''The angle of polarization of the linear polarizer.
		'''
		return self._polarization_angle

	@polarization_angle.setter
	def polarization_angle(self, angle):
		self._polarization_angle = angle

		self.clear_cache()

class LinearPolarizingBeamSplitter(OpticalElement):
	''' A linear polarizing beam splitter that accepts one wavefront and returns two orthogonally linearly polarized wavefronts.

	The first of the returned wavefront will have the polarization state set by the polarization angle. The second wavefront
	will be perpendicular to the first.

	Parameters
	----------
	polarization_angle : scalar or Field
		The polarization angle of the polarizer.
	'''
	def __init__(self, polarization_angle, wavelength=1):
		self.polarization_angle = polarization_angle

	@property
	def polarization_angle(self):
		'''The angle of polarization of the linear polarizer.
		'''
		return self._polarization_angle

	@polarization_angle.setter
	def polarization_angle(self, polarization_angle):
		self._polarization_angle = polarization_angle

		# Calculating two Jones matrices for the two ports.
		self.polarizer_port_1 = LinearPolarizer(polarization_angle)
		self.polarizer_port_2 = LinearPolarizer(polarization_angle + np.pi / 2)

	def forward(self, wavefront):
		'''Propgate the wavefront through the PBS.

		Parameters
		----------
		wavefront : Wavefront
			The wavefront to propagate.

		Returns
		-------
		wf_1 : Wavefront
			The wavefront propagated through a polarizer under the polarization angle.

		wf_2 : Wavefront
			The propagated wavefront through a polarizer perpendicular to the first one.
		'''

		wf_1 = self.polarizer_port_1.forward(wavefront)
		wf_2 = self.polarizer_port_2.forward(wavefront)

		return wf_1, wf_2

	def backward(self, wavefront):
		'''Propagate the wavefront backwards through the PBS.

		Not possible, will raise error.
		'''
		raise RuntimeError('Backward propagation through PolarizingBeamSplitter not possible.')

	@property
	def mueller_matrices(self):
		'''Returns the Mueller matrices of the two Jones matrices.
		'''
		return jones_to_mueller(self.polarizer_port_1.jones_matrix), jones_to_mueller(self.polarizer_port_2.jones_matrix)

class CircularPolarizingBeamSplitter(OpticalElement):
	''' A circular polarizing beam splitter that accepts one wavefront and returns two linearly polarized wavefronts.

	The circular polarizing beam splitter is a combination of a quarter-wave plate and a polarizing beam splitter. The
	quarter-wave plate rotation angle is 45 degrees and the polarizing beam-splitter rotation angle is zero degrees.
	Therefore, the first of the returned wavefronts will have a linear polarization state with an electric field equal
	to the amount of left-circular polarization in the incoming wavefront. The second wavefront will be perpendicular
	to the first and with electric field equal to the amount of right-circular polarization in the incoming wavefront.

	Parameters
	----------
	polarization_angle : scalar or Field
		The polarization angle of the polarizer.
	'''
	def __init__(self, wavelength=1):
		self.polarization_angle = 0
		self.quarter_wave_angle = np.pi / 4
		self.quarter_wave_plate = QuarterWavePlate(self.quarter_wave_angle)
		self.linear_polarizing_beam_splitter = LinearPolarizingBeamSplitter(self.polarization_angle)

	def forward(self, wavefront):
		'''Propgate the wavefront through the CBS.

		Parameters
		----------
		wavefront : Wavefront
			The wavefront to propagate.

		Returns
		-------
		wf_1 : Wavefront
			The wavefront propagated through a quarter-wave plate at 45 degrees and a polarizer 0 degrees.
		wf_2 : Wavefront
			The wavefront propagated through a quarter-wave plate at 45 degrees and a polarizer 90 degrees.
		'''
		wf_1, wf_2 = self.linear_polarizing_beam_splitter.forward(self.quarter_wave_plate.forward(wavefront))
		return wf_1, wf_2

	def backward(self, wavefront):
		'''Propagate the wavefront backwards through the CBS.

		Not possible, will raise error.
		'''
		raise RuntimeError('Backward propagation through PolarizingBeamSplitter not possible.')

	@property
	def mueller_matrices(self):
		'''Returns the Mueller matrices of the two Jones matrices.
		'''
		qwp_mueller = self.quarter_wave_plate.mueller_matrix
		lin_pol_bs_mueller_1 , lin_pol_bs_mueller_2 = self.linear_polarizing_beam_splitter.mueller_matrices

		return np.dot(lin_pol_bs_mueller_1, qwp_mueller), np.dot(lin_pol_bs_mueller_2, qwp_mueller)


text:
"""Loss functions

Jax loss functions for computing gradient updates.
"""

import jax
import jax.numpy as jnp
from jax.config import config
config.enable_omnistaging()

def mse_loss(Y, Y_hat):
    """Squeezes predictions and returns MSE loss."""

    if len(Y_hat.shape) > 1:
        Y_hat = jnp.squeeze(Y_hat, axis=1)

    loss = jnp.mean(jnp.square(Y - Y_hat))

    return loss

def cross_entropy_loss(Y, Y_hat, num_classes):
    """Applies log-softmax to predictions and one-hot encodes true values
       to compute and return cross-entropy loss.
    """

    Y_hat = jax.nn.log_softmax(Y_hat)

    Y = jax.nn.one_hot(Y, num_classes=num_classes)

    loss = -jnp.sum(Y * Y_hat)

    return loss

max_issues_repo_name: amirshane/protein-embedding-retrieval

max_forks_repo_name: amirshane/protein-embedding-retrieval

reason: import jax,from jax


text:
import yagmail
from sklearn.neighbors import NearestNeighbors
import yaml
import numpy as np
import collections

import torch
import copy

def flatten_list(d, parent_key='', sep='_'):
    items = []
    for num, element in enumerate(d):
        new_key = parent_key + sep + str(num) if parent_key else str(num)

        if isinstance(element, collections.MutableMapping):
            items.extend(flatten_dict(element, new_key, sep=sep).items())
        else:
            if isinstance(element, list):
                if isinstance(element[0], dict):
                    items.extend(flatten_list(element, new_key, sep=sep))
                    continue
            items.append((new_key, element))
    return items

def flatten_dict(d, parent_key='', sep='_'):
    items = []
    for k, v in d.items():
        new_key = parent_key + sep + k if parent_key else k
        if isinstance(v, collections.MutableMapping):
            items.extend(flatten_dict(v, new_key, sep=sep).items())
        else:
            if isinstance(v, list):
                if isinstance(v[0], dict):
                    items.extend(flatten_list(v, new_key, sep=sep))
                    continue
            items.append((new_key, v))
    return dict(items)

def norm_quat(q):
    # ToDo raise type and dim error
    return q / torch.sqrt(torch.sum(q * q))

def pad(s, sym='-', p='l', length=80):
    if len(s) > length:
        return s
    else:
        if p == 'c':
            front = int((length - len(s)) / 2)
            s = sym * front + s
            back = int(length - len(s))
            s = s + sym * back
        if p == 'l':
            back = int(length - len(s))
            s = s + sym * back
        return s

def re_quat(q, input_format):

    if input_format == 'xyzw':
        if isinstance(q, torch.Tensor):
            v0 = q[0].clone()
        else:
            v0 = copy.deepcopy(q[0])

        q[0] = q[3]
        q[3] = q[2]
        q[2] = q[1]
        q[1] = v0
        return q
    elif input_format == 'wxyz':
        if isinstance(q, torch.Tensor):
            v0 = q[0].clone()
        else:
            v0 = copy.deepcopy(q[0])

        q[0] = q[1]
        q[1] = q[2]
        q[2] = q[3]
        q[3] = v0
        return q

def send_email(text):
    yag = yagmail.SMTP('trackthisplr', "TrackThis")
    contents = [
        "Run is finished!",
        text
    ]
    yag.send('jonfrey@student.ethz.ch',
             'PLR - TrackThis - Lagopus', contents)
    yag.send('yavyas@student.ethz.ch',
             'PLR - TrackThis - Lagopus', contents)

def compose_quat(p, q, device):
    """
    input is wxyz
    """
    q = norm_quat(re_quat(q.squeeze(), 'wxyz')).unsqueeze(0)
    p = norm_quat(re_quat(p.squeeze(), 'wxyz')).unsqueeze(0)
    product = torch.zeros(
        (max(p.shape[0], q.shape[0]), 4), dtype=torch.float32, device=device)
    product[:, 3] = p[:, 3] * q[:, 3] - torch.sum(p[:, :3] * q[:, :3], (1))
    product[:, :3] = (p[:, None, 3] * q[:, :3] + q[:, None, 3] * p[:, :3] +
                      torch.cross(p[:, :3], q[:, :3]))

    return re_quat(product.squeeze(0), 'xyzw')

def rotation_angle(q, device):
    # in radians
    q = norm_quat(q)
    unit_r = torch.t(torch.tensor(
        [[0, 0, 0, 1]], dtype=torch.float32, device=device))
    return torch.asin(torch.mm(q, unit_r)) * 2

def nearest_neighbor(src, dst):
    assert src.shape[1] == dst.shape[1]

    neigh = NearestNeighbors(n_neighbors=1, n_jobs=8)
    neigh.fit(dst)
    distances, indices = neigh.kneighbors(src, return_distance=True)
    return distances.ravel(), indices.ravel()

def replace_item(obj, key, replace_value):
    for k, v in obj.items():
        if isinstance(v, dict):
            obj[k] = replace_item(v, key, replace_value)
    if key in obj:
        obj[key] = replace_value
    return obj

def generate_unique_idx(num, max_idx):
    a = random.sample(range(0, max_idx), k=min(num, max_idx))
    while len(a) < num:
        a = a + random.sample(
            range(0, max_idx), k=min(max_idx, num - len(a)))
    return a

def get_bbox_480_640(label):
    border_list = [-1, 40, 80, 120, 160, 200, 240, 280,
                   320, 360, 400, 440, 480, 520, 560, 600, 640, 680]
    img_width = 480
    img_length = 640

    # print(type(label))
    rows = np.any(label, axis=1)
    cols = np.any(label, axis=0)
    rmin, rmax = np.where(rows)[0][[0, -1]]
    cmin, cmax = np.where(cols)[0][[0, -1]]
    rmax += 1
    cmax += 1
    r_b = rmax - rmin
    for tt in range(len(border_list)):
        if r_b > border_list[tt] and r_b < border_list[tt + 1]:
            r_b = border_list[tt + 1]
            break
    c_b = cmax - cmin
    for tt in range(len(border_list)):
        if c_b > border_list[tt] and c_b < border_list[tt + 1]:
            c_b = border_list[tt + 1]
            break
    center = [int((rmin + rmax) / 2), int((cmin + cmax) / 2)]
    rmin = center[0] - int(r_b / 2)
    rmax = center[0] + int(r_b / 2)
    cmin = center[1] - int(c_b / 2)
    cmax = center[1] + int(c_b / 2)
    if rmin < 0:
        delt = -rmin
        rmin = 0
        rmax += delt
    if cmin < 0:
        delt = -cmin
        cmin = 0
        cmax += delt
    if rmax > img_width:
        delt = rmax - img_width
        rmax = img_width
        rmin -= delt
    if cmax > img_length:
        delt = cmax - img_length
        cmax = img_length
        cmin -= delt
    return rmin, rmax, cmin, cmax


text:
# Function to represent the state (of the pegs)
# state = [[1,2,3], [2,3,4], []]
from copy import deepcopy
def printState(state):
	statenew = deepcopy(state)
	for i in statenew:
		if len(i) == 0:
			i.insert(0, ' ')
			i.insert(0, ' ')
			i.insert(0, ' ')
		elif len(i) == 1:
			i.insert(0, ' ')
			i.insert(0, ' ')
		elif len(i) == 2:
			i.insert(0, ' ')

	for i in list(zip(*statenew)):
		for j in i:
			print(j, end=' ')
		print()
	print('-----')
# Function which returns valid moves from a state
def validMoves(state):
	valid = []
	for i in range(3):
		if state[i] == []:
			continue
		for j in range(3):
			if state[i] != state[j]:
				# Check if j is empty. If yes, then insert [i,j].
				if len(state[j]) == 0:
					valid.append([i+1,j+1])
				# Else, check if disc in peg j is greater than disc in peg i. If yes, then insert.
				else:
					if state[j][0] > state[i][0]:
						valid.append([i+1,j+1])
	return valid

# Function to apply a move to a state and return the new state.
def makeMove(state, move):
	if move != None:
		state2 = deepcopy(state)
		# Get the pegs from and to which the moves are to be done.
		source = move[0]
		dest = move[1]
		# Get the disk from source
		disk = state2[source - 1][0]
		# Remove from source
		state2[source-1].remove(disk)
		# Move disk to destination
		state2[dest - 1].insert(0, disk)
		return state2

# Function to check if the goal state has been reached.
def isGoalState(state):
	if state == [[], [], [1,2,3]]:
		return True
	else:
		return False

# Function which converts lists to tuples.
def stateMoveTuple(state, move):
	stateTuple = tuple(tuple(x) for x in state)
	moveTuple = tuple(move)
	stateMoveTuple = (stateTuple, moveTuple)
	return stateMoveTuple

import pprint
import random
# Epsilon Greedy
def epsilonGreedy(epsilon, Q, state):
    valid = validMoves(state)
    qlist = []
    if np.random.uniform() < epsilon:
        # Random Move
        return random.choice(valid)
    else:
        # Greedy Move - get action for which the Q value is maximum.
        # Get Q values for all (state, move) pairs and store in a list.
        for move in valid:
        	smt = stateMoveTuple(state, move)
        	Qval = Q.get(smt, -1)
        	qlist.append((Qval, move))
        maxMove = max(qlist, key=lambda x:x[0])
        return maxMove[1]

import numpy as np

# Function for training Q for Towers of Hanoi.
def trainQ(nRepetitions, learningRate, epsilonDecayFactor, validMovesF, makeMoveF):
	epsilon = 1.0
	Q = {}
	stepsToGoal = []
	for nGames in range(nRepetitions):
	    print("we're at rep: ", nGames)
	    # Decay epsilon
	    epsilon *= epsilonDecayFactor

	    # Start state for ToH
	    s = [[1,2,3], [], []]

	    # Play a game till solution occurs for ToH.
	    done = False
	    step = 0
	    while not done:
	        step += 1

	        # Choose a move.
	        move = epsilonGreedy(epsilon, Q, s)

	        # Apply the move on a copy of state.
	        sNew = deepcopy(s)
	        sNew = makeMoveF(sNew, move)

	        if stateMoveTuple(s, move) not in Q:
	            Q[stateMoveTuple(s, move)] = -1  # initial Q value for new board,move

            # If the goal state is reached, then update Q(s, move) += rho*(0 - Q(s, move)) and break game (inner loop).

	        if isGoalState(sNew):
	            Q[stateMoveTuple(s, move)] += learningRate * (0 - Q[stateMoveTuple(s, move)])
	            done = True

	        else:
	            if step > 1:
	                Q[stateMoveTuple(sOld,moveOld)] += learningRate * (-1 + Q[stateMoveTuple(s,move)] - Q[stateMoveTuple(sOld,moveOld)])
	            sOld, moveOld = s, move # remember board and move to Q(board,move) can be updated after next steps
	            s = sNew
	    stepsToGoal.append(step)

	return Q, stepsToGoal

if __name__ == '__main__':

    nRepetitions = 50
    learningRate = 0.5
    epsilonDecayFactor = 0.7
    Q, stepsToGoal = trainQ(nRepetitions, learningRate, epsilonDecayFactor, validMoves, makeMove)
    print(stepsToGoal)
    #print(validMoves([[], [1,2], [3]]))
    #print(epsilonGreedy(0, ))


text:
# -*- coding: utf-8 -*-
import numpy as np
import cv2

class SegPatternGenerator:
    """7セグパターン生成器"""

    """全体のスケール
    1(960, 480)を想定していたが処理速度が
    遅かったため、2(480, 240)に変更。"""
    SCALE = 2

    """画面サイズ"""
    SCREEN_HEIGHT = 480 // SCALE
    SCREEN_WIDTH  = 960 // SCALE

    """7セグ1個のサイズ"""
    SEG_HEIGHT = 60 // SCALE
    SEG_WIDTH  = 40 // SCALE

    """画面内の7セグ数"""
    SEG_X_NUM = SCREEN_WIDTH // SEG_WIDTH
    SEG_Y_NUM = SCREEN_HEIGHT // SEG_HEIGHT

    """サイズ(x=40, y=60)における7セグLEDの各セグメント多角形の頂点座標"""
    POINT_SEG_A = np.array([ [14, 7],  [32, 7],  [33, 8],  [30, 11], [16, 11], [13, 8]  ], np.int32)
    POINT_SEG_B = np.array([ [34, 10], [36, 12], [33, 26], [31, 28], [29, 26], [31, 13] ], np.int32)
    POINT_SEG_C = np.array([ [30, 31], [33, 34], [30, 48], [29, 49], [26, 46], [28, 33] ], np.int32)
    POINT_SEG_D = np.array([ [24, 47], [27, 50], [26, 51], [9, 51],  [8, 50],  [11, 47] ], np.int32)
    POINT_SEG_E = np.array([ [9, 30],  [11, 32], [9, 46],  [6, 49],  [4, 47],  [7, 32]  ], np.int32)
    POINT_SEG_F = np.array([ [11, 9],  [14, 12], [11, 26], [9, 28],  [7, 26],  [10, 10] ], np.int32)
    POINT_SEG_G = np.array([ [13, 27], [27, 27], [29, 29], [27, 31], [13, 31], [11, 29] ], np.int32)
    POINT_SEG_DOT_CENTER = (35, 49)
    POINT_SEG_DOT_RADIUS = 3

    POINT_SEG_A //= SCALE
    POINT_SEG_B //= SCALE
    POINT_SEG_C //= SCALE
    POINT_SEG_D //= SCALE
    POINT_SEG_E //= SCALE
    POINT_SEG_F //= SCALE
    POINT_SEG_G //= SCALE
    POINT_SEG_DOT_CENTER = (POINT_SEG_DOT_CENTER[0] // SCALE,
                            POINT_SEG_DOT_CENTER[1] // SCALE)
    POINT_SEG_DOT_RADIUS //= SCALE

    ############################################################
    #  事前処理
    ############################################################
    def __init__(self):
        self.all_matching_points = self.__get_all_matching_points()

    def __get_points_of_polygon(self, poly_points):
        """多角形を構成する座標群を返す"""
        img = np.zeros((SegPatternGenerator.SEG_HEIGHT,
                        SegPatternGenerator.SEG_WIDTH), np.uint8)
        img = cv2.fillPoly(img, [poly_points], 255)
        points = []
        for y in range(SegPatternGenerator.SEG_HEIGHT):
            for x in range(SegPatternGenerator.SEG_WIDTH):
                if (img[y, x] == 255):
                    points.append([y, x])
        return points

    def __get_points_of_circle(self, center, radius):
        """円を構成する座標群を返す"""
        img = np.zeros((SegPatternGenerator.SEG_HEIGHT,
                        SegPatternGenerator.SEG_WIDTH), np.uint8)
        img = cv2.circle(img, center, radius, 255, thickness=-1)
        points = []
        for y in range(SegPatternGenerator.SEG_HEIGHT):
            for x in range(SegPatternGenerator.SEG_WIDTH):
                if (img[y, x] == 255):
                    points.append([y, x])
        return points

    def __get_seven_seg_points(self):
        """7セグを構成する座標群を返す"""
        return [
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_A),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_B),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_C),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_D),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_E),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_F),
            self.__get_points_of_polygon(SegPatternGenerator.POINT_SEG_G),
            self.__get_points_of_circle(SegPatternGenerator.POINT_SEG_DOT_CENTER,
                                      SegPatternGenerator.POINT_SEG_DOT_RADIUS),
        ]

    def __get_all_matching_points(self):
        """
        走査する全ての座標を備えたリストを作成する。
        all_points[seven_y][seven_x][seg][pts][xy]
        - seven_y[0:8]  : 7セグ1個のy座標([0]=1行目, ..., [7]=8行目)
        - seven_x[0:16] : 7セグ1個のx座標([0]=1列目, ..., [7]=16列目)
        - seg[0:8]      : 7セグの各セグメント([0]=SEG_A, ..., [7]=SEG_DOT)
        - pts[0:N]      : 各セグメントを構成する座標群([0]=SEGの1番目の座標)
        - xy[2]         : 座標[y, x]
        ※ptsは各セグメントによって個数が異なる。
        例: all_points[0][0][0][0][0]=1行1列目の7セグのSEG_Aの1番目のy座標
        """
        seven_seg_points = self.__get_seven_seg_points()
        seven_y = []
        for y in range(0, SegPatternGenerator.SCREEN_HEIGHT, SegPatternGenerator.SEG_HEIGHT):
            seven_x = []
            for x in range(0, SegPatternGenerator.SCREEN_WIDTH, SegPatternGenerator.SEG_WIDTH):
                base = [y, x]
                seven_seg_points_based = [ (np.array(seven_seg_points[i]) + np.array(base)) for i in range(8) ]
                seven_x.append(seven_seg_points_based)
            seven_y.append(seven_x)
        return seven_y

    ############################################################
    #  リアルタイム処理
    ############################################################
    def match(self, cv_img):
        """
        画像から7セグパターンとマッチングした結果を返す
        @param cv_img (SCREEN_WIDTH, SCREEN_HEIGHT)のuint8型のNumPy二次元配列
        @return (SCREEN_WIDTH, SCREEN_HEIGHT)のuint8型のNumPy二次元配列
        """
        return self.__create_seven_seg_pattern(cv_img)

    def __is_seg_light(self, img, seg_pts):
        """入力画像imgに対してseg_ptsで指定した座標群が半分以上閾値以上ならTrueを返す"""
        and_count = 0
        for [y, x] in seg_pts:
            if (img.item(y, x) >= 127):
                and_count += 1
        if (and_count >= len(seg_pts)):
            return True
        else:
            return False

    def __create_seven_seg_pattern(self, img):
        pattern = np.zeros((SegPatternGenerator.SEG_Y_NUM,
                            SegPatternGenerator.SEG_X_NUM), np.uint8)
        for seven_y in range(SegPatternGenerator.SEG_Y_NUM):
            for seven_x in range(SegPatternGenerator.SEG_X_NUM):
                bit_pattern = 0x00
                for seg in range(8):
                    if (self.__is_seg_light(img, self.all_matching_points[seven_y][seven_x][seg])):
                        bit_pattern |= (1 << (7 - seg))
                pattern[seven_y][seven_x] = bit_pattern
        return pattern

    def create_seven_seg_pattern_image(self, pattern):
        """7セグパターンの画像を作成する(出力結果確認用)"""
        img = np.zeros((SegPatternGenerator.SCREEN_HEIGHT,
                        SegPatternGenerator.SCREEN_WIDTH), np.uint8)
        for y in range(SegPatternGenerator.SEG_Y_NUM):
            for x in range(SegPatternGenerator.SEG_X_NUM):
                bit_pattern = pattern[y][x]
                for seg in range(8):
                    if (bit_pattern & (1 << (7 - seg))):
                        for point in self.all_matching_points[y][x][seg]:
                            img.itemset((point[0], point[1]), 255)
        return img

    def print_seven_seg_pattern(self, pattern):
        """7セグパターンを標準出力に表示する(出力結果確認用)"""
        for y in range(SegPatternGenerator.SEG_Y_NUM):
            for x in range(SegPatternGenerator.SEG_X_NUM):
                print('%02X' % int(pattern[y][x]), end='')
            print('')
        print('')

if __name__ == '__main__':
    from PIL import Image, ImageDraw, ImageFont, ImageOps, ImageChops

    spg = SegPatternGenerator()
    font = ImageFont.truetype('C:\Windows\Fonts\meiryob.ttc', 144)

    print('Font: Load OK!')

    count = 0
    for i in range(0, 10000, 1):
        """加工元画像"""
        pil_img = Image.new('1', (spg.SCREEN_WIDTH, spg.SCREEN_HEIGHT))
        draw = ImageDraw.Draw(pil_img)
        draw.text(( - (i * 8 % (spg.SCREEN_WIDTH * 4)) + 1000, 10), u'ext@漢字明朝', fill=255, font=font)

        back_img = Image.new('1', (spg.SCREEN_WIDTH, spg.SCREEN_HEIGHT))
        draw_back = ImageDraw.Draw(back_img)

        radius = (i*8) % ((spg.SCREEN_WIDTH * 1.5) // 2)
        fill_color = 0
        if ((i*8) % (spg.SCREEN_WIDTH * 1.5) > radius):
            fill_color = 0
            back_img = ImageChops.invert(back_img)
        else:
            fill_color = 255

        box_top_xy = (spg.SCREEN_WIDTH // 2 - radius, spg.SCREEN_HEIGHT // 2 - radius)
        box_bottom_xy = (spg.SCREEN_WIDTH // 2 + radius, spg.SCREEN_HEIGHT // 2 + radius)
        draw_back.ellipse((box_top_xy, box_bottom_xy), fill=fill_color)

        pil_img = ImageChops.logical_xor(pil_img, back_img)

        """PIL画像からOpenCV画像に変換"""
        pil_img = pil_img.convert('L')
        cv_img = np.asarray(pil_img)

        """入力画像から7セグパターン生成"""
        pattern = spg.match(cv_img)

        """パターン出力結果確認"""
        cv_img = spg.create_seven_seg_pattern_image(pattern)
        cv2.imshow('Title', cv_img)

        k = cv2.waitKey(1)
        if k == 27:
            break