text:
from functools import partial
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
from estimagic import minimize
from scipy.optimize import minimize as scipy_minimize

from src.manfred.minimize_manfred import minimize_manfred
from src.manfred.minimize_manfred_estimagic import minimize_manfred_estimagic

def criterion_function(x, seed, true_x, data, noise_level=0):
    np.random.seed(seed)
    true_y_hat = data @ true_x
    calc_y = data @ x
    noise = (
        ((true_y_hat - calc_y) ** 2 + 1)
        * noise_level
        * np.random.normal(size=len(data))
    )
    true_y = true_y_hat + noise
    base_residuals = true_y - calc_y
    residuals = np.abs(base_residuals * 3) ** 1.5 * np.sign(base_residuals)

    value = residuals @ residuals
    return {"value": value, "root_contributions": residuals}

def scipy_criterion_function(x, true_x, data, noise_level):
    seed = np.random.randint(10000)
    return criterion_function(x, seed, true_x, data, noise_level)["value"]

def estimagic_criterion_function(params, true_x, data, noise_level):  # noqa: U100
    seed = np.random.choice(10000)
    out = criterion_function(
        params["value"].to_numpy(), seed, true_x, data, noise_level
    )
    return out

def plot_history(res, x_names=None):
    x_history = res["history"]["x"]
    dim = len(x_history[0])
    if x_names is None:
        x_names = [f"x_{i}" for i in range(dim)]
    plot_data = pd.DataFrame(x_history, columns=x_names)
    plot_data["criterion"] = res["history"]["criterion"]
    plot_data = plot_data[["criterion"] + x_names]

    n_plots = dim + 1
    fig, axes = plt.subplots(nrows=n_plots, figsize=(4, n_plots * 2.5))

    for col, ax in zip(plot_data.columns, axes):
        if col == "criterion":
            sns.lineplot(
                x=np.arange(len(plot_data)) + 1,
                y=np.log(plot_data[col]),
                ax=ax,
            )
        else:
            sns.lineplot(
                x=np.arange(len(plot_data)) + 1,
                y=plot_data[col],
                ax=ax,
            )

    fig.tight_layout()

    return fig

if __name__ == "__main__":
    true_x = np.array([0.08, 0.15, 0.22, 0.31, 0.37, 0.28])
    n_params = len(true_x)
    start_x = np.full(n_params, 0.23)
    lower_bounds = np.zeros(n_params)
    upper_bounds = np.ones(n_params)

    mean = np.zeros(n_params)
    corr = 0.25
    cov = np.eye(n_params) * (1 - corr) + np.ones((n_params, n_params)) * corr
    np.random.seed(1234)
    data = np.random.multivariate_normal(mean, cov, size=500)

    # ==================================================================================
    # Simple test
    # ==================================================================================

    scipy_test_func = partial(
        scipy_criterion_function, true_x=true_x, data=data, noise_level=0
    )
    test_func = partial(criterion_function, true_x=true_x, data=data, noise_level=0)

    gradient_weight = 0.6

    res = minimize_manfred(
        func=test_func,
        x=start_x,
        xtol=0.001,
        step_sizes=[0.1, 0.05, 0.0125],
        max_fun=10_000,
        lower_bounds=lower_bounds,
        upper_bounds=upper_bounds,
        max_step_sizes=[1, 0.2, 0.1],
        linesearch_n_points=12,
        gradient_weight=gradient_weight,
    )

    scipy_res = scipy_minimize(
        scipy_test_func, start_x, method="Nelder-Mead", options={"maxfev": 100_000}
    )

    fig = plot_history(res)

    fig.savefig(Path(__file__).resolve().parent / "convergence_plot.pdf")

    print("Noise Free Test:           ")  # noqa: T001
    print("Manfred Solution:     ", res["solution_x"].round(2))  # noqa: T001
    print("True Solution:        ", true_x.round(2))  # noqa: T001
    print("Nelder Mead Solution: ", scipy_res.x.round(2))  # noqa: T001
    print("Manfred n_evals:      ", res["n_criterion_evaluations"])  # noqa: T001
    print("Nelder Mead n_evals:  ", scipy_res.nfev, "\n")  # noqa: T001

    # ==================================================================================
    # Very noisy test
    # ==================================================================================
    noise_level = 0.15
    noisy_test_func = partial(
        criterion_function, true_x=true_x, data=data, noise_level=noise_level
    )

    gradient_weight = 0.4

    res = minimize_manfred(
        func=noisy_test_func,
        x=start_x,
        xtol=0.001,
        step_sizes=[0.1, 0.05, 0.02],
        max_fun=1_000_000,
        lower_bounds=lower_bounds,
        upper_bounds=upper_bounds,
        linesearch_active=[True, True, True],
        max_step_sizes=[0.3, 0.2, 0.1],
        linesearch_n_points=12,
        n_evaluations_per_x=[50, 90, 120],
        gradient_weight=gradient_weight,
    )

    fig = plot_history(res)

    fig.savefig(Path(__file__).resolve().parent / "very_noisy_convergence_plot.pdf")

    print("Very Noisy Test:           ")  # noqa: T001
    print("Manfred Solution:     ", res["solution_x"].round(2))  # noqa: T001
    print("True Solution:        ", true_x.round(2))  # noqa: T001
    print("Manfred n_evals:      ", res["n_criterion_evaluations"])  # noqa: T001

    # ==================================================================================
    # Noisy test
    # ==================================================================================

    gradient_weight = 0.5
    noise_level = 0.1
    noisy_test_func = partial(
        criterion_function,
        true_x=true_x,
        data=data,
        noise_level=noise_level,
    )

    res = minimize_manfred(
        func=noisy_test_func,
        x=start_x,
        xtol=0.001,
        step_sizes=[0.1, 0.05, 0.02],
        max_fun=1_000_000,
        lower_bounds=lower_bounds,
        upper_bounds=upper_bounds,
        max_step_sizes=[0.3, 0.2, 0.2],
        linesearch_n_points=12,
        n_evaluations_per_x=[50, 90, 120],
        gradient_weight=gradient_weight,
        direction_window=2,
    )

    fig = plot_history(res)

    fig.savefig(Path(__file__).resolve().parent / "noisy_convergence_plot.pdf")

    print("Noisy Test:           ")  # noqa: T001
    print("Manfred Solution:     ", res["solution_x"].round(2))  # noqa: T001
    print("True Solution:        ", true_x.round(2))  # noqa: T001
    print("Manfred n_evals:      ", res["n_criterion_evaluations"])  # noqa: T001

    # ==================================================================================
    # Simple test with estimagic interface
    # ==================================================================================

    gradient_weight = 0.6
    noise_level = 0
    params = pd.DataFrame()
    params["value"] = start_x
    params["lower_bound"] = lower_bounds
    params["upper_bound"] = upper_bounds

    estimagic_func = partial(
        estimagic_criterion_function,
        true_x=true_x,
        noise_level=noise_level,
        data=data,
    )

    algo_options = {
        "step_sizes": [0.1, 0.05, 0.0125],
        "max_step_sizes": [1, 0.2, 0.1],
        "linesearch_n_points": 12,
        "gradient_weight": gradient_weight,
        "convergence_relative_params_tolerance": 0.001,
    }

    estimagic_res = minimize(
        criterion=estimagic_func,
        params=params,
        algorithm=minimize_manfred_estimagic,
        algo_options=algo_options,
        logging=False,
    )

    print("Estimagic Test:       ")  # noqa: T001
    print("Manfred Solution:     ", estimagic_res["solution_x"].round(2))  # noqa: T001
    print("True Solution:        ", true_x.round(2))  # noqa: T001
    print(  # noqa: T001
        "Manfred n_evals:      ", estimagic_res["n_criterion_evaluations"]
    )

    # ==================================================================================
    # Noisy test with estimagic interface
    # ==================================================================================

    gradient_weight = 0.5
    noise_level = 0.1
    params = pd.DataFrame()
    params["value"] = start_x
    params["lower_bound"] = lower_bounds
    params["upper_bound"] = upper_bounds

    estimagic_func = partial(
        estimagic_criterion_function,
        true_x=true_x,
        noise_level=noise_level,
        data=data,
    )

    algo_options = {
        "step_sizes": [0.1, 0.05, 0.02],
        "max_step_sizes": [0.3, 0.2, 0.2],
        "linesearch_n_points": 12,
        "gradient_weight": gradient_weight,
        "noise_n_evaluations_per_x": [50, 90, 120],
        "convergence_relative_params_tolerance": 0.001,
        "direction_window": 3,
    }

    estimagic_res = minimize(
        criterion=estimagic_func,
        params=params,
        algorithm=minimize_manfred_estimagic,
        algo_options=algo_options,
        logging=False,
    )

    print("Estimagic Test:       ")  # noqa: T001
    print("Manfred Solution:     ", estimagic_res["solution_x"].round(2))  # noqa: T001
    print("True Solution:        ", true_x.round(2))  # noqa: T001
    print(  # noqa: T001
        "Manfred n_evals:      ", estimagic_res["n_criterion_evaluations"]
    )

reason: import numpy,from scipy


text:
import copy
import os
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path

import mmcv
import numpy as np
import torch
from mmcv.image.misc import tensor2imgs
from mmcv.runner import load_checkpoint
from mmcv.utils.config import Config

from mmdet.apis import init_detector

# from mmocr.apis.inference import model_inference

from mmcv.ops import RoIPool
from mmcv.parallel import collate, scatter
from mmdet.datasets import replace_ImageToTensor
from mmdet.datasets.pipelines import Compose

def disable_text_recog_aug_test(cfg, set_types=None):
    """Remove aug_test from test pipeline of text recognition.
    Args:
        cfg (mmcv.Config): Input config.
        set_types (list[str]): Type of dataset source. Should be
            None or sublist of ['test', 'val']

    Returns:
        cfg (mmcv.Config): Output config removing
            `MultiRotateAugOCR` in test pipeline.
    """
    assert set_types is None or isinstance(set_types, list)
    if set_types is None:
        set_types = ['val', 'test']
    for set_type in set_types:
        if cfg.data[set_type].pipeline[1].type == 'MultiRotateAugOCR':
            cfg.data[set_type].pipeline = [
                cfg.data[set_type].pipeline[0],
                *cfg.data[set_type].pipeline[1].transforms
            ]

    return cfg

def model_inference(model,
                    imgs,
                    ann=None,
                    batch_mode=False,
                    return_data=False):
    """Inference image(s) with the detector.

    Args:
        model (nn.Module): The loaded detector.
        imgs (str/ndarray or list[str/ndarray] or tuple[str/ndarray]):
            Either image files or loaded images.
        batch_mode (bool): If True, use batch mode for inference.
        ann (dict): Annotation info for key information extraction.
        return_data: Return postprocessed data.
    Returns:
        result (dict): Predicted results.
    """

    if isinstance(imgs, (list, tuple)):
        is_batch = True
        if not isinstance(imgs[0], (np.ndarray, str)):
            raise AssertionError('imgs must be strings or numpy arrays')

    elif isinstance(imgs, (np.ndarray, str)):
        imgs = [imgs]
        is_batch = False
    else:
        raise AssertionError('imgs must be strings or numpy arrays')

    is_ndarray = isinstance(imgs[0], np.ndarray)

    cfg = model.cfg

    if batch_mode:
        cfg = disable_text_recog_aug_test(cfg, set_types=['test'])

    device = next(model.parameters()).device  # model device

    if is_ndarray:
        cfg = cfg.copy()
        # set loading pipeline type
        cfg.data.test.pipeline[0].type = 'LoadImageFromNdarray'

    cfg.data.test.pipeline = replace_ImageToTensor(cfg.data.test.pipeline)
    test_pipeline = Compose(cfg.data.test.pipeline)

    datas = []
    for img in imgs:
        # prepare data
        if is_ndarray:
            # directly add img
            data = dict(img=img, ann_info=ann, bbox_fields=[])
        else:
            # add information into dict
            data = dict(
                img_info=dict(filename=img),
                img_prefix=None,
                ann_info=ann,
                bbox_fields=[])

        # build the data pipeline
        data = test_pipeline(data)
        # get tensor from list to stack for batch mode (text detection)
        if batch_mode:
            if cfg.data.test.pipeline[1].type == 'MultiScaleFlipAug':
                for key, value in data.items():
                    data[key] = value[0]
        datas.append(data)

    if isinstance(datas[0]['img'], list) and len(datas) > 1:
        raise Exception('aug test does not support '
                        f'inference with batch size '
                        f'{len(datas)}')

    data = collate(datas, samples_per_gpu=len(imgs))

    # process img_metas
    if isinstance(data['img_metas'], list):
        data['img_metas'] = [
            img_metas.data[0] for img_metas in data['img_metas']
        ]
    else:
        data['img_metas'] = data['img_metas'].data

    if isinstance(data['img'], list):
        data['img'] = [img.data for img in data['img']]
        if isinstance(data['img'][0], list):
            data['img'] = [img[0] for img in data['img']]
    else:
        data['img'] = data['img'].data

    # for KIE models
    if ann is not None:
        data['relations'] = data['relations'].data[0]
        data['gt_bboxes'] = data['gt_bboxes'].data[0]
        data['texts'] = data['texts'].data[0]
        data['img'] = data['img'][0]
        data['img_metas'] = data['img_metas'][0]

    if next(model.parameters()).is_cuda:
        # scatter to specified GPU
        data = scatter(data, [device])[0]
    else:
        for m in model.modules():
            assert not isinstance(
                m, RoIPool
            ), 'CPU inference with RoIPool is not supported currently.'

    # forward the model
    with torch.no_grad():
        results = model(return_loss=False, rescale=True, **data)

    if not is_batch:
        if not return_data:
            return results[0]
        return results[0], datas[0]
    else:
        if not return_data:
            return results
        return results, datas

# from mmocr.core.visualize import det_recog_show_result
# from mmocr.datasets.kie_dataset import KIEDataset

# from mmocr.datasets.pipelines.crop import crop_img

def crop_img(src_img, box, long_edge_pad_ratio=0.4, short_edge_pad_ratio=0.2):
    """Crop text region with their bounding box.

    Args:
        src_img (np.array): The original image.
        box (list[float | int]): Points of quadrangle.
        long_edge_pad_ratio (float): Box pad ratio for long edge
            corresponding to font size.
        short_edge_pad_ratio (float): Box pad ratio for short edge
            corresponding to font size.
    """
    # assert utils.is_type_list(box, float) or utils.is_type_list(box, int)
    assert len(box) == 8
    assert 0. <= long_edge_pad_ratio < 1.0
    assert 0. <= short_edge_pad_ratio < 1.0

    h, w = src_img.shape[:2]
    points_x = np.clip(np.array(box[0::2]), 0, w)
    points_y = np.clip(np.array(box[1::2]), 0, h)

    box_width = np.max(points_x) - np.min(points_x)
    box_height = np.max(points_y) - np.min(points_y)
    font_size = min(box_height, box_width)

    if box_height < box_width:
        horizontal_pad = long_edge_pad_ratio * font_size
        vertical_pad = short_edge_pad_ratio * font_size
    else:
        horizontal_pad = short_edge_pad_ratio * font_size
        vertical_pad = long_edge_pad_ratio * font_size

    left = np.clip(int(np.min(points_x) - horizontal_pad), 0, w)
    top = np.clip(int(np.min(points_y) - vertical_pad), 0, h)
    right = np.clip(int(np.max(points_x) + horizontal_pad), 0, w)
    bottom = np.clip(int(np.max(points_y) + vertical_pad), 0, h)

    dst_img = src_img[top:bottom, left:right]

    return dst_img

# from mmocr.models import build_detector

# from mmocr.utils.box_util import stitch_boxes_into_lines
# from mmocr.utils.fileio import list_from_file
# from mmocr.utils.model import revert_sync_batchnorm
from .box_util import stitch_boxes_into_lines
from .fileio import list_from_file
from .model import revert_sync_batchnorm

# Parse CLI arguments
def parse_args():
    parser = ArgumentParser()
    parser.add_argument(
        'img', type=str, help='Input image file or folder path.')
    parser.add_argument(
        '--output',
        type=str,
        default='',
        help='Output file/folder name for visualization')
    parser.add_argument(
        '--det',
        type=str,
        default='PANet_IC15',
        help='Pretrained text detection algorithm')
    parser.add_argument(
        '--det-config',
        type=str,
        default='',
        help='Path to the custom config file of the selected det model. It '
        'overrides the settings in det')
    parser.add_argument(
        '--det-ckpt',
        type=str,
        default='',
        help='Path to the custom checkpoint file of the selected det model. '
        'It overrides the settings in det')
    parser.add_argument(
        '--recog',
        type=str,
        default='SEG',
        help='Pretrained text recognition algorithm')
    parser.add_argument(
        '--recog-config',
        type=str,
        default='',
        help='Path to the custom config file of the selected recog model. It'
        'overrides the settings in recog')
    parser.add_argument(
        '--recog-ckpt',
        type=str,
        default='',
        help='Path to the custom checkpoint file of the selected recog model. '
        'It overrides the settings in recog')
    parser.add_argument(
        '--kie',
        type=str,
        default='',
        help='Pretrained key information extraction algorithm')
    parser.add_argument(
        '--kie-config',
        type=str,
        default='',
        help='Path to the custom config file of the selected kie model. It'
        'overrides the settings in kie')
    parser.add_argument(
        '--kie-ckpt',
        type=str,
        default='',
        help='Path to the custom checkpoint file of the selected kie model. '
        'It overrides the settings in kie')
    parser.add_argument(
        '--config-dir',
        type=str,
        default=os.path.join(str(Path.cwd()), 'configs/'),
        help='Path to the config directory where all the config files '
        'are located. Defaults to "configs/"')
    parser.add_argument(
        '--batch-mode',
        action='store_true',
        help='Whether use batch mode for inference')
    parser.add_argument(
        '--recog-batch-size',
        type=int,
        default=0,
        help='Batch size for text recognition')
    parser.add_argument(
        '--det-batch-size',
        type=int,
        default=0,
        help='Batch size for text detection')
    parser.add_argument(
        '--single-batch-size',
        type=int,
        default=0,
        help='Batch size for separate det/recog inference')
    parser.add_argument(
        '--device', default='cuda:0', help='Device used for inference.')
    parser.add_argument(
        '--export',
        type=str,
        default='',
        help='Folder where the results of each image are exported')
    parser.add_argument(
        '--export-format',
        type=str,
        default='json',
        help='Format of the exported result file(s)')
    parser.add_argument(
        '--details',
        action='store_true',
        help='Whether include the text boxes coordinates and confidence values'
    )
    parser.add_argument(
        '--imshow',
        action='store_true',
        help='Whether show image with OpenCV.')
    parser.add_argument(
        '--print-result',
        action='store_true',
        help='Prints the recognised text')
    parser.add_argument(
        '--merge', action='store_true', help='Merge neighboring boxes')
    parser.add_argument(
        '--merge-xdist',
        type=float,
        default=20,
        help='The maximum x-axis distance to merge boxes')
    args = parser.parse_args()
    if args.det == 'None':
        args.det = None
    if args.recog == 'None':
        args.recog = None
    # Warnings
    if args.merge and not (args.det and args.recog):
        warnings.warn(
            'Box merging will not work if the script is not'
            ' running in detection + recognition mode.', UserWarning)
    if not os.path.samefile(args.config_dir, os.path.join(str(
            Path.cwd()))) and (args.det_config != ''
                               or args.recog_config != ''):
        warnings.warn(
            'config_dir will be overrided by det-config or recog-config.',
            UserWarning)
    return args

class MMOCR:

    def __init__(self,
                 det='PANet_IC15',
                 det_config='',
                 det_ckpt='',
                 recog='SEG',
                 recog_config='',
                 recog_ckpt='',
                 kie='',
                 kie_config='',
                 kie_ckpt='',
                 config_dir=os.path.join(str(Path.cwd()), 'configs/'),
                 device='cuda:0',
                 **kwargs):

        textdet_models = {
            'DB_r18': {
                'config':
                'dbnet/dbnet_r18_fpnc_1200e_icdar2015.py',
                'ckpt':
                'dbnet/'
                'dbnet_r18_fpnc_sbn_1200e_icdar2015_20210329-ba3ab597.pth'
            },
            'DB_r50': {
                'config':
                'dbnet/dbnet_r50dcnv2_fpnc_1200e_icdar2015.py',
                'ckpt':
                'dbnet/'
                'dbnet_r50dcnv2_fpnc_sbn_1200e_icdar2015_20210325-91cef9af.pth'
            },
            'DRRG': {
                'config': 'drrg/drrg_r50_fpn_unet_1200e_ctw1500.py',
                'ckpt': 'drrg/drrg_r50_fpn_unet_1200e_ctw1500-1abf4f67.pth'
            },
            'FCE_IC15': {
                'config': 'fcenet/fcenet_r50_fpn_1500e_icdar2015.py',
                'ckpt': 'fcenet/fcenet_r50_fpn_1500e_icdar2015-d435c061.pth'
            },
            'FCE_CTW_DCNv2': {
                'config': 'fcenet/fcenet_r50dcnv2_fpn_1500e_ctw1500.py',
                'ckpt': 'fcenet/fcenet_r50dcnv2_fpn_1500e_ctw1500-05d740bb.pth'
            },
            'MaskRCNN_CTW': {
                'config':
                'maskrcnn/mask_rcnn_r50_fpn_160e_ctw1500.py',
                'ckpt':
                'maskrcnn/'
                'mask_rcnn_r50_fpn_160e_ctw1500_20210219-96497a76.pth'
            },
            'MaskRCNN_IC15': {
                'config':
                'maskrcnn/mask_rcnn_r50_fpn_160e_icdar2015.py',
                'ckpt':
                'maskrcnn/'
                'mask_rcnn_r50_fpn_160e_icdar2015_20210219-8eb340a3.pth'
            },
            'MaskRCNN_IC17': {
                'config':
                'maskrcnn/mask_rcnn_r50_fpn_160e_icdar2017.py',
                'ckpt':
                'maskrcnn/'
                'mask_rcnn_r50_fpn_160e_icdar2017_20210218-c6ec3ebb.pth'
            },
            'PANet_CTW': {
                'config':
                'panet/panet_r18_fpem_ffm_600e_ctw1500.py',
                'ckpt':
                'panet/'
                'panet_r18_fpem_ffm_sbn_600e_ctw1500_20210219-3b3a9aa3.pth'
            },
            'PANet_IC15': {
                'config':
                'panet/panet_r18_fpem_ffm_600e_icdar2015.py',
                'ckpt':
                'panet/'
                'panet_r18_fpem_ffm_sbn_600e_icdar2015_20210219-42dbe46a.pth'
            },
            'PS_CTW': {
                'config': 'psenet/psenet_r50_fpnf_600e_ctw1500.py',
                'ckpt':
                'psenet/psenet_r50_fpnf_600e_ctw1500_20210401-216fed50.pth'
            },
            'PS_IC15': {
                'config':
                'psenet/psenet_r50_fpnf_600e_icdar2015.py',
                'ckpt':
                'psenet/psenet_r50_fpnf_600e_icdar2015_pretrain-eefd8fe6.pth'
            },
            'TextSnake': {
                'config':
                'textsnake/textsnake_r50_fpn_unet_1200e_ctw1500.py',
                'ckpt':
                'textsnake/textsnake_r50_fpn_unet_1200e_ctw1500-27f65b64.pth'
            }
        }

        textrecog_models = {
            'CRNN': {
                'config': 'crnn/crnn_academic_dataset.py',
                'ckpt': 'crnn/crnn_academic-a723a1c5.pth'
            },
            'SAR': {
                'config': 'sar/sar_r31_parallel_decoder_academic.py',
                'ckpt': 'sar/sar_r31_parallel_decoder_academic-dba3a4a3.pth'
            },
            'NRTR_1/16-1/8': {
                'config': 'nrtr/nrtr_r31_1by16_1by8_academic.py',
                'ckpt': 'nrtr/nrtr_r31_academic_20210406-954db95e.pth'
            },
            'NRTR_1/8-1/4': {
                'config': 'nrtr/nrtr_r31_1by8_1by4_academic.py',
                'ckpt':
                'nrtr/nrtr_r31_1by8_1by4_academic_20210406-ce16e7cc.pth'
            },
            'RobustScanner': {
                'config': 'robust_scanner/robustscanner_r31_academic.py',
                'ckpt': 'robustscanner/robustscanner_r31_academic-5f05874f.pth'
            },
            'SEG': {
                'config': 'seg/seg_r31_1by16_fpnocr_academic.py',
                'ckpt': 'seg/seg_r31_1by16_fpnocr_academic-72235b11.pth'
            },
            'CRNN_TPS': {
                'config': 'tps/crnn_tps_academic_dataset.py',
                'ckpt': 'tps/crnn_tps_academic_dataset_20210510-d221a905.pth'
            }
        }

        kie_models = {
            'SDMGR': {
                'config': 'sdmgr/sdmgr_unet16_60e_wildreceipt.py',
                'ckpt':
                'sdmgr/sdmgr_unet16_60e_wildreceipt_20210520-7489e6de.pth'
            }
        }

        self.td = det
        self.tr = recog
        self.kie = kie
        self.device = device

        # Check if the det/recog model choice is valid
        if self.td and self.td not in textdet_models:
            raise ValueError(self.td,
                             'is not a supported text detection algorthm')
        elif self.tr and self.tr not in textrecog_models:
            raise ValueError(self.tr,
                             'is not a supported text recognition algorithm')
        elif self.kie:
            if self.kie not in kie_models:
                raise ValueError(
                    self.kie, 'is not a supported key information extraction'
                    ' algorithm')
            elif not (self.td and self.tr):
                raise NotImplementedError(
                    self.kie, 'has to run together'
                    ' with text detection and recognition algorithms.')

        self.detect_model = None
        if self.td:
            # Build detection model
            if not det_config:
                det_config = os.path.join(config_dir, 'textdet/',
                                          textdet_models[self.td]['config'])
            if not det_ckpt:
                det_ckpt = 'https://download.openmmlab.com/mmocr/textdet/' + \
                    textdet_models[self.td]['ckpt']

            self.detect_model = init_detector(
                det_config, det_ckpt, device=self.device)
            self.detect_model = revert_sync_batchnorm(self.detect_model)

        self.recog_model = None
        if self.tr:
            # Build recognition model
            if not recog_config:
                recog_config = os.path.join(
                    config_dir, 'textrecog/',
                    textrecog_models[self.tr]['config'])
            if not recog_ckpt:
                recog_ckpt = 'https://download.openmmlab.com/mmocr/' + \
                    'textrecog/' + textrecog_models[self.tr]['ckpt']

            self.recog_model = init_detector(
                recog_config, recog_ckpt, device=self.device)
            self.recog_model = revert_sync_batchnorm(self.recog_model)

        self.kie_model = None
        # if self.kie:
        #     # Build key information extraction model
        #     if not kie_config:
        #         kie_config = os.path.join(config_dir, 'kie/',
        #                                   kie_models[self.kie]['config'])
        #     if not kie_ckpt:
        #         kie_ckpt = 'https://download.openmmlab.com/mmocr/' + \
        #             'kie/' + kie_models[self.kie]['ckpt']

        #     kie_cfg = Config.fromfile(kie_config)
        #     self.kie_model = build_detector(
        #         kie_cfg.model, test_cfg=kie_cfg.get('test_cfg'))
        #     self.kie_model = revert_sync_batchnorm(self.kie_model)
        #     self.kie_model.cfg = kie_cfg
        #     load_checkpoint(self.kie_model, kie_ckpt, map_location=self.device)

        # Attribute check
        for model in list(filter(None, [self.recog_model, self.detect_model])):
            if hasattr(model, 'module'):
                model = model.module
            if model.cfg.data.test['type'] == 'ConcatDataset':
                model.cfg.data.test.pipeline = \
                    model.cfg.data.test['datasets'][0].pipeline

    def readtext(self,
                 img,
                 output=None,
                 details=False,
                 export=None,
                 export_format='json',
                 batch_mode=False,
                 recog_batch_size=0,
                 det_batch_size=0,
                 single_batch_size=0,
                 imshow=False,
                 print_result=False,
                 merge=False,
                 merge_xdist=20,
                 **kwargs):
        args = locals()
        [args.pop(x, None) for x in ['kwargs', 'self']]
        args = Namespace(**args)

        # Input and output arguments processing
        self._args_processing(args)
        self.args = args

        pp_result = None

        # Send args and models to the MMOCR model inference API
        # and call post-processing functions for the output
        if self.detect_model and self.recog_model:
            det_recog_result = self.det_recog_kie_inference(
                self.detect_model, self.recog_model, kie_model=self.kie_model)
            pp_result = self.det_recog_pp(det_recog_result)
        else:
            for model in list(
                    filter(None, [self.recog_model, self.detect_model])):
                result = self.single_inference(model, args.arrays,
                                               args.batch_mode,
                                               args.single_batch_size)
                pp_result = self.single_pp(result, model)

        return pp_result

    # Post processing function for end2end ocr
    def det_recog_pp(self, result):
        final_results = []
        args = self.args
        for arr, output, export, det_recog_result in zip(
                args.arrays, args.output, args.export, result):
            # if output or args.imshow:
            #     if self.kie_model:
            #         res_img = det_recog_show_result(arr, det_recog_result)
            #     else:
            #         res_img = det_recog_show_result(
            #             arr, det_recog_result, out_file=output)
            #     if args.imshow and not self.kie_model:
            #         mmcv.imshow(res_img, 'inference results')
            if not args.details:
                simple_res = {}
                simple_res['filename'] = det_recog_result['filename']
                simple_res['text'] = [
                    x['text'] for x in det_recog_result['result']
                ]
                final_result = simple_res
            else:
                final_result = det_recog_result
            if export:
                mmcv.dump(final_result, export, indent=4)
            if args.print_result:
                print(final_result, end='\n\n')
            final_results.append(final_result)
        return final_results

    # Post processing function for separate det/recog inference
    def single_pp(self, result, model):
        for arr, output, export, res in zip(self.args.arrays, self.args.output,
                                            self.args.export, result):
            if export:
                mmcv.dump(res, export, indent=4)
            if output or self.args.imshow:
                res_img = model.show_result(arr, res, out_file=output)
                if self.args.imshow:
                    mmcv.imshow(res_img, 'inference results')
            if self.args.print_result:
                print(res, end='\n\n')
        return result

    def generate_kie_labels(self, result, boxes, class_list):
        idx_to_cls = {}
        if class_list is not None:
            for line in list_from_file(class_list):
                class_idx, class_label = line.strip().split()
                idx_to_cls[class_idx] = class_label

        max_value, max_idx = torch.max(result['nodes'].detach().cpu(), -1)
        node_pred_label = max_idx.numpy().tolist()
        node_pred_score = max_value.numpy().tolist()
        labels = []
        for i in range(len(boxes)):
            pred_label = str(node_pred_label[i])
            if pred_label in idx_to_cls:
                pred_label = idx_to_cls[pred_label]
            pred_score = node_pred_score[i]
            labels.append((pred_label, pred_score))
        return labels

    def visualize_kie_output(self,
                             model,
                             data,
                             result,
                             out_file=None,
                             show=False):
        """Visualizes KIE output."""
        img_tensor = data['img'].data
        img_meta = data['img_metas'].data
        gt_bboxes = data['gt_bboxes'].data.numpy().tolist()
        img = tensor2imgs(img_tensor.unsqueeze(0),
                          **img_meta['img_norm_cfg'])[0]
        h, w, _ = img_meta['img_shape']
        img_show = img[:h, :w, :]
        model.show_result(
            img_show, result, gt_bboxes, show=show, out_file=out_file)

    # End2end ocr inference pipeline
    def det_recog_kie_inference(self, det_model, recog_model, kie_model=None):
        end2end_res = []
        # Find bounding boxes in the images (text detection)
        det_result = self.single_inference(det_model, self.args.arrays,
                                           self.args.batch_mode,
                                           self.args.det_batch_size)
        bboxes_list = [res['boundary_result'] for res in det_result]

        # if kie_model:
        #     kie_dataset = KIEDataset(
        #         dict_file=kie_model.cfg.data.test.dict_file)

        # For each bounding box, the image is cropped and
        # sent to the recognition model either one by one
        # or all together depending on the batch_mode
        for filename, arr, bboxes, out_file in zip(self.args.filenames,
                                                   self.args.arrays,
                                                   bboxes_list,
                                                   self.args.output):
            img_e2e_res = {}
            img_e2e_res['filename'] = filename
            img_e2e_res['result'] = []
            box_imgs = []
            for bbox in bboxes:
                box_res = {}
                box_res['box'] = [round(x) for x in bbox[:-1]]
                box_res['box_score'] = float(bbox[-1])
                box = bbox[:8]
                if len(bbox) > 9:
                    min_x = min(bbox[0:-1:2])
                    min_y = min(bbox[1:-1:2])
                    max_x = max(bbox[0:-1:2])
                    max_y = max(bbox[1:-1:2])
                    box = [
                        min_x, min_y, max_x, min_y, max_x, max_y, min_x, max_y
                    ]
                box_img = crop_img(arr, box)
                if self.args.batch_mode:
                    box_imgs.append(box_img)
                else:
                    recog_result = model_inference(recog_model, box_img)
                    text = recog_result['text']
                    text_score = recog_result['score']
                    if isinstance(text_score, list):
                        text_score = sum(text_score) / max(1, len(text))
                    box_res['text'] = text
                    box_res['text_score'] = text_score
                img_e2e_res['result'].append(box_res)

            if self.args.batch_mode:
                recog_results = self.single_inference(
                    recog_model, box_imgs, True, self.args.recog_batch_size)
                for i, recog_result in enumerate(recog_results):
                    text = recog_result['text']
                    text_score = recog_result['score']
                    if isinstance(text_score, (list, tuple)):
                        text_score = sum(text_score) / max(1, len(text))
                    img_e2e_res['result'][i]['text'] = text
                    img_e2e_res['result'][i]['text_score'] = text_score

            if self.args.merge:
                img_e2e_res['result'] = stitch_boxes_into_lines(
                    img_e2e_res['result'], self.args.merge_xdist, 0.5)

            if kie_model:
                annotations = copy.deepcopy(img_e2e_res['result'])
                # Customized for kie_dataset, which
                # assumes that boxes are represented by only 4 points
                for i, ann in enumerate(annotations):
                    min_x = min(ann['box'][::2])
                    min_y = min(ann['box'][1::2])
                    max_x = max(ann['box'][::2])
                    max_y = max(ann['box'][1::2])
                    annotations[i]['box'] = [
                        min_x, min_y, max_x, min_y, max_x, max_y, min_x, max_y
                    ]
                ann_info = kie_dataset._parse_anno_info(annotations)
                kie_result, data = model_inference(
                    kie_model,
                    arr,
                    ann=ann_info,
                    return_data=True,
                    batch_mode=self.args.batch_mode)
                # visualize KIE results
                self.visualize_kie_output(
                    kie_model,
                    data,
                    kie_result,
                    out_file=out_file,
                    show=self.args.imshow)
                gt_bboxes = data['gt_bboxes'].data.numpy().tolist()
                labels = self.generate_kie_labels(kie_result, gt_bboxes,
                                                  kie_model.class_list)
                for i in range(len(gt_bboxes)):
                    img_e2e_res['result'][i]['label'] = labels[i][0]
                    img_e2e_res['result'][i]['label_score'] = labels[i][1]

            end2end_res.append(img_e2e_res)
        return end2end_res

    # Separate det/recog inference pipeline
    def single_inference(self, model, arrays, batch_mode, batch_size=0):
        result = []
        if batch_mode:
            if batch_size == 0:
                result = model_inference(model, arrays, batch_mode=True)
            else:
                n = batch_size
                arr_chunks = [
                    arrays[i:i + n] for i in range(0, len(arrays), n)
                ]
                for chunk in arr_chunks:
                    result.extend(
                        model_inference(model, chunk, batch_mode=True))
        else:
            for arr in arrays:
                result.append(model_inference(model, arr, batch_mode=False))
        return result

    # Arguments pre-processing function
    def _args_processing(self, args):
        # Check if the input is a list/tuple that
        # contains only np arrays or strings
        # if isinstance(args.img, (list, tuple)):
        #     img_list = args.img
        #     if not all([isinstance(x, (np.ndarray, str)) for x in args.img]):
        #         raise AssertionError('Images must be strings or numpy arrays')

        # # Create a list of the images
        # if isinstance(args.img, str):
        #     img_path = Path(args.img)
        #     if img_path.is_dir():
        #         img_list = [str(x) for x in img_path.glob('*')]
        #     else:
        #         img_list = [str(img_path)]
        # elif isinstance(args.img, np.ndarray):
        #     img_list = [args.img]

        # Read all image(s) in advance to reduce wasted time
        # re-reading the images for vizualisation output
        # args.arrays = [mmcv.imread(x) for x in img_list]
        args.arrays = args.img

        # Create a list of filenames (used for output images and result files)
        # if isinstance(img_list[0], str):
        #     args.filenames = [str(Path(x).stem) for x in img_list]
        # else:
        #     args.filenames = [str(x) for x in range(len(img_list))]

        # If given an output argument, create a list of output image filenames
        num_res = len(args.img)
        if args.output:
            output_path = Path(args.output)
            if output_path.is_dir():
                args.output = [
                    str(output_path / f'out_{x}.png') for x in args.filenames
                ]
            else:
                args.output = [str(args.output)]
                if args.batch_mode:
                    raise AssertionError('Output of multiple images inference'
                                         ' must be a directory')
        else:
            args.output = [None] * num_res

        # If given an export argument, create a list of
        # result filenames for each image
        if args.export:
            export_path = Path(args.export)
            args.export = [
                str(export_path / f'out_{x}.{args.export_format}')
                for x in args.filenames
            ]
        else:
            args.export = [None] * num_res

        return args

# Create an inference pipeline with parsed arguments
def main():
    args = parse_args()
    ocr = MMOCR(**vars(args))
    ocr.readtext(**vars(args))

if __name__ == '__main__':
    main()

ext: py

max_stars_repo_name: PhiDCH/mmocr

max_issues_repo_name: PhiDCH/mmocr

max_forks_repo_name: PhiDCH/mmocr

reason: import numpy


text:
"""
Created on 27/May/2021

Contains processing of trajectories in Trajnet++ format. This data comes in ndjson file and has the following types of
data:
- A scene, identifying an interval of frames where there are one or more trajectories. The trajectory has one pedestrian
called the primary pedestrian ("p") - the most relevant pedestrian in the scene; in terms of computing metrics in that
scene, it will only be done for the primary pedestrian, the other pedestrians serve more as social context. It also
has a type of trajectory (I:static, II:linear, III:interaction, IV:non-interacting) and subtype for type III:interaction
(I:leader-follower, II:Collision-Avoidance, III:Group, IV:Other)
{“scene”: {“id”: 266, “p”: 254, “s”: 10238, “e”: 10358, “fps”: 2.5, “tag”: [2, []]}}
- A track, a certain position of a certain pedestrian. A track can belong to more than one scene, in the case that
scenes have overlapping intervals [s_scene1, e_scene1] and [s_scene2, e_scene2]
{“track”: {“f”: 10238, “p”: 248, “x”: 13.2, “y”: 5.85}}
    When a model predicts trajectories and outputs to a ndjson file, it has information like the prediction number
    (can be > 0 in case of multimodality), and the identifier of the scene the track belongs to
    {“track”: {“f”: 10238, “p”: 248, “x”: 13.2, “y”: 5.85, “pred_number”: 0, “scene_id”: 123}}

For more information about how these trajectories are made and structured, and also to learn more about Trajnet++,
here are a few links of interest:
- https://www.aicrowd.com/challenges/trajnet-a-trajectory-forecasting-challenge
- https://thedebugger811.github.io/posts/2020/03/intro_trajnetpp/
- https://github.com/vita-epfl/trajnetplusplusdata/
- https://github.com/vita-epfl/trajnetplusplusdataset/
- https://github.com/vita-epfl/trajnetplusplustools/
- https://github.com/vita-epfl/trajnetplusplusbaselines/

CREDITS: The code is built from Trajnet++ repositories, so if you use this work, make sure you cite the work from
Trajnet++ authors:
@article{Kothari2020HumanTF,
  title={Human Trajectory Forecasting in Crowds: A Deep Learning Perspective},
  author={Parth Kothari and S. Kreiss and Alexandre Alahi},
  journal={ArXiv},
  year={2020},
  volume={abs/2007.03639}
}
"""
import os

import numpy as np
import torch
from torch.utils.data import DataLoader, Dataset
import trajnetplusplustools

from models.data.common_dataset import read_ndjson_file, Metadata
from models.utils.utils import random_rotation, center_scene

class MetadataTrajnetpp(Metadata):
    """
    Class to store metadata-related information regarding a trajectory or sequence of trajectories. This data is not
    100% essential for training / inference, but can provide additional information for auxiliary tasks
    """

    def __init__(self, origin_label, seq_id, rot_angle, center, prim_id, all_ids, size=0):
        """
        builds the Metadata object
        :param origin_label: label identifying the source of the trajectory (e.g. the file name)
        :param seq_id: ID that identifies the sequence of pedestrian trajectories (which is named scene in Trajnet++)
        :param rot_angle: The rotation angle of trajectories, in radius. Applies for cases where --normalize_scene or
        --augment_rotation command line arguments are supplied
        :param center: The displacement in xy coordinates. Applies for cases where --normalize_scene or
        :param prim_id: number identifying the primary pedestrian
        :param all_ids: list of numbers identifying all pedestrians, by order
        :param size: size (assumed MegaBytes/MB) in memory of the trajectories this metadata refers to
        """
        super(MetadataTrajnetpp, self).__init__(origin_label, size)
        self.id = seq_id
        self.primary_ped_id = prim_id
        self.all_ped_ids = all_ids
        self.angle = rot_angle
        self.center = center

class TrajnetppDataset(Dataset):
    """Dataset for the Trajectory datasets - Can handle Trajnetpp format of fixed length trajectories (although there
    is the possibility of having partial trajectories, for non-primary pedestrians).
    Each scene, which will be referred to as a sequence is said to be a sort of sub-batch
    (not to be confused with a batch returned at each step, that can contain multiple sequences in it), and comprises
    set of trajectories of the same length, synchronized in time.
    This idea of synchronized in time means that all those trajectories are present in the exact same time frame.
    For format of the trajectories, see the above file description comment, or go to the following link:
    https://www.aicrowd.com/challenges/trajnet-a-trajectory-forecasting-challenge
    """

    def __init__(self, data_location, device=torch.device('cpu'), obs_len=9, pred_len=12, filter_tags=(),
                 filter_sub_tags=(), consider_partial_trajectories=True, normalize_scene=False, augment_rotation=False):
        """
        Initializes the dataset contents from the file(s)
        :param data_location: Location of the dataset files in the format (can be a directory or a single file).
        Assumed .ndjson format. See https://www.aicrowd.com/challenges/trajnet-a-trajectory-forecasting-challenge
        :param device: device to map the tensors to. By default will go to cpu
        :param obs_len: Number of time-steps in input trajectories (to observe and process)
        :param pred_len: Number of time-steps in output trajectories (to perform prediction)
        :param filter_tags: list of tags to filter (i.e. to not consider). By default empty - consider all tags
        :param filter_sub_tags: list of sub-tags to filter. By default empty - consider all sub-tags
        :param consider_partial_trajectories: Use this flag if meant to include partial trajectories (of length smaller
        than obs_len+pred_len). For the instants without positions, will fill with NaN.
        :param
        """
        super(TrajnetppDataset, self).__init__()

        self.data_location = data_location
        self.filter_tags, self.filter_sub_tags = filter_tags, filter_sub_tags
        self.consider_partial_trajectories = consider_partial_trajectories
        self.normalize_scene, self.augment_rotation = normalize_scene, augment_rotation

        self.total_num_seqs = 0  # total number of sequences of trajectories spanned across time.

        if os.path.isdir(self.data_location):
            self.all_files = sorted(os.listdir(self.data_location))
            self.all_files = [os.path.join(self.data_location, _path) for _path in self.all_files]
        else:
            # is a single file
            self.all_files = [self.data_location]
        self.num_peds_in_seq = []
        self.obs_seq_list = []
        self.obs_seq_list_rel = []
        self.pred_seq_list = []
        self.pred_seq_list_rel = []
        self.seq_start_end_list = []
        self.metadata_list = []

        self.num_distinct_primary_pedestrians = 0
        self.seen_primary_ped = {}

        for path in self.all_files:
            data_reader = read_ndjson_file(path)
            self.seen_primary_ped[path] = []
            for scene_id, scene in data_reader.scenes():
                [main_tag, sub_tags_list] = data_reader.scenes_by_id[scene_id].tag
                if main_tag not in self.filter_tags or (
                        sub_tags_list and not any([s in self.filter_sub_tags for s in sub_tags_list])):
                    continue  # not in list of tags/sub-tags to filter
                ped_id_list = [t[0].pedestrian for t in scene]
                primary_ped_id = scene[0][0].pedestrian
                if primary_ped_id not in self.seen_primary_ped[path]:
                    self.num_distinct_primary_pedestrians += 1
                    self.seen_primary_ped[path].append(primary_ped_id)
                data = trajnetplusplustools.Reader.paths_to_xy(scene)
                # tensor of shape [obs_len+pred_len,num_peds,2]
                seq = torch.from_numpy(data).to(device).to(torch.float32)
                if seq.shape[0] != (obs_len + pred_len):
                    raise Exception(f'Got trajectory length {seq.shape[0]}, but expected length of '
                                    f'{obs_len}+{pred_len}={obs_len + pred_len}')
                if not self.consider_partial_trajectories:
                    # discard the trajectories that have nan values
                    seq = seq[:, torch.any(torch.all(~torch.isnan(seq), dim=0), dim=1), :]
                rot, center = None, None
                if normalize_scene:
                    seq, rot, center = center_scene(seq, obs_length=obs_len)
                elif augment_rotation:
                    seq, rot = random_rotation(seq)
                seq_rel = torch.zeros_like(seq)
                seq_rel[1:, :, :] = seq[1:, :, :] - seq[:-1, :, :]
                self.obs_seq_list.append(seq[:obs_len, :, :])
                self.obs_seq_list_rel.append(seq_rel[:obs_len, :, :])
                self.pred_seq_list.append(seq[obs_len:, :, :])
                self.pred_seq_list_rel.append(seq_rel[obs_len:, :, :])
                # primary pedestrian will always be the one with id 0 - the 'start' in seq_start_end
                # in trajnetpp repositories - seq_start_end <=> batch_split
                num_peds = seq.shape[1]
                if self.seq_start_end_list:
                    new_start = self.seq_start_end_list[-1][-1]
                    self.seq_start_end_list.append([new_start, new_start + num_peds])
                else:
                    self.seq_start_end_list.append([0, num_peds])
                # for the case where multiple files are loaded, one needs to distinguish between several scenes
                self.num_peds_in_seq.append(num_peds)
                self.total_num_seqs += 1
                tensors_size = seq.element_size() * seq.nelement() + seq_rel.element_size() * seq_rel.nelement()
                metadata = MetadataTrajnetpp(path, scene_id, rot, center, primary_ped_id, ped_id_list,
                                             size=tensors_size / (1024 * 1024))  # size in MB
                self.metadata_list.append(metadata)

    def __len__(self):
        return self.total_num_seqs

    def __getitem__(self, index):
        num_peds = self.obs_seq_list[index].shape[1]
        traj_len = self.obs_seq_list[index].shape[0] + self.pred_seq_list[index].shape[0]
        # non_linear_peds and loss_mask doesn't really apply to Trajnet++ format
        out = [
            self.obs_seq_list[index], self.pred_seq_list[index],
            self.obs_seq_list_rel[index], self.pred_seq_list_rel[index],
            self.metadata_list[index], torch.ones(num_peds, traj_len, device=self.obs_seq_list[index].device)
        ]
        return out

class TrajnetppPredictionsDataset(Dataset):
    """

    """

    def __init__(self, data_location, device=torch.device('cpu'), pred_len=12, consider_partial_trajectories=True,
                 num_samples=1):
        self.total_num_seqs = 0
        self.data_location = data_location
        self.consider_partial_trajectories = consider_partial_trajectories
        self.num_samples = num_samples

        self.total_num_seqs = 0  # total number of sequences of trajectories spanned across time.

        if os.path.isdir(self.data_location):
            self.all_files = sorted(os.listdir(self.data_location))
            self.all_files = [os.path.join(self.data_location, _path) for _path in self.all_files]
        else:
            # is a single file
            self.all_files = [self.data_location]
        self.num_peds_in_seq = []
        self.pred_seq_list = []
        self.seq_start_end_list = []
        self.metadata_list = []

        for path in self.all_files:
            data_reader = read_ndjson_file(path)
            for scene_id, scene in data_reader.scenes():
                ped_id_list = [t[0].pedestrian for t in scene]
                primary_ped_id = scene[0][0].pedestrian
                self.num_peds_in_seq.append(len(ped_id_list))
                num_pred = max([row.prediction_number for row in scene[0]]) + 1
                if num_pred < self.num_samples:
                    raise Exception(f'Some predictions in {path} lack the required number of samples '
                                    f'({self.num_samples})')
                # if there are more predictions than self.num_samples, those remaining will be discarded
                pred_list_for_scene = []
                for s in range(self.num_samples):
                    scene_sample_s = [[r for r in track if r.prediction_number == s and r.scene_id == scene_id]
                                      for track in scene]
                    # sometimes predictions from different scenes may be mixed, and the above line may cause "empty"
                    #   lists to be left in the variable, which causes the program to crash. The line below avoids it
                    scene_sample_s = [track for track in scene_sample_s if len(track) > 0]
                    data = trajnetplusplustools.Reader.paths_to_xy(scene_sample_s)
                    pred_sample = torch.from_numpy(data).to(device).to(torch.float32)
                    pred_list_for_scene.append(pred_sample)
                # tensor of shape [obs_len+pred_len,num_peds,2]
                if self.num_samples == 1:
                    pred = pred_list_for_scene[0]
                    if pred.shape[0] != pred_len:
                        raise Exception(f'Got trajectory length {pred.shape[0]}, but expected length {pred_len}')
                else:
                    pred = torch.cat([p.unsqueeze(0) for p in pred_list_for_scene])
                self.pred_seq_list.append(pred)
                new_start = self.seq_start_end_list[-1][-1] if self.seq_start_end_list else 0
                self.seq_start_end_list.append([new_start, new_start + len(ped_id_list)])
                tensors_size = pred.element_size() * pred.nelement()
                metadata = MetadataTrajnetpp(path, scene_id, None, None, primary_ped_id, ped_id_list,
                                             size=tensors_size / (1024 * 1024))  # size in MB
                self.metadata_list.append(metadata)
                self.total_num_seqs += 1

    def __getitem__(self, index):
        return self.pred_seq_list[index], self.metadata_list[index]

    def __len__(self):
        return self.total_num_seqs

def seq_collate(data):
    """
    Collect the several fields of a sequence and combine them for direct use by models (collate <-> collect and combine)
    :param data: the original data, not formatted
    :return: a tuple containing the several parts of the data:
        - obs_traj: Tensor of shape (obs_seq_len, batch, 2). The observed trajectory, in absolute coordinates
        - pred_traj: Tensor of shape (pred_seq_len, batch, 2). The predicted trajectory, in absolute coordinates
        - obs_traj_rel: Tensor of shape (obs_seq_len, batch, 2). The observed trajectory, in relative displacements
        - pred_traj_rel: Tensor of shape (pred_seq_len, batch, 2). The predicted trajectory, in relative displacements
        - non_linear_ped: Tensor of shape (batch). 1 if trajectory is non linear, 0 otherwise
        - loss_mask: Tensor of shape (traj_seq_len, batch). Loss mask to (optionally) use in training.
        - seq_start_end: Tensor containing the beginning and end (in terms of pedestrians in this batch) of a sequence.
            A sequence corresponds to a set of trajectories that belong to the same time frame. Note that the batch may
            have several sequences.
    """
    if len(data[0]) < 6:
        # It is not actual data, but predictions stored in files
        (pred_seq_list, metadata_list) = zip(*data)
        p_index = 1 if pred_seq_list[0].ndim < 4 else 2
        _len = [seq.shape[p_index] for seq in pred_seq_list]
        cum_start_idx = [0] + np.cumsum(_len).tolist()
        seq_start_end = [[start, end] for start, end in zip(cum_start_idx, cum_start_idx[1:])]
        seq_start_end = torch.tensor(seq_start_end, device=pred_seq_list[0].device, dtype=torch.long)
        return torch.cat(pred_seq_list, dim=p_index), seq_start_end, metadata_list

    (obs_seq_list, pred_seq_list, obs_seq_rel_list, pred_seq_rel_list, metadata_list, loss_mask_list) = zip(*data)

    _len = [seq.shape[1] for seq in obs_seq_list]
    cum_start_idx = [0] + np.cumsum(_len).tolist()
    seq_start_end = [[start, end] for start, end in zip(cum_start_idx, cum_start_idx[1:])]

    obs_traj = torch.cat(obs_seq_list, dim=1)
    pred_traj = torch.cat(pred_seq_list, dim=1)
    obs_traj_rel = torch.cat(obs_seq_rel_list, dim=1)
    pred_traj_rel = torch.cat(pred_seq_rel_list, dim=1)
    loss_mask = torch.cat(loss_mask_list, dim=0)
    seq_start_end = torch.tensor(seq_start_end, device=obs_traj.device, dtype=torch.long)
    out = [obs_traj, pred_traj, obs_traj_rel, pred_traj_rel, metadata_list, loss_mask, seq_start_end]

    return tuple(out)

def data_loader_trajnetpp(args, device, path, load_pred=False):
    """
    provide the data loader that can be used to iterate through the dataset with Trajnet++ configuration
    :param args: arguments containing several dataset-related options
    :param device: torch.device to map the tensors to (like 'cpu', 'cuda')
    :param path: the path to the directory or file containing the data
    :param load_pred: if meant to load predictions from file instead of actual data
    :return: the dataset, of type TrajnetppDataset, and the data loader, of type torch.utils.data.DataLoader
    """
    if load_pred:
        dataset = TrajnetppPredictionsDataset(
            path, device,
            pred_len=args.pred_len,
            consider_partial_trajectories=not args.no_partial_trajectories,
            num_samples=args.num_samples
        )
    else:
        dataset = TrajnetppDataset(
            path, device,
            obs_len=args.obs_len,
            pred_len=args.pred_len,
            consider_partial_trajectories=not args.no_partial_trajectories,
            normalize_scene=args.normalize_scene,
            augment_rotation=args.augment_rotation,
            filter_tags=args.filter_tags,
            filter_sub_tags=args.filter_sub_tags)
    loader = DataLoader(
        dataset,
        batch_size=args.batch_size,
        shuffle=not args.do_not_shuffle if hasattr(args, 'do_not_shuffle') else False,
        collate_fn=seq_collate)
    return dataset, loader

max_stars_repo_name: pedro-mgb/pedestrian-arc-lstm-smf

max_issues_repo_name: pedro-mgb/pedestrian-arc-lstm-smf

max_forks_repo_name: pedro-mgb/pedestrian-arc-lstm-smf


text:
# -*- coding: utf-8 -*-

"""
General description
-------------------
As the csv-reader was removed with version 0.2 this example shows how to create
an excel-reader. The example is equivalent to the old csv-reader example.
Following the example one can customise the excel reader to ones own needs.

The pandas package supports the '.xls' and the '.xlsx' format but one can
create read and adept the files with open source software such as libreoffice,
openoffice, gnumeric,...

Data
----
scenario.xlsx

Installation requirements
-------------------------
This example requires the latest version of oemof (v0.3.x) and others.
Install by:

    pip install 'oemof.solph>=0.4,<0.5'
    pip3 install xlrd
    pip3 install matplotlib
    pip3 install networkx

If you want to plot the energy system's graph, you have to install pygraphviz
using:

    pip3 install pygraphviz

For pygraphviz under Windows, some hints are available in the oemof Wiki:
https://github.com/oemof/oemof/wiki/Windows---general

5.1.2018 - uwe.krien@rl-institut.de
7.5.2018 - jonathan.amme@rl-institut.de
"""

__copyright__ = "oemof developer group"
__license__ = "GPLv3"

import os
import logging
import pandas as pd

from oemof.tools import logger
from oemof import solph

from oemof.network.graph import create_nx_graph
from matplotlib import pyplot as plt
import networkx as nx

def nodes_from_excel(filename):
    """Read node data from Excel sheet

    Parameters
    ----------
    filename : :obj:`str`
        Path to excel file

    Returns
    -------
    :obj:`dict`
        Imported nodes data
    """

    # does Excel file exist?
    if not filename or not os.path.isfile(filename):
        raise FileNotFoundError(
            "Excel data file {} not found.".format(filename)
        )

    xls = pd.ExcelFile(filename)

    nodes_data = {
        "buses": xls.parse("buses"),
        "commodity_sources": xls.parse("commodity_sources"),
        "transformers": xls.parse("transformers"),
        "renewables": xls.parse("renewables"),
        "demand": xls.parse("demand"),
        "storages": xls.parse("storages"),
        "powerlines": xls.parse("powerlines"),
        "timeseries": xls.parse("time_series"),
    }

    # set datetime index
    nodes_data["timeseries"].set_index("timestamp", inplace=True)
    nodes_data["timeseries"].index = pd.to_datetime(
        nodes_data["timeseries"].index
    )

    print("Data from Excel file {} imported.".format(filename))

    return nodes_data

def create_nodes(nd=None):
    """Create nodes (oemof objects) from node dict

    Parameters
    ----------
    nd : :obj:`dict`
        Nodes data

    Returns
    -------
    nodes : `obj`:dict of :class:`nodes <oemof.network.Node>`
    """

    if not nd:
        raise ValueError("No nodes data provided.")

    nodes = []

    # Create Bus objects from buses table
    busd = {}

    for i, b in nd["buses"].iterrows():
        if b["active"]:
            bus = solph.Bus(label=b["label"])
            nodes.append(bus)

            busd[b["label"]] = bus
            if b["excess"]:
                nodes.append(
                    solph.Sink(
                        label=b["label"] + "_excess",
                        inputs={
                            busd[b["label"]]: solph.Flow(
                                variable_costs=b["excess costs"]
                            )
                        },
                    )
                )
            if b["shortage"]:
                nodes.append(
                    solph.Source(
                        label=b["label"] + "_shortage",
                        outputs={
                            busd[b["label"]]: solph.Flow(
                                variable_costs=b["shortage costs"]
                            )
                        },
                    )
                )

    # Create Source objects from table 'commodity sources'
    for i, cs in nd["commodity_sources"].iterrows():
        if cs["active"]:
            nodes.append(
                solph.Source(
                    label=cs["label"],
                    outputs={
                        busd[cs["to"]]: solph.Flow(
                            variable_costs=cs["variable costs"]
                        )
                    },
                )
            )

    # Create Source objects with fixed time series from 'renewables' table
    for i, re in nd["renewables"].iterrows():
        if re["active"]:
            # set static outflow values
            outflow_args = {"nominal_value": re["capacity"]}
            # get time series for node and parameter
            for col in nd["timeseries"].columns.values:
                if col.split(".")[0] == re["label"]:
                    outflow_args[col.split(".")[1]] = nd["timeseries"][col]

            # create
            nodes.append(
                solph.Source(
                    label=re["label"],
                    outputs={busd[re["to"]]: solph.Flow(**outflow_args)},
                )
            )

    # Create Sink objects with fixed time series from 'demand' table
    for i, de in nd["demand"].iterrows():
        if de["active"]:
            # set static inflow values
            inflow_args = {"nominal_value": de["nominal value"]}
            # get time series for node and parameter
            for col in nd["timeseries"].columns.values:
                if col.split(".")[0] == de["label"]:
                    inflow_args[col.split(".")[1]] = nd["timeseries"][col]

            # create
            nodes.append(
                solph.Sink(
                    label=de["label"],
                    inputs={busd[de["from"]]: solph.Flow(**inflow_args)},
                )
            )

    # Create Transformer objects from 'transformers' table
    for i, t in nd["transformers"].iterrows():
        if t["active"]:
            # set static inflow values
            inflow_args = {"variable_costs": t["variable input costs"]}
            # get time series for inflow of transformer
            for col in nd["timeseries"].columns.values:
                if col.split(".")[0] == t["label"]:
                    inflow_args[col.split(".")[1]] = nd["timeseries"][col]
            # create
            nodes.append(
                solph.Transformer(
                    label=t["label"],
                    inputs={busd[t["from"]]: solph.Flow(**inflow_args)},
                    outputs={
                        busd[t["to"]]: solph.Flow(nominal_value=t["capacity"])
                    },
                    conversion_factors={busd[t["to"]]: t["efficiency"]},
                )
            )

    for i, s in nd["storages"].iterrows():
        if s["active"]:
            nodes.append(
                solph.components.GenericStorage(
                    label=s["label"],
                    inputs={
                        busd[s["bus"]]: solph.Flow(
                            nominal_value=s["capacity inflow"],
                            variable_costs=s["variable input costs"],
                        )
                    },
                    outputs={
                        busd[s["bus"]]: solph.Flow(
                            nominal_value=s["capacity outflow"],
                            variable_costs=s["variable output costs"],
                        )
                    },
                    nominal_storage_capacity=s["nominal capacity"],
                    loss_rate=s["capacity loss"],
                    initial_storage_level=s["initial capacity"],
                    max_storage_level=s["capacity max"],
                    min_storage_level=s["capacity min"],
                    inflow_conversion_factor=s["efficiency inflow"],
                    outflow_conversion_factor=s["efficiency outflow"],
                )
            )

    for i, p in nd["powerlines"].iterrows():
        if p["active"]:
            bus1 = busd[p["bus_1"]]
            bus2 = busd[p["bus_2"]]
            nodes.append(
                solph.custom.Link(
                    label="powerline" + "_" + p["bus_1"] + "_" + p["bus_2"],
                    inputs={bus1: solph.Flow(), bus2: solph.Flow()},
                    outputs={
                        bus1: solph.Flow(nominal_value=p["capacity"]),
                        bus2: solph.Flow(nominal_value=p["capacity"]),
                    },
                    conversion_factors={
                        (bus1, bus2): p["efficiency"],
                        (bus2, bus1): p["efficiency"],
                    },
                )
            )

    return nodes

def draw_graph(
    grph,
    edge_labels=True,
    node_color="#AFAFAF",
    edge_color="#CFCFCF",
    plot=True,
    node_size=2000,
    with_labels=True,
    arrows=True,
    layout="neato",
):
    """
    Parameters
    ----------
    grph : networkxGraph
        A graph to draw.
    edge_labels : boolean
        Use nominal values of flow as edge label
    node_color : dict or string
        Hex color code oder matplotlib color for each node. If string, all
        colors are the same.

    edge_color : string
        Hex color code oder matplotlib color for edge color.

    plot : boolean
        Show matplotlib plot.

    node_size : integer
        Size of nodes.

    with_labels : boolean
        Draw node labels.

    arrows : boolean
        Draw arrows on directed edges. Works only if an optimization_model has
        been passed.
    layout : string
        networkx graph layout, one of: neato, dot, twopi, circo, fdp, sfdp.
    """
    if type(node_color) is dict:
        node_color = [node_color.get(g, "#AFAFAF") for g in grph.nodes()]

    # set drawing options
    options = {
        "prog": "dot",
        "with_labels": with_labels,
        "node_color": node_color,
        "edge_color": edge_color,
        "node_size": node_size,
        "arrows": arrows,
    }

    # try to use pygraphviz for graph layout
    try:
        import pygraphviz

        pos = nx.drawing.nx_agraph.graphviz_layout(grph, prog=layout)
    except ImportError:
        logging.error("Module pygraphviz not found, I won't plot the graph.")
        return

    # draw graph
    nx.draw(grph, pos=pos, **options)

    # add edge labels for all edges
    if edge_labels is True and plt:
        labels = nx.get_edge_attributes(grph, "weight")
        nx.draw_networkx_edge_labels(grph, pos=pos, edge_labels=labels)

    # show output
    if plot is True:
        plt.show()

logger.define_logging()
datetime_index = pd.date_range(
    "2016-01-01 00:00:00", "2016-01-01 23:00:00", freq="60min"
)

# model creation and solving
logging.info("Starting optimization")

# initialisation of the energy system
esys = solph.EnergySystem(timeindex=datetime_index)

# read node data from Excel sheet
excel_nodes = nodes_from_excel(os.path.join(os.getcwd(), "scenario.xlsx",))

# create nodes from Excel sheet data
my_nodes = create_nodes(nd=excel_nodes)

# add nodes and flows to energy system
esys.add(*my_nodes)

print("*********************************************************")
print("The following objects have been created from excel sheet:")
for n in esys.nodes:
    oobj = str(type(n)).replace("<class 'oemof.solph.", "").replace("'>", "")
    print(oobj + ":", n.label)
print("*********************************************************")

# creation of a least cost model from the energy system
om = solph.Model(esys)
om.receive_duals()

# solving the linear problem using the given solver
om.solve(solver="cbc")

# create graph of esys
# You can use argument filename='/home/somebody/my_graph.graphml'
# to dump your graph to disc. You can open it using e.g. yEd or gephi
graph = create_nx_graph(esys)

# plot esys graph
draw_graph(
    grph=graph,
    plot=True,
    layout="neato",
    node_size=1000,
    node_color={"R1_bus_el": "#cd3333", "R2_bus_el": "#cd3333"},
)

# print and plot some results
results = solph.processing.results(om)

region2 = solph.views.node(results, "R2_bus_el")
region1 = solph.views.node(results, "R1_bus_el")

print(region2["sequences"].sum())
print(region1["sequences"].sum())

fig, ax = plt.subplots(figsize=(10, 5))
region1["sequences"].plot(ax=ax)
ax.legend(
    loc="upper center", prop={"size": 8}, bbox_to_anchor=(0.5, 1.4), ncol=3
)
fig.subplots_adjust(top=0.7)
plt.show()
logging.info("Done!")

max_stars_repo_name: ekatef/oemof-examples

max_issues_repo_name: ekatef/oemof-examples

max_forks_repo_name: ekatef/oemof-examples

reason: import networkx


text:
from __future__ import absolute_import, print_function, division
import sys, timeit
import numpy
import scikits.image.opencv

try:
    img_shape =  int(sys.argv[1]), int(sys.argv[2])
    ker_shape =  int(sys.argv[3]), int(sys.argv[4])
    dtype = sys.argv[5]
except:
    print("Usage: %s <img rows> <img cols> <ker rows> <ker cols> <dtype> [nb_call]" % sys.argv[0], file=sys.stderr)
    sys.exit(-1)

nb_call = 1
if len(sys.argv)>6:
    nb_call=int(sys.argv[6])

T = timeit.Timer("f()","""
import scikits.image.opencv, sys, numpy
img_shape =  int(sys.argv[1]), int(sys.argv[2])
ker_shape =  int(sys.argv[3]), int(sys.argv[4])
dtype = sys.argv[5]

img = numpy.ones(img_shape, dtype=dtype)
ker = numpy.ones(ker_shape, dtype=dtype)

def f():
    scikits.image.opencv.cvFilter2D(img, ker)
""")
time = T.repeat(repeat=3, number=nb_call)
print(min(time), "opencv")

max_stars_repo_name: intel/Theano-dev

max_issues_repo_name: intel/Theano-dev

max_forks_repo_name: intel/Theano-dev


text:
from milk.supervised.multi import ecoc_learner
from milk.supervised.classifier import ctransforms
from milk.supervised import svm
import milk.tests.fast_classifier
import milk.supervised.multi
from milksets.yeast import load
import numpy as np

def test_ecoc_learner():
    base = milk.tests.fast_classifier.fast_classifier()
    learner = milk.supervised.multi.ecoc_learner(base)
    features, labels = load()
    nlabels = len(set(labels))
    model = learner.train(features[::2],labels[::2])

    testl = np.array(model.apply_many(features[1::2]))
    assert np.mean(testl == labels[1::2]) > 1./nlabels
    assert testl.min() >= 0
    assert testl.max() < nlabels

# This failed at one point:
    learner = ecoc_learner(svm.svm_to_binary(svm.svm_raw(kernel=svm.dot_kernel(), C=1.)))
    model = learner.train(features[:200], labels[:200])
    assert (model is not None)

def test_ecoc_probability():
    features,labels = load()
    features = features[labels < 5]
    labels = labels[labels < 5]
    raw = svm.svm_raw(kernel=svm.dot_kernel(), C=1.)
    base = ctransforms(raw, svm.svm_sigmoidal_correction())
    learner = ecoc_learner(base, probability=True)
    model = learner.train(features[::2], labels[::2])
    results = list(map(model.apply, features[1::2]))
    results = np.array(results)
    assert results.shape[1] == len(set(labels))
    assert np.mean(results.argmax(1) == labels[1::2]) > .5

max_stars_repo_name: luispedro/milk


text:
#!/usr/bin/env python
# encoding: utf-8
"""
Identify which PHAT brick+field each AST field corresponds to.

2015-07-08 - Created by Jonathan Sick
"""

import numpy as np
from sklearn.cluster import KMeans
# from astroML.stats import binned_statistic

# import matplotlib as mpl
from matplotlib.figure import Figure
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
import matplotlib.gridspec as gridspec
from matplotlib.patches import Polygon
from matplotlib.path import Path

from astropy.table import Table
from astropy import units as u
from astropy.coordinates import SkyCoord

from m31hst.phatast import load_phat_ast_table

from androcmd.phatpatchfit.galexmap import (load_galex_map, setup_galex_axes,
                                            plot_patch_footprints)
from androcmd.phatpatchfit.pipeline import load_field_patches

def main():
    fields = load_field_patches()
    ast_centers = load_ast_centers()
    matches = match_fields(fields, ast_centers)
    print matches
    plot_ast_fields(fields, matches, ast_centers=ast_centers)

    print_ast_table(matches)

def load_ast_centers():
    t = load_phat_ast_table()
    km = KMeans(n_clusters=6)
    xy = np.vstack((t['ra'], t['dec'])).T
    km.fit(xy)
    centers = km.cluster_centers_

    srt = np.argsort(centers[:, 1])
    return centers[srt, :]

def plot_ast_fields(fields, matches, ast_centers=None):
    fig = Figure(figsize=(3.5, 3.5), frameon=False)
    canvas = FigureCanvas(fig)
    gs = gridspec.GridSpec(1, 1,
                           left=0.15, right=0.95, bottom=0.15, top=0.95,
                           wspace=None, hspace=None,
                           width_ratios=None, height_ratios=None)
    basemap = load_galex_map()
    ax = setup_galex_axes(fig, gs[0], basemap)
    plot_patch_footprints(ax, alpha=0.8, edgecolor='dodgerblue')
    for n, m in matches.iteritems():
        footprint = np.array(m['poly'])
        patch = Polygon(footprint, closed=True,
                        transform=ax.get_transform('world'),
                        facecolor='y', alpha=1,
                        edgecolor='k', lw=0.5, zorder=10)
        ax.add_patch(patch)
        x = footprint[:, 0].mean()
        y = footprint[:, 1].mean()
        ax.annotate('{0:d}'.format(n), xy=(x, y),
                    xycoords=ax.get_transform('world'),
                    xytext=(3, -3), textcoords="offset points",
                    size=8,
                    bbox=dict(boxstyle="round",
                              fc=(1., 1., 1., 0.8),
                              edgecolor='None'))
    if ast_centers is not None:
        ax.scatter(ast_centers[:, 0], ast_centers[:, 1],
                   marker='*', c='y',
                   transform=ax.get_transform('world'))
    gs.tight_layout(fig, pad=1.08, h_pad=None, w_pad=None, rect=None)
    ax.coords[0].ticklabels.set_size(11)
    ax.coords[1].ticklabels.set_size(11)
    canvas.print_figure("phat_ast_fields.pdf", format="pdf")

def match_fields(fields, centers):
    matches = {}
    for i in xrange(centers.shape[0]):
        for field in fields:
            # poly = Polygon(field['poly'])
            poly = Path(field['poly'])
            if poly.contains_point((centers[i, 0], centers[i, 1])):
                match = {'brick': field['brick'],
                         'field': field['field'],
                         'ra': centers[i, 0],
                         'dec': centers[i, 1],
                         'poly': field['poly']}
                matches[i + 1] = match
    return matches

def print_ast_table(matches):
    rows = []
    for i in range(1, 7):
        m = matches[i]
        coord = SkyCoord(ra=m['ra'] * u.degree, dec=m['dec'] * u.degree)
        rows.append((i,
                     m['brick'],
                     m['field'],
                     coord.ra.to_string(unit=u.hour, sep=":"),
                     coord.dec.to_string(unit=u.degree, sep=":")))
    t = Table(rows=rows, names=('Number', 'Brick', 'Field', 'R.A.',
                                'Dec.'))
    t.write('phat_ast_fields.tex',
            col_align='lllll',
            caption=r'\Phat\ artificial star test fields used '
                    r'for mock testing.',
            latexdict={'preamble': r'\begin{center}',
                       'tablefoot': r'\label{tab:phat_ast_fields}'
                                    r'\end{center}'},
            format='ascii.latex')

if __name__ == '__main__':
    main()

max_stars_repo_name: jonathansick/androcmd

max_issues_repo_name: jonathansick/androcmd

max_forks_repo_name: jonathansick/androcmd

reason: import numpy,from astropy