text:
from typing import Iterator, Tuple

from depmap_analysis.tests import *
from networkx import DiGraph

from indra_network_search.data_models import Node, SubgraphRestQuery, SubgraphResults
from indra_network_search.pathfinding import *
from indra_network_search.query import SubgraphQuery
from indra_network_search.result_handler import (
    DB_URL_EDGE,
    DB_URL_HASH,
    SubgraphResultManager,
)
from indra_network_search.tests import edge_data
from indra_network_search.tests.util import expanded_unsigned_graph

def test_shared_parents():
    ns1 = "HGNC"
    ns2 = "HGNC"
    id1 = "1100"
    id2 = "1101"
    res = shared_parents(source_ns=ns1, target_ns=ns2, source_id=id1, target_id=id2)
    short_res = [(ns, _id) for _, ns, _id, _ in res]
    assert ("FPLX", "BRCA") in short_res
    assert ("FPLX", "FANC") in short_res

def test_shared_targets():
    idg = get_dg()
    source = "X1"
    target = "X2"
    shared_target = "Z1"

    res = shared_interactors(graph=idg, source=source, target=target, regulators=False)
    res_list = [t for t in res]
    assert ([source, shared_target], [target, shared_target]) in res_list

def test_shared_targets_w_big_graph():
    source_edges = [("BRCA1", n) for n in ["AR", "testosterone", "NR2C2", "MBD2", "PATZ1"]]
    target_edges = [("HDAC3", n) for n in ["AR", "testosterone", "NR2C2", "MBD2", "PATZ1"]]
    edges = shared_interactors(graph=expanded_unsigned_graph, source="BRCA1", target="HDAC3", regulators=False)
    edges_list = list(edges)
    for ix, (se, te) in enumerate(edges_list):
        assert tuple(se) == source_edges[ix]
        assert tuple(te) == target_edges[ix]

def test_shared_regulators_w_big_graph():
    source_edges = [(n, "CHEK1") for n in ["AR", "testosterone", "NR2C2", "MBD2", "PATZ1"]]
    target_edges = [(n, "H2AZ1") for n in ["AR", "testosterone", "NR2C2", "MBD2", "PATZ1"]]
    assert edge_data[("AR", "CHEK1")]["belief"] == 0.999999
    assert edge_data[("testosterone", "CHEK1")]["belief"] == 0.99999
    assert edge_data[("NR2C2", "CHEK1")]["belief"] == 0.9999
    assert edge_data[("MBD2", "CHEK1")]["belief"] == 0.999
    assert edge_data[("PATZ1", "CHEK1")]["belief"] == 0.99
    edges = shared_interactors(graph=expanded_unsigned_graph, source="CHEK1", target="H2AZ1", regulators=True)
    edges_list = list(edges)
    for ix, (se, te) in enumerate(edges_list):
        assert tuple(se) == source_edges[ix]
        assert tuple(te) == target_edges[ix]

def test_shared_regulators():
    idg = get_dg()
    source = "X1"
    target = "X2"
    shared_regulator = "Z2"

    res = shared_interactors(graph=idg, source=source, target=target, regulators=True)
    res_list = [t for t in res]
    assert ([shared_regulator, source], [shared_regulator, target]) in res_list

mock_edge_dict = {
    "statements": [
        {
            "stmt_hash": 31955807459270625,
            "stmt_type": "Inhibition",
            "evidence_count": 1,
            "belief": 0.65,
            "source_counts": {"reach": 1},
            "english": "AR inhibits testosterone.",
            "weight": 0.4307829160924542,
            "position": None,
            "curated": False,
            "residue": None,
            "initial_sign": 1,
        }
    ],
    "belief": 0.9999998555477862469,
    "weight": 1.4445222418630995515e-07,
}

def _setup_query_graph() -> DiGraph:
    # Sets up shared parameters for shared_targets and shared_regulators

    # Add 4 nodes: two nodes that have two other nodes as shared up- and
    # downstream
    g = DiGraph()
    n1 = "n1"
    n2 = "n2"
    nsr = "nsr"
    nst = "nst"
    ns1, ns2, ns_sr, ns_st = ("HGNC",) * 4
    id1, id2, id_sr, id_st = "1100", "1101", "1102", "1103"

    g.add_node(n1, ns=ns1, id=id1)
    g.add_node(n2, ns=ns2, id=id2)
    g.add_node(nst, ns=ns_st, id=id_st)
    g.add_node(nsr, ns=ns_sr, id=id_sr)
    g.graph["node_by_ns_id"] = {
        (ns, _id): n for n, ns, _id in zip([n1, n2, nsr, nst], [ns1, ns2, ns_sr, ns_st], [id1, id2, id_sr, id_st])
    }
    sd12 = mock_edge_dict

    g.add_edge(n1, n2, **sd12)
    g.add_edge(nsr, n1, **sd12)
    g.add_edge(nsr, n2, **sd12)
    g.add_edge(n1, nst, **sd12)
    g.add_edge(n2, nst, **sd12)

    # Return graph and the instances that are shared between the two
    return g

def test_subgraph():
    g = _setup_query_graph()
    input_node = Node(name="n1", namespace="HGNC", identifier="1100")
    subgrap_rest_query = SubgraphRestQuery(nodes=[input_node])
    subgraph_query = SubgraphQuery(query=subgrap_rest_query)
    options = subgraph_query.run_options(graph=g)
    edge_iter: Iterator[Tuple[str, str]] = get_subgraph_edges(graph=g, **options)

    # Should have zero results
    edges = list(edge_iter)
    assert len(edges) == 0

    # Get result manager
    res_mngr = SubgraphResultManager(path_generator=iter(edges), graph=g, **subgraph_query.result_options())
    results: SubgraphResults = res_mngr.get_results()

    # Should have zero results
    assert len(results.edges) == 0

    # Test with n1, n2
    input_nodes = [
        Node(name="n1", namespace="HGNC", identifier="1100"),
        Node(name="n2", namespace="HGNC", identifier="1101"),
    ]
    subgrap_rest_query = SubgraphRestQuery(nodes=input_nodes)
    subgraph_query = SubgraphQuery(query=subgrap_rest_query)
    options = subgraph_query.run_options(graph=g)
    edge_iter = get_subgraph_edges(graph=g, **options)

    # Should have one result total: B1 -> B2
    edges = list(edge_iter)
    assert len(edges) == 1
    assert set(edges) == {("n1", "n2")}

    # Get result manager
    res_mngr = SubgraphResultManager(path_generator=iter(edges), graph=g, **subgraph_query.result_options())
    results: SubgraphResults = res_mngr.get_results()

    # Should have one result total: B1 -> B2
    assert len(results.edges) == 1

    str_edges = {tuple([n.name for n in e.edge]) for e in results.edges}
    assert str_edges == {("n1", "n2")}
    assert results.edges[0].weight == mock_edge_dict["weight"]
    assert results.edges[0].belief == mock_edge_dict["belief"]
    assert results.edges[0].db_url_edge == DB_URL_EDGE.format(
        subj_id="1100", subj_ns="HGNC", obj_id="1101", obj_ns="HGNC", ev_limit=10
    ), results.edges[0].db_url_edge
    assert (
        list(results.edges[0].stmts.values())[0].db_url_hash
        == DB_URL_HASH.format(stmt_hash=mock_edge_dict["statements"][0]["stmt_hash"]) + "&ev_limit=10"
    ), list(results.edges[0].stmts.values())[0].db_url_hash
    assert set(list(results.edges[0].stmts.values())[0].dict().keys()).difference({"db_url_hash"}) == set(
        mock_edge_dict["statements"][0].keys()
    )
    assert all(
        mock_edge_dict["statements"][0][k] == v
        for k, v in list(results.edges[0].stmts.values())[0].dict().items()
        if k != "db_url_hash"
    )

    # Check that input nodes were mapped properly
    assert results.input_nodes[0].name == input_node.name
    assert results.input_nodes[0].namespace == input_node.namespace
    assert results.input_nodes[0].identifier == input_node.identifier
    assert len(results.not_in_graph) == 0
    assert results.available_nodes[0].name == "n1"
    assert results.available_nodes[0].namespace == input_node.namespace
    assert results.available_nodes[0].identifier == input_node.identifier

    # Check correct name, missing/bad ns & id
    input_nodes = [
        Node(name="n1", namespace="bad ns", identifier="bad id"),
        Node(name="n2", namespace="HGNC", identifier="1101"),
        Node(name="nst", namespace="HGNC", identifier="1103"),
        Node(name="nsr", namespace="HGNC", identifier="1102"),
    ]
    subgrap_rest_query = SubgraphRestQuery(nodes=input_nodes)
    subgraph_query = SubgraphQuery(query=subgrap_rest_query)
    options = subgraph_query.run_options(graph=g)
    edge_iter = get_subgraph_edges(graph=g, **options)

    # Should have five results total
    edges = list(edge_iter)
    assert len(edges) == 5
    assert set(edges) == {("nsr", "n1"), ("nsr", "n2"), ("n1", "n2"), ("n1", "nst"), ("n2", "nst")}

    # Get result manager
    res_mngr = SubgraphResultManager(path_generator=iter(edges), graph=g, **subgraph_query.result_options())
    results: SubgraphResults = res_mngr.get_results()

    # Should have five results total
    assert len(results.edges) == 5

    str_edges = {tuple([n.name for n in e.edge]) for e in results.edges}
    assert str_edges == {("nsr", "n1"), ("nsr", "n2"), ("n1", "n2"), ("n1", "nst"), ("n2", "nst")}
    assert results.edges[0].weight == mock_edge_dict["weight"]
    assert results.edges[0].belief == mock_edge_dict["belief"]
    assert (
        list(results.edges[0].stmts.values())[0].db_url_hash
        == DB_URL_HASH.format(stmt_hash=mock_edge_dict["statements"][0]["stmt_hash"]) + "&ev_limit=10"
    ), list(results.edges[0].stmts.values())[0].db_url_hash
    assert set(list(results.edges[0].stmts.values())[0].dict().keys()).difference({"db_url_hash"}) == set(
        mock_edge_dict["statements"][0].keys()
    )
    assert all(
        mock_edge_dict["statements"][0][k] == v
        for k, v in list(results.edges[0].stmts.values())[0].dict().items()
        if k != "db_url_hash"
    )

    # Check that input nodes were mapped properly
    assert results.input_nodes[0].name == input_nodes[0].name
    assert results.input_nodes[0].namespace == input_nodes[0].namespace
    assert results.input_nodes[0].identifier == input_nodes[0].identifier
    assert len(results.not_in_graph) == 0
    assert results.available_nodes[0].name == input_nodes[0].name
    assert results.available_nodes[0].namespace == "HGNC"
    assert results.available_nodes[0].identifier == "1100"

    # Check node not in graph
    input_node = Node(name="not in graph", namespace="bad ns ", identifier="bad id")
    subgrap_rest_query = SubgraphRestQuery(nodes=[input_node])
    subgraph_query = SubgraphQuery(query=subgrap_rest_query)
    options = subgraph_query.run_options(graph=g)
    edge_iter_list = list(get_subgraph_edges(graph=g, **options))

    assert len(edge_iter_list) == 0

    # Get result manager
    res_mngr = SubgraphResultManager(path_generator=edge_iter_list, graph=g, **subgraph_query.result_options())
    results: SubgraphResults = res_mngr.get_results()
    assert len(results.edges) == 0
    assert len(results.not_in_graph) == 1
    assert results.input_nodes[0].name == input_node.name
    assert results.input_nodes[0].namespace == input_node.namespace
    assert results.input_nodes[0].identifier == input_node.identifier
    assert results.not_in_graph[0].name == input_node.name
    assert results.not_in_graph[0].namespace == input_node.namespace
    assert results.not_in_graph[0].identifier == input_node.identifier

def test_multi_interactors():
    # BRCA1 and HDAC3 should have AR, testosterone, NR2C2, MBD2, PATZ1 in
    # common

    # Test downstream from BRCA1, HDAC3
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True)
    res_set = set(res_iter)
    assert res_set == {"AR", "testosterone", "NR2C2", "MBD2", "PATZ1"}

    # Test upstream from CHEK1, H2AZ1
    input_nodes = ["CHEK1", "H2AZ1"]
    res_iter = direct_multi_interactors(graph=expanded_unsigned_graph, nodes=input_nodes, downstream=False)
    res_set = set(res_iter)
    assert res_set == {"AR", "testosterone", "NR2C2", "MBD2", "PATZ1"}

def test_multi_interactors_allowed_ns():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, allowed_ns=["chebi"]
    )
    res_set = set(res_iter)
    assert res_set == {"testosterone"}

def test_multi_interactors_stmt_types():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, stmt_types=["activation"]
    )
    res_set = set(res_iter)
    assert res_set == {"AR"}

def test_multi_interactors_source_filter():
    #   - source filter
    input_nodes = ["CHEK1", "H2AZ1"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=False, source_filter=["pc"]
    )
    res_set = set(res_iter)
    assert res_set == {"AR", "testosterone", "NR2C2"}

def test_multi_interactors_max_results():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, max_results=3
    )
    res_set = list(res_iter)
    assert res_set == ["AR", "testosterone", "NR2C2"]

def test_multi_interactors_hash_blacklist():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph,
        nodes=input_nodes,
        downstream=True,
        hash_blacklist={
            5603789525715921,
        },  # blacklist BRCA1->AR
    )
    res_set = set(res_iter)
    assert res_set == {"testosterone", "NR2C2", "MBD2", "PATZ1"}

def test_multi_interactors_node_blacklist():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, node_blacklist=["AR"]
    )
    res_set = set(res_iter)
    assert res_set == {"testosterone", "NR2C2", "MBD2", "PATZ1"}

def test_multi_interactors_belief_cutoff():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, belief_cutoff=0.9995
    )
    res_set = set(res_iter)
    assert res_set == {"AR", "testosterone", "NR2C2"}

def test_multi_interactors_curated_db_only():
    input_nodes = ["BRCA1", "HDAC3"]
    res_iter = direct_multi_interactors(
        graph=expanded_unsigned_graph, nodes=input_nodes, downstream=True, curated_db_only=True
    )
    res_set = set(res_iter)
    assert res_set == {"AR", "testosterone", "NR2C2"}

reason: from networkx


text:
import logging
import numpy as np
import unittest

import batchglm.api as glm
from batchglm.unit_test.test_graph_glm_all import _TestGraphGlmAll

glm.setup_logging(verbosity="WARNING", stream="STDOUT")
logger = logging.getLogger(__name__)

class _TestAccuracyXtremeAll(_TestGraphGlmAll):
    """
    Test whether numerical extremes throw error in initialisation or during first training steps.
    """

    def _modify_sim(self, idx, val):
        if self.noise_model is None:
            raise ValueError("noise_model is None")
        else:
            if self.noise_model == "nb":
                from batchglm.api.models.numpy.glm_nb import Estimator, InputDataGLM
            elif self.noise_model == "norm":
                from batchglm.api.models import Estimator, InputDataGLM
            elif self.noise_model == "beta":
                from batchglm.api.models.numpy.glm_beta import Estimator, InputDataGLM
            else:
                raise ValueError("noise_model not recognized")

        self.simulate1()
        x = self.sim1.x
        x[:, idx] = val
        input_data = InputDataGLM(
            data=x,
            design_loc=self.sim1.input_data.design_loc,
            design_scale=self.sim1.input_data.design_scale
        )
        self.sim1.input_data = input_data

    def _test_low_values_a_and_b(self):
        self._modify_sim(idx=0, val=0.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
        )

    def _test_low_values_a_only(self):
        self._modify_sim(idx=0, val=0.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
        )

    def _test_low_values_b_only(self):
        self._modify_sim(idx=0, val=0.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
        )

    def _test_zero_variance_a_and_b(self):
        self._modify_sim(idx=0, val=5.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
        )

    def _test_zero_variance_a_only(self):
        self._modify_sim(idx=0, val=5.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
    )

    def _test_zero_variance_b_only(self):
        self._modify_sim(idx=0, val=5.)
        return self.basic_test(
            batched=False,
            train_loc=True,
            train_scale=True,
            sparse=False
        )

    def _test_all(self):
        self._test_low_values_a_and_b()
        self._test_low_values_a_only()
        self._test_low_values_b_only()
        self._test_zero_variance_a_and_b()
        self._test_zero_variance_a_only()
        self._test_zero_variance_b_only()

class TestAccuracyXtremeNb(
    _TestAccuracyXtremeAll,
    unittest.TestCase
):
    """
    Test whether optimizers yield exact results for negative binomial distributed data.
    """

    def test_nb(self):
        logging.getLogger("tensorflow").setLevel(logging.ERROR)
        logging.getLogger("batchglm").setLevel(logging.WARNING)
        logger.error("TestAccuracyXtremeNb.test_nb()")

        np.random.seed(1)
        self.noise_model = "nb"
        self._test_all()

class TestAccuracyXtremeNorm(
    _TestAccuracyXtremeAll,
    unittest.TestCase
):
    """
    Test whether optimizers yield exact results for normal distributed data.
    """

    def test_norm(self):
        logging.getLogger("tensorflow").setLevel(logging.ERROR)
        logging.getLogger("batchglm").setLevel(logging.WARNING)
        logger.error("TestAccuracyXtremeNorm.test_norm()")

        np.random.seed(1)
        self.noise_model = "norm"
        self._test_all()

class TestAccuracyXtremeBeta(
    _TestAccuracyXtremeAll,
    unittest.TestCase
):
    """
    Test whether optimizers yield exact results for beta distributed data.
    """

    def test_beta(self):
        logging.getLogger("tensorflow").setLevel(logging.ERROR)
        logging.getLogger("batchglm").setLevel(logging.WARNING)
        logger.error("TestAccuracyXtremeBeta.test_beta()")

        np.random.seed(1)
        self.noise_model = "beta"
        self._test_all()

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

ext: py

max_stars_repo_name: le-ander/batchglm

max_issues_repo_name: le-ander/batchglm

max_forks_repo_name: le-ander/batchglm

reason: import numpy


text:
"""
Usage:
> python3 analysis.py [--help] [--latex] [--plot] [--tables] [--save-plot]

The following script analyzes benchmarks for Karger (and variants) algorithm.

The aim of the script is to produce in output some graphical representation of
the benchmarks, in particular for running times and solution error.

Options:
- [--help]: show this message
- [--tables]: show comparison tables
- [--latex]: export comparison tables in latex format (use with [--tables])
- [--plot]: plot comparisons
- [--save-plot]: save charts to a specific folder instead of visualizing them (use with [--plot])
"""

from functools import reduce
import sys
import os
from io import StringIO

import pandas as pd
import numpy as np
import glob
from itertools import chain
from tabulate import tabulate
from typing import List, Dict
import matplotlib.pyplot as plt
from types import SimpleNamespace
from tably import Tably
import seaborn as sns

# number of floating point decimals displayed in output
N_DECIMALS = 3

# number of floating point decimals displayed in output for values that represent percentages
N_DECIMALS_PERCENTAGE = 2

IS_HELP = '--help' in sys.argv
# if the script is given the argument '--latex', turn on LaTeX table generation
FOR_LATEX = '--latex' in sys.argv
# if the script is given the argument '--tables', turn on tables comparisons
IS_TABLE_ENABLED = '--tables' in sys.argv
# if the script is given the argument '--plot', turn on comparison plots (with options)
IS_PLOT_ENABLED = '--plot' in sys.argv
# if the script is given the argument '--save-plot', save plots to folder.
IS_SAVE_PLOT_ENABLED = '--save-plot' in sys.argv

PLOT_IMG_DIR = 'images'

KARGER = 'KargerMinCut'
KARGER_TOUT = 'KargerMinCutTimeout'
KARGER_STEIN = 'KargerSteinMinCut'

programs = [
    KARGER,
    KARGER_TOUT,
    KARGER_STEIN,
]

programs_use_meta = [
    True,
    True,
    False,
]

GROUND_TRUTH_DF = pd.DataFrame({
    # k : [ v, v, ... ]
})

def read_csvs_of_program(program: str) -> List[pd.DataFrame]:
    """
    Given the name of an algorithm program, collects the list of CVS benchmarks recorded
    for that particular program.
    :param program: name of the program which benchmarks should be retrieved
    :return: list of benchmark CSV files
    """

    csv_files = glob.glob(f'./{program}_*.csv')
    dataframes_per_program = []

    for csv_file in csv_files:
        dataframe = pd.read_csv(csv_file, sep=';', decimal='.', encoding='utf-8')
        dataframes_per_program.append(dataframe)

    return dataframes_per_program

def check_same_n_rows(dfs_flat: List[pd.DataFrame]):
    """
    Utility function
    :param dfs_flat:
    """

    dfs_size = map(lambda x: x.shape[0], dfs_flat)
    is_valid = len(set(dfs_size)) == 1
    if not is_valid:
        raise AssertionError('The CSV files have different number of rows')

def check_same_outputs(dfs_flat: List[pd.DataFrame]):
    """
    :param dfs_flat:
    :return:
    """
    # Nothing to do, solution can be approximated.
    pass

def check_not_empty(dfs:  Dict[str, List[pd.DataFrame]]):
    for name, dfs_list in dfs.items():
        if len(dfs_list) == 0:
            raise AssertionError(f'There are no data for {name}')

def check_output_not_null(dfs: Dict[str, List[pd.DataFrame]]):
    for name, dfs_list in dfs.items():
        for df in dfs_list:
            if np.any(df.isnull()):
                raise AssertionError(f'Output values cannot be null, check {name} benchmarks')

def check_validity(dfs: Dict[str, List[pd.DataFrame]]):
    """
    Utility functions that checks whether the imported benchmark CSVs are valid
    :param dfs_list: list of list of CSVs represented as DataFrames
    """
    dfs_list = dfs.values()
    dfs_flat: List[pd.DataFrame] = list(chain.from_iterable(dfs_list))

    check_not_empty(dfs)
    check_same_n_rows(dfs_flat)
    check_output_not_null(dfs)
    check_same_outputs(dfs_flat)

def get_row_index_with_max_edges(d: int) -> int:
    """
    Return the row index of any benchmark CSV file with given dimension.
    The number of nodes considered "inseresting" are:
    - 14
    - 16
    - 22
    - 52
    - 202
    - 1000

    :param d: graph dimension in term nodes
    :return: index of line with given graph dimension
    """

    # these values correspond to the (line - 2) of the CSV line
    obj = {
        14: 1,
        16: 11,
        22: 12,
        52: 0,
        202: 6,
        1000: 4,
    }

    return obj[d]

def get_min_at_row(dfs: List[pd.DataFrame], row: int, column: str = 'program_time') -> np.double:
    """
    Return the minimum value of a certain column among the given list of benchmark dataframes in a certain row
    :param dfs: list of benchmark dataframes
    :param row: index of the row to be selected
    :param column: column of the dataframe to be selected. Default: 'program_time'
    :return: minimum value of column at row
    """

    values_at_row_column = [df[column].loc[row] for df in dfs]
    min_index, min_value = min(enumerate(values_at_row_column), key=lambda x: x[1])
    return min_index, min_value

def get_median_at_row(dfs: List[pd.DataFrame], row: int, column: str = 'output') -> np.double:
    """
    Return the median value of a certain column among the given list of benchmark dataframes in a certain row
    :param dfs: list of benchmark dataframes
    :param row: index of the row to be selected
    :param column: column of the dataframe to be selected. Default: 'program_time'
    :return: median value of column at row
    """
    values_at_row_column = [df[column].loc[row] for df in dfs]
    return np.median(values_at_row_column)

def get_mean_at_row(dfs: List[pd.DataFrame], row: int, column: str) -> np.double:
    """
    Return the mean value of a certain column among the given list of benchmark dataframes in a certain row
    :param dfs: list of benchmark dataframes
    :param row: index of the row to be selected
    :param column: column of the dataframe to be selected. Default: 'program_time'
    :return: mean value of column at row
    """
    values_at_row_column = [df[column].loc[row] for df in dfs]
    return np.mean(values_at_row_column)

def create_list_of_ms_at_row(dfs_list: List[pd.DataFrame], columns: List[int]) -> List[List[np.double]]:
    """
    :param dfs_list: list of benchmark dataframes
    :param columns: list of columns to be displayed in a table, aka list of "interesting" number of nodes
    :return: 2-d list of minimum execution times. For each column in columns, the row which stores the maximum
    number of edges wrt to the fixed number of nodes (column) is selected.
    """

    return [
        [
            df['program_time'].loc[get_row_index_with_max_edges(n_nodes_as_str)]
            for n_nodes_as_str in columns
        ]
        for df in dfs_list
    ]

def compare_2_programs(dfs_dict: Dict[str, pd.DataFrame], program_1: str, program_2: str) -> pd.DataFrame:
    """
    Compare 2 programs, putting in evidence their difference in runtime execution and how much the 2nd program
    improves the execution time compared to the 1st program.
    :param dfs_dict: map of DataFrames minimized wrt to column 'program_time'
    :param program_1: name of the 1st program to compare
    :param program_2: name of the 2nd program to compare
    :return: new DataFrame containing the comparison information between the given two programs
    """

    # subset of dfs_dict containing only the list of dataframes related to program_1 and program_2
    dfs_filter: Dict[str, pd.DataFrame] = dict((k, v) for k, v in dfs_dict.items() if k in [program_1, program_2])
    dfs_valid: List[pd.DataFrame] = [dfs_filter[program] for program in [program_1, program_2]]

    # comparison DataFrame columns
    columns = [ 14, 16, 22, 52, 202, 1000 ]

    # compute rows of program_1 and program_2
    program_rows_np = np.asarray(create_list_of_ms_at_row(dfs_valid, columns), dtype=np.double)

    # compute difference row
    diff_np = program_rows_np[0] - program_rows_np[1]

    # compute improvement row (expressed as a percentage number)
    improvement_percentage = np.around(100 * diff_np / program_rows_np[0], decimals=N_DECIMALS_PERCENTAGE).tolist()

    # remove numpy wrapper
    program_1_row, program_2_row = program_rows_np.tolist()
    diff = diff_np.tolist()

    data = [
        [program_1, *program_1_row],
        [program_2, *program_2_row],
        ['Difference', *diff],
        ['Improvement %', *improvement_percentage]
    ]

    return pd.DataFrame(data, columns=['', *columns])

def compare_n_programs(dfs_dict: Dict[str, pd.DataFrame], programs: List[str], columns: List[str]):

    df = pd.DataFrame()

    headers = False

    for program in programs:

        if not headers:
            data = dfs_dict[program][['filename', 'nodes']]
            df = pd.concat([df, data], axis=1, sort=False)
            headers = True

        data = dfs_dict[program][columns]
        df = pd.concat([df, data], axis=1, sort=False)

    # sort by nodes and then by instances
    df = df.sort_values(by=['nodes', 'filename'])

    return df

def calculate_error(exact, output):
    """
    :param exact: A list with ground truth values.
    :param output: A list with actual values.
    :return: The absolute error in %.
    """
    return np.abs(np.around(100 * (np.subtract(exact, output)) / exact, decimals=N_DECIMALS_PERCENTAGE))

def print_comparison_time(dfs_dict: Dict[str, pd.DataFrame], programs: List[str]):
    print(f'Comparison {names_to_vs(programs)}.')

    df_comparison = compare_2_programs(dfs_dict, programs[0], programs[1])
    pretty_print_pandas(df_comparison)

def print_comparison(dfs_dict: Dict[str, pd.DataFrame], programs: List[str], columns: List[str]):
    print(f'Comparison {names_to_vs(programs)}.')

    df_comparison = compare_n_programs(dfs_dict, programs, columns)
    pretty_print_pandas(df_comparison)

def pretty_print_pandas(df: pd.DataFrame, tablefmt='pretty'):
    """
    Pretty-printer utility for Pandas DataFrames.
    :param df: DataFrame to pretty-print
    :param tablefmt: 'pretty' | 'psql' | 'latex'
    """
    if FOR_LATEX:
        tablefmt = 'latex'  # print out \tabular{} LaTeX tables
    print(tabulate(df.round(N_DECIMALS), headers='keys', tablefmt=tablefmt, showindex=False))
    print('\n')

def merge_dataframes_helper(dfs: List[pd.DataFrame], use_meta=True) -> pd.DataFrame:
    """
    Create a new in-memory DataFrame that, for each row of all the dataframes in dfs, keeps only the first row
    values merging `program_time`, `discovery_time` and `full_contraction` by mean, `discovery_iteration` by median,
    while other columns are taken arbitrarily from the first.
    :param dfs: List of DataFrames for a single program
    :return: new DataFrame merged row-wise
    """
    columns_common = ['filename', 'nodes', 'k', 'expected_min_cut', 'min_cut', 'program_time']

    if use_meta:
        columns = [*columns_common, 'discovery_time', 'discovery_iteration', 'full_contraction', 'min_cut_error']
    else:
        columns = [*columns_common, 'min_cut_error']

    # columns = ['filename', 'nodes', 'k', 'expected_min_cut', 'min_cut', 'program_time', 'discovery_time',
    #            'discovery_iteration', 'full_contraction', 'min_cut_error' ]
    merged_df = pd.DataFrame(columns=columns)

    n_rows = dfs[0].shape[0]
    for row in range(n_rows):
        row_index = 0  # arbitrarily choosing first row
        program_time = get_mean_at_row(dfs, row, column='program_time') / 1000.0 # to ms

        if use_meta:
            discovery_time = get_mean_at_row(dfs, row, column='discovery_time') / 1000.0 # to ms
            discovery_iteration = get_median_at_row(dfs, row, column='discovery_iteration')
            full_contraction = get_mean_at_row(dfs, row, column='full_contraction')

        data = dfs[row_index].loc[row].copy()  # need to copy to avoid side effect on original dataframe.
        data['program_time'] = program_time

        if use_meta:
            data['discovery_time'] = discovery_time
            data['discovery_iteration'] = discovery_iteration
            data['full_contraction'] = full_contraction

        data['min_cut_error'] = 100 * (data['min_cut'] - data['expected_min_cut']) / data['expected_min_cut']

        merged_df = merged_df.append(data)

    return merged_df.round(decimals=N_DECIMALS)

def merge_dataframes(dfs_dict: Dict[str, List[pd.DataFrame]]) -> Dict[str, pd.DataFrame]:
    """
    For each list of DataFrames stored in dfs_dict keys, merge rows with some logics (see `merge_dataframes_helper`).
    :param dfs_dict: original map of list of benchmark DataFrames for each program
    :return: merged map of 1 dataframe for each program
    """
    dfs_flat_min = [merge_dataframes_helper(dfs, use_meta=programs_use_meta[i]) for i, dfs in enumerate(dfs_dict.values())]
    return dict(zip(programs, dfs_flat_min))

def keep_only_interesting_nodes(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    return pd.DataFrame([
        df[columns].loc[get_row_index_with_max_edges(14)],
        df[columns].loc[get_row_index_with_max_edges(16)],
        df[columns].loc[get_row_index_with_max_edges(22)],
        df[columns].loc[get_row_index_with_max_edges(52)],
        df[columns].loc[get_row_index_with_max_edges(202)],
        df[columns].loc[get_row_index_with_max_edges(1000)],
    ], columns=columns)

def dataframe_to_stringio(df: pd.DataFrame) -> StringIO:
    csv_buffer = StringIO()

    # removes the number of vertexes/edges
    df.drop(columns=['n', 'm']) \
        .rename(columns={'output': 'MST', 'filename': 'File'}) \
        .to_csv(csv_buffer, index=None, sep=';')

    csv_buffer.seek(0)
    return csv_buffer

def export_dataframes_min_to_latex(dfs: Dict[str, pd.DataFrame]):
    dfs_list = dfs.values()
    csv_list = list(map(dataframe_to_stringio, dfs_list))

    for i, program in enumerate(dfs.keys()):
        tably_instance = Tably(SimpleNamespace(
            files=[csv_list[i]],
            align=['l', 'l', 'c'],
            caption=f'Risultati di {program}',
            no_indent=False,
            skip=0,
            label=f'table:{program}-results',
            no_header=False,
            outfile=f'../report/{program}.min.tex',
            separate_outfiles=None,
            preamble=False,
            sep='semi',
            units='-',
            no_escape=False,
            fragment=False,
            fragment_skip_header=False,
            replace=True,
        ))
        tably_instance.run()

def show_or_save_plot(title: str):
    if IS_SAVE_PLOT_ENABLED:
        if not os.path.exists(f'./{PLOT_IMG_DIR}'):
            os.makedirs(f'./{PLOT_IMG_DIR}')
        out_title = title.translate ({ord(c): "_" for c in " !@#$%^&*()[]{};:,./<>?\|`~-=+\n"})
        plt.savefig(f'./{PLOT_IMG_DIR}/{out_title}.png')
        plt.close()
    else:
        plt.show()

def names_to_vs(names: List[str]) -> str:
    """
    Return a string made by names joined by "vs".
    """
    return reduce(lambda x, y: x + ' vs ' + y, names) if len(names) > 1 else names[0]

def names_to_dfs(names: List[str], dfs) -> Dict[str, List[pd.DataFrame]]:
    """
    Return a dict where for each name the key is one of the name given and the
    value is the corresponding benchmark dataframe.
    """
    return reduce(lambda x, y: {**x, **y}, map(lambda n: {n: dfs[n]}, names))

def filter_df(df: pd.DataFrame, pred):
    """
    Return the filtered version of given dataframe w.r.t. pred, i.e. a dataframe where pred holds on each row.
    :param df: A dataframe.
    :param pred: A predicate over a dataset row.
    """
    return pd.DataFrame([df.loc[i] for i in range(len(df)) if pred(df.loc[i])], columns=df.columns)

def plot_line(dfs: Dict[str, pd.DataFrame], x: str, y: str, y_log=False):
    """
    Plots `dfs` data.
    :dfs: a dictionary with the benchmark dataframe to plot.
    :x: the x axis name.
    :y: the y axis name.
    :y_log: a boolean flag that indicates whether y should be log scaled or not.
    """
    plt.figure(figsize=(14, 7))
    for k, v in dfs.items():
        g = sns.lineplot(v[x], v[y], label=k)
        if y_log:
            g.set_yscale('log')

def plot_series(names: List[str], dfs: pd.DataFrame, y_log=False):
    """
    Plot given programs.
    """
    benchmark_subset = names_to_dfs(names, dfs)
    plot_line(benchmark_subset, x='nodes', y='program_time', y_log=y_log)

def plot_comparison(names: List[str], dfs: Dict[str, pd.DataFrame], title, pred=lambda _: True, filename=None, y_log=False):
    """
    Plot filtered dataframes w.r.t. pred.
    :param names: A list of programs.
    :param dfs: A dictionary of dataframes of the type (name -> dataframe).
    :param pred: A predicate over a dataframe benchmark row.
    :param title: The plot title.
    :param filename: The plot filename, default is `title`.
    :param y_log: a boolean flag that indicates whether y should be log scaled or not.
    """
    d = {}
    for i in range(len(names)):
        name = names[i]
        df = filter_df(dfs[name], pred=pred)
        d = {**d, **{name: df}}

    plot_series(names, d, y_log=y_log)

    plt.title(title)

    show_or_save_plot(filename if filename is not None else title)

def plot_precision_comparison(names: List[str], dfs: Dict[str, pd.DataFrame], title: str, pred=lambda _: True, filename=None, y_log=False):
    """
    Plot filtered dataframes w.r.t. pred.
    :param names: A list of programs.
    :param dfs: A dictionary of dataframes of the type (name -> dataframe).
    :param pred: A predicate over a dataframe benchmark row.
    :param title: The plot title.
    :param filename: The plot filename, default is `title`.
    :param y_log: A flag for scaling y axis.
    """
    df = pd.DataFrame()

    for i in range(len(names)):
        name = names[i]
        dfx = dfs[name]
        dfx['error'] = calculate_error(GROUND_TRUTH_DF['exact'].tolist(), dfs[name]['output'].tolist())
        dfx['name'] = name
        dfx = filter_df(dfx, pred=pred)
        df = df.append(dfx)

    # not working properly
    # sns.set(rc={'figure.figsize':(14, 7)})

    g = sns.catplot(x='d', y='error', hue='name', data=df,  kind='bar', aspect=2,  margin_titles=True)
    g.set_ylabels("Error (%)")
    g.set_xlabels("Nodes")
    g.set_titles(title)

    if y_log:
        ax = g.axes
        ax[0,0].set_yscale('log')

    show_or_save_plot(filename if filename is not None else title)

def karger_full_contraction_chart(dfs):
    karger_df = dfs[KARGER].copy()
    karger_df['full_contraction_asymptotic'] = 0.3 * np.square(karger_df['nodes']).astype(float)
    title = 'Tempo di esecuzione di full contraction rispetto al numero di nodi'

    g = sns.lineplot(karger_df['nodes'], karger_df['full_contraction'], label='Full Contraction')
    g = sns.lineplot(karger_df['nodes'], karger_df['full_contraction_asymptotic'], label='Full Contraction Asintotico: n^2')
    g.set(xlabel='Nodi', ylabel='Tempo (μs)')
    g.set_xlim(6, 200)
    # g.set_yscale('log')

    plt.title(title)
    show_or_save_plot(title)

def karger_discovery_vs_program_time_chart(dfs):
    karger_df = dfs[KARGER].copy()
    title = f'Confronto tra discovery time e runtime rispetto al numero di nodi'

    karger_df['program_time'] = karger_df['program_time'] / 1000.0  # to seconds
    karger_df['discovery_time'] = karger_df['discovery_time'] / 1000.0  # to seconds

    g = sns.lineplot(karger_df['nodes'], karger_df['discovery_time'], label=f'{KARGER} (Discovery Time)')
    g = sns.lineplot(karger_df['nodes'], karger_df['program_time'], label=f'{KARGER} (Runtime)')
    g.set(xlabel='Nodi', ylabel='Tempo (s)')
    g.set_yscale('log')

    plt.title(title)
    show_or_save_plot(title)

def karger_relative_error(dfs):
    karger_df = dfs[KARGER].copy()
    title = f'Errore relativo dell\'output rispetto al numero di nodi'

    g = sns.barplot(karger_df['nodes'], karger_df['min_cut_error'], label=f'${KARGER} (Errore Relativo)')
    g.set(xlabel='Nodi', ylabel='Errore (%)')

    plt.title(title)
    show_or_save_plot(title)

def karger_def_vs_tout_running_time(dfs):
    # create copy to avoid side effect
    karger_def_df = dfs[KARGER].copy()
    karger_tout_df = dfs[KARGER_TOUT].copy()

    # add a column program that represents the program name
    karger_def_df["program"] = KARGER
    karger_tout_df["program"] = KARGER_TOUT

    karger_def_df['program_time'] = karger_def_df['program_time'] / (1000 * 60)
    karger_tout_df['program_time'] = karger_tout_df['program_time'] / (1000 * 60)

    # merge dataset to a single one by appending rows
    df = karger_def_df
    df = df.append(karger_tout_df)

    df.to_csv('karger_vs_karger_timeout.csv')

    # create the barplot with `hue='program'`
    g = sns.barplot(x='nodes', y='program_time', hue='program', data=df)
    g.set(xlabel='Nodi', ylabel='Tempo (min)')

    title = f'Confronto del running time rispetto al numero di nodi per\n {KARGER} e {KARGER_TOUT}'

    plt.title(title)
    show_or_save_plot(title)

def karger_def_vs_tout_relative_error(dfs):
    # create copy to avoid side effect
    karger_def_df = dfs[KARGER].copy()
    karger_tout_df = dfs[KARGER_TOUT].copy()

    # add a column program that represents the program name
    karger_def_df["program"] = KARGER
    karger_tout_df["program"] = KARGER_TOUT

    # merge dataset to a single one by appending rows
    df = karger_def_df
    df = df.append(karger_tout_df)

    # create the barplot with `hue='program'`
    g = sns.barplot(x='nodes', y='min_cut_error', hue='program', data=df)
    g.set(xlabel='Nodi', ylabel='Errore (%)')

    title = f'Confronto dell\'errore relativo rispetto al numero di nodi per\n {KARGER} e {KARGER_TOUT}'

    plt.title(title)
    show_or_save_plot(title)

def karger_discovery_iter_vs_estimated(dfs):
    karger_df = dfs[KARGER].copy()

    g = sns.lineplot(karger_df['nodes'], karger_df['discovery_iteration'], label=f'{KARGER} (Discovery Iteration)')
    g = sns.lineplot(karger_df['nodes'], karger_df['k'].astype(np.double), label=f'{KARGER} (K)')
    g.set(xlabel='Nodi', ylabel='Discovery Iteration')
    g.set_yscale('log')

    title = f'Confronto Discovery Iteration stimata vs attuale\n rispetto al numero di nodi per {KARGER}'

    plt.title(title)
    show_or_save_plot(title)

def karger_runtime(dfs):
    karger_df = dfs[KARGER].copy()

    # add estimated runtime
    n = karger_df['nodes'].astype(np.double)
    r = 0.0002 * (n ** 4) * np.log(n)

    # save it in the dataframe
    karger_df['estimated_program_time'] = r

    # plot the result
    g = sns.lineplot(karger_df['nodes'], karger_df['program_time'], label=f'{KARGER} (Runtime)')
    g = sns.lineplot(karger_df['nodes'], karger_df['estimated_program_time'], label=f'{KARGER} Asintotico: n^4 log(n)')
    g.set(xlabel='Nodi', ylabel='Tempo (ms)')
    # g.set_yscale('log')

    title = f'Tempo di esecuzione attuale e asintotico\n rispetto al numero di nodi per {KARGER}'

    plt.title(title)
    show_or_save_plot(title)

def karger_vs_stein_runtime(dfs):
    karger_df = dfs[KARGER].copy()
    karger_stein_df = dfs[KARGER_STEIN].copy()

    karger_df['program_time'] = karger_df['program_time'] / 1000.0  # to seconds
    karger_stein_df['program_time'] = karger_stein_df['program_time'] / 1000.0  # to seconds

    g = sns.lineplot(karger_df['nodes'], karger_df['program_time'], label=f'{KARGER} (Runtime)')
    g = sns.lineplot(karger_stein_df['nodes'], karger_stein_df['program_time'], label=f'{KARGER_STEIN} (Runtime)')
    g.set(xlabel='Nodi', ylabel='Tempo (s)')
    g.set_yscale('log')

    title = f'{KARGER} vs {KARGER_STEIN}'

    plt.title(title)
    show_or_save_plot(title)

def karger_timeout_vs_stein_runtime(dfs):
    karger_df = dfs[KARGER_TOUT].copy()
    karger_stein_df = dfs[KARGER_STEIN].copy()

    karger_df['program_time'] = karger_df['program_time'] / 1000.0  # to seconds
    karger_stein_df['program_time'] = karger_stein_df['program_time'] / 1000.0  # to seconds

    g = sns.lineplot(karger_df['nodes'], karger_df['program_time'] , label=f'{KARGER_TOUT} (Runtime)')
    g = sns.lineplot(karger_stein_df['nodes'], karger_stein_df['program_time'], label=f'{KARGER_STEIN} (Runtime)')
    g.set(xlabel='Nodi', ylabel='Tempo (s)')
    # g.set_yscale('log')

    title = f'{KARGER_TOUT} vs {KARGER_STEIN}'

    plt.title(title)
    show_or_save_plot(title)

if __name__ == '__main__':
    if IS_HELP:
        print(__doc__)
        exit(0)

    # benchmark CSVs stored in DataFrames and grouped by program name
    generator = zip(programs, [read_csvs_of_program(program) for program in programs])
    dataframes: Dict[str, List[pd.DataFrame]] = dict(generator)

    # is some CSV is invalid, throws an AssertionError
    check_validity(dataframes)

    # for each CSV, only the minimum values for the 'program_time' column are retained
    # len(dataframes_min) == len(programs)
    dataframes_merge = merge_dataframes(dataframes)

    if IS_TABLE_ENABLED:
        # compare multiple programs to show potential improvements

        # Appendix (runtime): Karger vs KargerTimeout running time (program vs discovery time)
        print_comparison(dataframes_merge, [ KARGER ], ['discovery_time', 'program_time'])
        print_comparison(dataframes_merge, [ KARGER_TOUT ], ['discovery_time', 'program_time'])

        # Appendix (error): Karger vs KargerTimeout approx error
        print_comparison(dataframes_merge, [ KARGER ], ['min_cut', 'expected_min_cut', 'min_cut_error'])
        print_comparison(dataframes_merge, [ KARGER_TOUT ], ['min_cut', 'expected_min_cut', 'min_cut_error'])

        print_comparison(dataframes_merge, [KARGER_STEIN], ['program_time'])

        # Appendix: full contraction statistics
        print_comparison(dataframes_merge, [ KARGER ], ['program_time', 'full_contraction'])

    # export minimized in-memory CSV files to LaTeX tables (they will still require some manual work tho)
    # export_dataframes_merge_to_latex(dataframes_merge)

    if IS_PLOT_ENABLED:

        # Q1: Full contraction scaling with nodes
        karger_full_contraction_chart(dataframes_merge)

        # Q2: Estimated discovery iter
        karger_runtime(dataframes_merge)

        # Q3: For each dataset compare discovery time with algorithm runtime.
        karger_discovery_vs_program_time_chart(dataframes_merge)
        karger_discovery_iter_vs_estimated(dataframes_merge)

        # Q4: Output, Expected, Relative Error
        karger_relative_error(dataframes_merge)  # all zeros!
        karger_def_vs_tout_running_time(dataframes_merge)
        karger_def_vs_tout_relative_error(dataframes_merge)

        # Compare KargerMinCut with KargerSteinMinCut
        karger_vs_stein_runtime(dataframes_merge)
        karger_timeout_vs_stein_runtime(dataframes_merge)


text:
import numpy as numpy
x = numpy.array([(1,2,3) , (3,4,5)])
print(x.ravel())


text:
from logging import getLogger

import numpy as np
from numpy import ndarray
from ordered_set import OrderedSet
from text_selection.common.helper import get_empty_columns
from text_selection.greedy.greedy_iterator import GreedyIterator
from text_selection.selection import KeySelector

class OptimizedGreedyIterator(GreedyIterator):
  def __init__(self, data: ndarray, data_indices: OrderedSet[int], preselection: ndarray, key_selector: KeySelector) -> None:
    logger = getLogger(__name__)

    empty_columns = get_empty_columns(data, data_indices, preselection)
    if len(empty_columns) > 0:
      old_count = data.shape[1]
      data: ndarray = np.delete(data, empty_columns, axis=1)
      preselection: ndarray = np.delete(preselection, empty_columns, axis=0)
      logger.info(
          f"Removed {len(empty_columns)} out of {old_count} columns.")
      del empty_columns

    super().__init__(
      data=data,
      preselection=preselection,
      data_indices=data_indices,
      key_selector=key_selector,
    )

max_stars_repo_name: stefantaubert/text-selection

max_issues_repo_name: stefantaubert/text-selection

max_forks_repo_name: stefantaubert/text-selection

reason: import numpy,from numpy


text:
import logging
from collections import OrderedDict

import numpy as np
from PIL import Image

from model_tools.activations.core import ActivationsExtractorHelper
from model_tools.utils import fullname

SUBMODULE_SEPARATOR = '.'

class PytorchWrapper:
    def __init__(self, model, preprocessing, identifier=None, *args, **kwargs):
        import torch
        logger = logging.getLogger(fullname(self))
        self._device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        logger.debug(f"Using device {self._device}")
        self._model = model
        self._model = self._model.to(self._device)
        identifier = identifier or model.__class__.__name__
        self._extractor = self._build_extractor(
            identifier=identifier, preprocessing=preprocessing, get_activations=self.get_activations, *args, **kwargs)
        self._extractor.insert_attrs(self)

    def _build_extractor(self, identifier, preprocessing, get_activations, *args, **kwargs):
        return ActivationsExtractorHelper(
            identifier=identifier, get_activations=get_activations, preprocessing=preprocessing,
            *args, **kwargs)

    @property
    def identifier(self):
        return self._extractor.identifier

    @identifier.setter
    def identifier(self, value):
        self._extractor.identifier = value

    def __call__(self, *args, **kwargs):  # cannot assign __call__ as attribute due to Python convention
        return self._extractor(*args, **kwargs)

    def get_activations(self, images, layer_names):
        import torch
        from torch.autograd import Variable
        images = [torch.from_numpy(image) for image in images]
        images = Variable(torch.stack(images))
        images = images.to(self._device)
        self._model.eval()

        layer_results = OrderedDict()
        hooks = []

        for layer_name in layer_names:
            layer = self.get_layer(layer_name)
            hook = self.register_hook(layer, layer_name, target_dict=layer_results)
            hooks.append(hook)

        self._model(images)
        for hook in hooks:
            hook.remove()
        return layer_results

    def get_layer(self, layer_name):
        if layer_name == 'logits':
            return self._output_layer()
        module = self._model
        for part in layer_name.split(SUBMODULE_SEPARATOR):
            module = module._modules.get(part)
            assert module is not None, f"No submodule found for layer {layer_name}, at part {part}"
        return module

    def _output_layer(self):
        module = self._model
        while module._modules:
            module = module._modules[next(reversed(module._modules))]
        return module

    @classmethod
    def _tensor_to_numpy(cls, output):
        return output.cpu().data.numpy()

    def register_hook(self, layer, layer_name, target_dict):
        def hook_function(_layer, _input, output, name=layer_name):
            target_dict[name] = PytorchWrapper._tensor_to_numpy(output)

        hook = layer.register_forward_hook(hook_function)
        return hook

    def __repr__(self):
        return repr(self._model)

    def layers(self):
        for name, module in self._model.named_modules():
            if len(list(module.children())) > 0:  # this module only holds other modules
                continue
            yield name, module

    def graph(self):
        import networkx as nx
        g = nx.DiGraph()
        for layer_name, layer in self.layers():
            g.add_node(layer_name, object=layer, type=type(layer))
        return g

def load_preprocess_images(image_filepaths, image_size, **kwargs):
    images = load_images(image_filepaths)
    images = preprocess_images(images, image_size=image_size, **kwargs)
    return images

def load_images(image_filepaths):
    return [load_image(image_filepath) for image_filepath in image_filepaths]

def load_image(image_filepath):
    with Image.open(image_filepath) as pil_image:
        if 'L' not in pil_image.mode.upper() and 'A' not in pil_image.mode.upper():  # not binary and not alpha
            # work around to https://github.com/python-pillow/Pillow/issues/1144,
            # see https://stackoverflow.com/a/30376272/2225200
            return pil_image.copy()
        else:  # make sure potential binary images are in RGB
            rgb_image = Image.new("RGB", pil_image.size)
            rgb_image.paste(pil_image)
            return rgb_image

def preprocess_images(images, image_size, **kwargs):
    preprocess = torchvision_preprocess_input(image_size, **kwargs)
    images = [preprocess(image) for image in images]
    images = np.concatenate(images)
    return images

def torchvision_preprocess_input(image_size, **kwargs):
    from torchvision import transforms
    return transforms.Compose([
        transforms.Resize((image_size, image_size)),
        torchvision_preprocess(**kwargs),
    ])

def torchvision_preprocess(normalize_mean=(0.485, 0.456, 0.406), normalize_std=(0.229, 0.224, 0.225)):
    from torchvision import transforms
    return transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean=normalize_mean, std=normalize_std),
        lambda img: img.unsqueeze(0)
    ])

max_stars_repo_name: shashikg/model-tools

max_issues_repo_name: shashikg/model-tools

max_forks_repo_name: shashikg/model-tools

reason: import numpy,import networkx


text:
import numpy as np
import os
import gym
from gym.spaces.box import Box
from core.envs.atari_wrappers import make_atari, wrap_deepmind

try:
    import pybullet_envs
    import roboschool
except ImportError:
    pass

def make_env(args, rank):
    def _thunk():
        env = gym.make(args.game)
        is_atari = hasattr(gym.envs, 'atari') and isinstance(env.unwrapped, gym.envs.atari.atari_env.AtariEnv)
        if is_atari:
            env = make_atari(args.game)
        env.seed(args.seed + rank)
        if is_atari:
            env = wrap_deepmind(env, frame_stack=True)
        return env
    return _thunk

max_stars_repo_name: onlytailei/pytorch-distributed

max_issues_repo_name: onlytailei/pytorch-distributed

max_forks_repo_name: onlytailei/pytorch-distributed


text:
#! /usr/bin/env python
# ______________________________________________________________________

from numba.decorators import numba_compile

import numpy

import unittest

# ______________________________________________________________________

def while_loop_fn_0(max_index, indexable):
    i = 0
    acc = 0.
    while i < max_index:
        acc += indexable[i]
        i += 1
    return acc

# ______________________________________________________________________

def while_loop_fn_1(indexable):
    i = 0
    acc = 0.
    while i < len(indexable):
        acc += indexable[i]
        i += 1
    return acc

# ______________________________________________________________________

def while_loop_fn_2(ndarr):
    i = 0
    acc = 0.
    while i < ndarr.shape[0]:
        acc += ndarr[i]
        i += 1
    return acc

# ______________________________________________________________________

def while_loop_fn_3(count):
    i = 0
    acc = 1.
    while i < count:
        acc *= 2
        i += 1
    return acc

# ______________________________________________________________________

def while_loop_fn_4(start, stop, inc):
    '''Intended to parallel desired translation target for
    test_forloop.for_loop_fn_1.'''
    acc = 0
    i = start
    while i != stop:
        acc += i
        i += inc
    return acc

# ______________________________________________________________________

def while_loop_fn_5(i_max, j_max):
    j = 1.
    acc = 0.
    while j < j_max:
        i = 1.
        while i < i_max:
            acc += i * j
            i += 1.
        j += 1.
    return acc

# ______________________________________________________________________

class TestWhile(unittest.TestCase):
    def _do_test(self, function, arg_types, *args, **kws):
        _numba_compile = (numba_compile(arg_types = arg_types)
                          if arg_types is not None else numba_compile())
        compiled_fn = _numba_compile(function)
        self.assertEqual(compiled_fn(*args, **kws), function(*args, **kws))

    def test_while_loop_fn_0(self):
        test_data = numpy.array([1., 2., 3.])
        self._do_test(while_loop_fn_0, ['l', ['d']], len(test_data), test_data)

    def test_while_loop_fn_1(self):
        self._do_test(while_loop_fn_1, [['d']], numpy.array([1., 2., 3.]))

    def test_while_loop_fn_2(self):
        self._do_test(while_loop_fn_2, [['d']], numpy.array([1., 2., 3.]))

    def test_while_loop_fn_3(self):
        compiled_fn = numba_compile(arg_types = ['l'])(while_loop_fn_3)
        compiled_result = compiled_fn(3)
        self.assertEqual(compiled_result, while_loop_fn_3(3))
        self.assertEqual(compiled_result, 8.)

    def test_while_loop_fn_4(self):
        compiled_fn = numba_compile(arg_types = ['l', 'l', 'l'],
                                    ret_type = 'l')(while_loop_fn_4)
        compiled_result = compiled_fn(1, 4, 1)
        self.assertEqual(compiled_result, while_loop_fn_4(1, 4, 1))
        self.assertEqual(compiled_result, 6)

    def test_while_loop_fn_5(self):
        compiled_fn = numba_compile(arg_types = ['d', 'd'])(while_loop_fn_5)
        compiled_result = compiled_fn(3, 4)
        self.assertEqual(compiled_result, while_loop_fn_5(3, 4))
        self.assertEqual(compiled_result, 18.)

# ______________________________________________________________________

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

# ______________________________________________________________________
# End of test_while.py

max_stars_repo_name: teoliphant/numba

max_issues_repo_name: teoliphant/numba

max_forks_repo_name: teoliphant/numba

reason: import numpy,import numba,from numba


text:
import os

import numpy as np

# c1 [EUR/m], c2 [EUR/m2]
c1_default, c2_default = 212, 4464
const_coeff = {
    "DE": (349, 4213),
    "ES": (354, 4314),
    "FR": (349, 4213),
    "HR": (349, 4213),
    "IT": (349, 4213),
    "LT": (71, 3262),
    "HU": (349, 4213),
    "NL": (549, 3370),
    "SE": (439, 4073),
    "UK": (549, 2236),
    "AT": (349, 4213),
    "BE": (549, 3370),
    "BG": (349, 4213),
    "CY": (540, 2087),
    "CZ": (349, 4213),
    "DK": (439, 4073),
    "EE": (71, 3262),
    "FI": (439, 4073),
    "EL": (540, 2087),
    "IE": (549, 2236),
    "LV": (71, 3262),
    "LU": (549, 3370),
    "MT": (540, 2087),
    "PL": (349, 4213),
    "PT": (354, 4314),
    "RO": (349, 4213),
    "SK": (349, 4213),
    "SI": (540, 2087),
}

class Param:
    def __init__(self, params):
        self.output_layer_selection = params["output_layer_selection"]
        self.scenario = params["scenario"]
        self.country = params["country"]
        self.distribution_grid_cost_ceiling = params["distribution_grid_cost_ceiling"]
        # pixel threshold may not be changed. 20 is the correct value.
        self.pix_threshold = params["pix_threshold"]
        self.DH_threshold = params["DH_threshold"]
        self.total_investment_annuity = np.inf
        self.investment_increasing_factor = 1
        self.start_year = params["start_year"]
        if params["start_year"] > params["last_year"]:
            params["last_year"] = params["start_year"] + 1
        self.last_year = params["last_year"]
        self.st_dh_connection_rate = params["st_dh_connection_rate"]
        self.end_dh_connection_rate = params["end_dh_connection_rate"]
        self.depreciation_period = params["depreciation_period"]
        self.interest = params["interest"]
        if params["use_default_cost_factors"] is True:
            self.c1, self.c2 = const_coeff[self.country]
        else:
            self.c1 = params["c1"]
            self.c2 = params["c2"]
        self.case = (
            str(self.st_dh_connection_rate)
            + "_"
            + str(self.end_dh_connection_rate)
            + "_"
            + str(self.distribution_grid_cost_ceiling)
        )

class Out_File_Path(Param):
    def __init__(self, directory, in_raster_hdm, in_raster_gfa, params):
        super().__init__(params)
        self.inRasterHDM_st = in_raster_hdm
        self.inRasterHDM_end = in_raster_hdm
        self.inRasterGFA_st = in_raster_gfa
        self.inRasterGFA_end = in_raster_gfa
        self.dstDir = os.path.join(directory, self.case)
        self.maxDHdem = os.path.join(self.dstDir, "max_dh_demand_in_year.tif")
        self.supplied_heat_during_investment_period = os.path.join(
            self.dstDir, "total_supplied_heat_during_investment_period.tif"
        )
        self.invest_Euro = os.path.join(self.dstDir, "absolute_investment_Euro.tif")
        # self.hdm_cut_last_year = os.path.join(self.dstDir, 'hdm_cut_last_year.tif')
        self.total_dist_pipe_length = os.path.join(
            self.dstDir, "total_distribution_pipe_length.tif"
        )
        self.total_serv_pipe_length = os.path.join(
            self.dstDir, "total_service_pipe_length.tif"
        )
        self.coh_area_bool = os.path.join(self.dstDir, "coherent_areas_bool.tif")
        self.labels = os.path.join(self.dstDir, "labels.tif")
        self.inv_dist = os.path.join(self.dstDir, "specific_distribution_pipe_cost.tif")
        self.inv_service_pipe = os.path.join(
            self.dstDir, "specific_service_pipe_cost.tif"
        )
        self.inv_sum = os.path.join(self.dstDir, "specific_investment_total.tif")
        self.output_shp1 = os.path.join(self.dstDir, "tmp_shape.shp")
        self.output_shp2 = os.path.join(self.dstDir, "tmp_summary.shp")
        self.output_geojson = os.path.join(self.dstDir, "gis_summary.geojson")
        self.output_csv = os.path.join(self.dstDir, "summary.csv")
        self.logfile = os.path.join(self.dstDir, "info.log")

max_stars_repo_name: historeno/enermaps

max_issues_repo_name: historeno/enermaps

max_forks_repo_name: historeno/enermaps


text:
import numpy as np
from typing import NewType

metro = NewType('m', float)
pressao = NewType('MPa', float)
pressao_em_kn = NewType('kN/cm²', float)
momento = NewType('kNm', float)

def secao(bw, h):
    c = 2
    d = h - c
    return bw/100, d/100

def concreto(fck):
    gammac = 1.4
    return (fck * gammac) * 1000

def aco(fyk):
    gammas = 1.15
    return fyk / gammas

def momento(mk):
    gammad = 1.4
    return mk * gammad

def calculo_da_linha_neutra(bw, d, md, fcd):
    A = 0.68 * d
    B = 1.088 * (md / (bw * fcd))
    x_pos = (A + np.sqrt(A ** 2 - B)) / 0.544
    x_neg = (A - np.sqrt(A ** 2 - B)) / 0.544
    return x_pos, x_neg

def verificao_linha_neutra(d, x_pos, x_neg):
    # vericação da linha neutra
    if x_pos < 0.259 * d or 0.259 <= x_pos <= 0.45 * d:
        print('Xpos foi escolhido!')
        x_esc = x_pos
        return x_esc

    if x_neg < 0.259 * d or 0.259 <= x_neg <= 0.45 * d:
        print('Xneg foi escolhido!')
        x_esc = x_neg
        return x_neg

def calculo_armadura(d, x_esc, md, fyd):
    z = d - 0.4 * x_esc
    armadura_longitudinal = md / (z * fyd)
    return armadura_longitudinal

def calculo_total(bw: metro, h: metro, fck: pressao, fyk: pressao_em_kn, mk: momento) -> float:
    # dados de entrada do problema
    bw, d = secao(bw=bw, h=h)
    fcd = concreto(fck)
    fyd = aco(fyk)
    md = momento(mk)

    # calculo do problema
    x_pos, x_neg = calculo_da_linha_neutra(bw=bw, d=d, md=md, fcd=fcd)
    x = verificao_linha_neutra(d=d, x_pos=x_pos, x_neg=x_neg)
    return calculo_armadura(d=d, x_esc=x, md=md, fyd=fyd)

max_stars_repo_name: kitawarairon/curso-pyton-engenharia

max_issues_repo_name: kitawarairon/curso-pyton-engenharia

max_forks_repo_name: kitawarairon/curso-pyton-engenharia


text:
import numpy as np

def classify_persons(frame_time, frame, detections, classify_person_flow):
    person_class_dict = {}
    for i, obj in enumerate(detections):
        left, top, right, bottom = obj.rect
        label = obj.label
        id = int(label.split(' ')[-1]
                 ) if isinstance(label, str) else int(label)

        if id >= 0:
            crop_img = frame[top:bottom, left:right]

            x1y1x2y2 = (left, top, right, bottom)
            person_class, detect_lines, person_action = classify_person_flow(frame_time, id, crop_img, x1y1x2y2)
            person_class_dict[id] = (person_class, detect_lines, person_action)
    return person_class_dict

def classify_persons_per_frame(frame_times, frames, all_objects, classify_person_flow, fps='', show_all_detections=True,
                               max_window_size=(1920, 1080), stack_frames='vertical'):
    assert len(frames) == len(all_objects)
    assert stack_frames in ['vertical', 'horizontal']
    vis = None
    person_class_dict = {}
    for i, (frame_time, frame, objects) in enumerate(zip(frame_times, frames, all_objects)):
        person_class_dict.update(classify_persons(
            frame_time, frame, objects, classify_person_flow))
        if vis is not None:
            if stack_frames == 'vertical':
                vis = np.vstack([vis, frame])
            elif stack_frames == 'horizontal':
                vis = np.hstack([vis, frame])
        else:
            vis = frame
    return person_class_dict


text:
import os
import logging
from PIL import Image
from model import FaceCascade, FindFaces, Blur, HaarBoxes, yolo_v2_model
from load_data import load_pictures
import matplotlib.pyplot as plt
from optparse import OptionParser
import cv2
import numpy as np
from utils_yolo_v2 import WeightReader, decode_netout, draw_boxes
import glob
import time
import xml.etree.ElementTree as ET

try:
   base_dir = os.environ["BASE_DIR"]
except KeyError:
   print("Not exist environment variable %s" % "try sourcing build/environment.sh")

blur_mode = 'medianBlur'

dir_jpg_all_files = base_dir + '/data/raw/face/test/*'
all_test_jpg_files = glob.glob(dir_jpg_all_files)

dir_ann_all_files = base_dir + '/data/raw/face/test_ann/*'
all_test_ann_files = glob.glob(dir_ann_all_files)

####### define IoU

############################################

name = []
boxes_true = []
boxes_yolo = []
boxes_haar = []
time_yolo = []
time_haar = []

############################################

haar_model = FaceCascade()

yolo_model = yolo_v2_model()

time_yolo_all = 0
time_haar_all = 0

i = 0
for item_ann in all_test_ann_files:
    i += 1
    print(str(i) + " / " + str(len(all_test_jpg_files)))
    print(item_ann)

    name.append(item_ann)

    ##### parse true xml file #####

    tree = ET.parse(item_ann)

    for elem in tree.iter():
        if 'object' in elem.tag or 'part' in elem.tag:
            box = []
            for attr in list(elem):
                if 'bndbox' in attr.tag:
                    for dim in list(attr):
                        if 'xmin' in dim.tag:
                            box.append(int(round(float(dim.text))))
                        if 'ymin' in dim.tag:
                            box.append(int(round(float(dim.text))))
                        if 'xmax' in dim.tag:
                            box.append(int(round(float(dim.text))))
                        if 'ymax' in dim.tag:
                            box.append(int(round(float(dim.text))))
            boxes_true.append(box)
    print(boxes_true)
    print(len(boxes_true))

    print(item_ann)
    item = item_ann.replace( "xml", "jpg" )
    item = item.replace("test_ann", "test")
    print(item)

    imagePath = item
    img, img_arr, gray, img_shape = load_pictures(imagePath)

    try:
        img.shape
        print("checked for shape".format(img.shape))
    except AttributeError:
        print("shape not found")

    print("Haar Cascade")

    start_time = time.clock()
    haar_faces = FindFaces(gray, haar_model)
    execution_time_haar = time.clock() - start_time

    time_haar_all += execution_time_haar

    print(haar_faces)
    print(len(haar_faces))

    time_haar.append(execution_time_haar)
    print("Haar execution time = " + str(execution_time_haar))

    final_picture = HaarBoxes(img_arr, haar_faces)

    outPath = item.replace(".jpg", "_blur_Haar.jpg")
    outPath = outPath.replace("test", "tested")

    print(outPath)
    plt.imshow(final_picture)
    plt.savefig(outPath)

    print("YOLO")
    image = cv2.imread(imagePath)
    dummy_array = np.zeros((1,1,1,1,50,4))

    plt.figure(figsize=(10,10))

    input_image = cv2.resize(image, (416, 416))
    input_image = input_image / 255.
    input_image = input_image[:,:,::-1]
    input_image = np.expand_dims(input_image, 0)

    start_time = time.clock()
    yolo_netout = yolo_model.predict([input_image, dummy_array])
    execution_time_yolo = time.clock() - start_time
    time_yolo.append(execution_time_yolo)

    time_yolo_all += execution_time_yolo

    print("YOLO execution time = " + str(execution_time_yolo))

    boxes = decode_netout(yolo_netout[0],
                      obj_threshold=0.3,
                      nms_threshold=0.3,
                      anchors=[0.57273, 0.677385, 1.87446, 2.06253, 3.33843, 5.47434, 7.88282, 3.52778, 9.77052, 9.16828],
                      nb_class=1)

    yolo_faces = []
    for box in boxes:
        image_h, image_w, _ = image.shape
        x = int(box.xmin*image_w)
        w = int(box.ymin*image_h)
        y = int(box.xmax*image_w)
        h = int(box.ymax*image_h)
        yolo_faces.append([x, y, w, h])
    print(yolo_faces)
    print(len(yolo_faces))

    LABELS = ['face']
    image_with_boxes = draw_boxes(image, boxes, labels=LABELS)

    #final_picture = Blur(image, faces, blur_mode)

    outPath = item.replace(".jpg", "_blur_YOLO.jpg")
    outPath = outPath.replace("test", "tested")
    print(outPath)

    plt.imshow(image_with_boxes[:,:,::-1]);
    #plt.imshow(final_picture)
    plt.savefig(outPath)

print(name)
print(boxes_true)
print(boxes_yolo)
print(boxes_haar)
print(time_yolo)
print(time_haar)

print(" ")
print(" ")
print("Haar Time total")
print(time_haar_all)
print("YOLO Time total")
print(time_yolo_all)

max_stars_repo_name: tpalczew/Insight

max_issues_repo_name: tpalczew/Insight

max_forks_repo_name: tpalczew/Insight


text:
#!/usr/bin/python3
# Copyright 2013, Mozilla Corporation
# Copyright 2017-2020 Robert-André Mauchin
# Loosely based on a script written by Josh Aas
# https://github.com/bdaehlie/web_image_formats
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#     and/or other materials provided with the distribution.
#
# 3. Neither the name of the copyright holder nor the names of its
#    contributors may be used to endorse or promote products derived from this
#     software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#

import os
import errno
import subprocess
import sys
import glob
import re
import shlex
import string
import json
from multiprocessing import Pool
from timeit import Timer
import numpy as np

# Tests
vmaf = "vmaf --json --model version=vmaf_v0.6.1 --feature psnr --feature psnr_hvs --feature float_ssim --feature float_ms_ssim --feature ciede"

# Path to tmp dir to be used by the tests
tmpdir = "/tmp/"

#############################################################################

def split(cmd):
    lex = shlex.shlex(cmd)
    lex.quotes = '"'
    lex.whitespace_split = True
    lex.commenters = ""
    return list(lex)

def run_silent(cmd):
    FNULL = open(os.devnull, "w")
    rv = subprocess.call(split(cmd), stdout=FNULL, stderr=FNULL)
    if rv != 0:
        sys.stderr.write("Failure from subprocess:\n")
        sys.stderr.write("\t" + cmd + "\n")
        sys.stderr.write("Aborting!\n")
        sys.exit(rv)
    return rv

def wrapper(func, *args, **kwargs):
    def wrapped():
        return func(*args, **kwargs)

    return wrapped

def create_dir(path):
    if not os.path.exists(os.path.dirname(path)):
        try:
            os.makedirs(os.path.dirname(path))
        except OSError as exc:
            if exc.errno != errno.EEXIST:
                raise

def path_for_file_in_tmp(path):
    return tmpdir + str(os.getpid()) + os.path.basename(path)

def get_img_width(path):
    cmd = "identify -format %%w %s" % (path)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: identify\n")
        sys.exit(proc.returncode)
    lines = out.split(os.linesep)
    return int(lines[0].strip())

def get_img_height(path):
    cmd = "identify -format %%h %s" % (path)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: identify\n")
        sys.exit(proc.returncode)
    lines = out.split(os.linesep)
    return int(lines[0].strip())

def convert_img(inn, out):
    # PNG24: needed otherwise grayscale image lose their sRGB colorspace
    cmd = "convert %s PNG24:%s" % (inn, out)
    run_silent(cmd)

def detect_alpha(img):
    cmd = "identify -format '%%[channels]' %s" % (img)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: %s\n" % (cmd))
        sys.exit(proc.returncode)

    channels = out.split(os.linesep)[0]
    if channels == "srgba":
        return True;
    return False

def remove_alpha(inn, out):
    # PNG24: needed otherwise grayscale image lose their sRGB colorspace
    cmd = "convert %s -alpha off PNG24:%s" % (inn, out)
    run_silent(cmd)

def convertff_img(inn, out):
    # 10le -strict -1
    cmd = (
        "ffmpeg -y -i %s -pix_fmt yuv444p -vf scale=in_range=full:out_range=full %s"
        % (inn, out)
    )
    run_silent(cmd)

def get_score(y4m1, y4m2, target_json):
    cmd = "%s -r %s -d %s -o %s" % (vmaf, y4m1, y4m2, target_json)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: %s\n" % (cmd))
        sys.exit(proc.returncode)

    with open(target_json) as f:
        data = f.read()
        scores = json.loads(data)
        psnrhvs_score = scores["pooled_metrics"]["psnr_hvs"]["mean"]
        ssim_score = scores["pooled_metrics"]["float_ssim"]["mean"]
        msssim_score = scores["pooled_metrics"]["float_ms_ssim"]["mean"]
        ciede2000_score = scores["pooled_metrics"]["ciede2000"]["mean"]
        vmaf_score = scores["pooled_metrics"]["vmaf"]["mean"]

    return ssim_score, msssim_score, ciede2000_score, psnrhvs_score, vmaf_score

def get_butteraugli(png1, png2):
    cmd = "butteraugli_main %s %s" % (png1, png2)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: %s\n" % (cmd))
        sys.exit(proc.returncode)

    butteraugli_score = float(out.split(os.linesep)[0])
    return butteraugli_score

def get_dssim(png1, png2):
    cmd = "dssim %s %s" % (png1, png2)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: %s\n" % (cmd))
        sys.exit(proc.returncode)
    line = out.split(os.linesep)[0]
    dssim_score = float(re.search("^\d+\.?\d*", line).group(0))
    return dssim_score

def get_ssimulacra(png1, png2):
    cmd = "ssimulacra_main %s %s" % (png1, png2)
    proc = subprocess.Popen(
        split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
    )
    out, err = proc.communicate()
    if proc.returncode != 0:
        sys.stderr.write("Failed process: %s\n" % (cmd))
        sys.exit(proc.returncode)
    ssimulacra_score = float(out.split(os.linesep)[0])
    return ssimulacra_score

# Returns tuple containing:
#   (target_file_size, encode_time, decode_time)
def get_lossless_results(subset_name, origpng, format, format_recipe):

    origppm = os.path.splitext(origpng)[0] + ".ppm";

    target = (
        format.upper()
        + "_out/"
        + subset_name
        + "/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "-lossless"
    )
    create_dir(target)
    target_dec = path_for_file_in_tmp(target)

    target += "." + format_recipe["encode_extension"]
    cmd = string.Template(format_recipe["lossless_cmd"]).substitute(locals())
    wrapped = wrapper(run_silent, cmd)
    encode_time = Timer(wrapped).timeit(5) / 5

    target_dec += "." + format_recipe["decode_extension"]
    cmd = string.Template(format_recipe["decode_cmd"]).substitute(locals())
    wrapped = wrapper(run_silent, cmd)
    decode_time = Timer(wrapped).timeit(5) / 5

    target_file_size = os.path.getsize(target)

    try:
        os.remove(target_dec)
    except FileNotFoundError:
        pass

    return (target_file_size, encode_time, decode_time)

# Returns tuple containing:
#   (target_file_size, encode_time, decode_time, yssim_score, rgbssim_score,
#   psnrhvsm_score, msssim_score)
def get_lossy_results(
    subset_name, origpng, width, height, format, format_recipe, quality
):

    origy4m = path_for_file_in_tmp(origpng) + ".y4m"
    convertff_img(origpng, origy4m)

    target = (
        format.upper()
        + "_out/"
        + subset_name
        + "/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "-q"
        + str(quality)
    )
    create_dir(target)
    target_dec = path_for_file_in_tmp(target)

    target += "." + format_recipe["encode_extension"]
    cmd = string.Template(format_recipe["encode_cmd"]).substitute(locals())
    wrapped = wrapper(run_silent, cmd)
    encode_time = Timer(wrapped).timeit(1)

    target_json = target_dec + ".json"

    target_dec += "." + format_recipe["decode_extension"]
    cmd = string.Template(format_recipe["decode_cmd"]).substitute(locals())
    wrapped = wrapper(run_silent, cmd)
    decode_time = Timer(wrapped).timeit(1)

    if format_recipe["decode_extension"] != "png":
        target_png = path_for_file_in_tmp(target_dec) + ".png"
        convert_img(target_dec, target_png)
    else:
        target_png = target_dec

    # libavif bug?
    if not detect_alpha(origpng):
        remove_alpha(target_png, target_png)

    target_y4m = path_for_file_in_tmp(target_dec) + ".y4m"
    convertff_img(target_dec, target_y4m)

    ssim_score, msssim_score, ciede2000_score, psnrhvs_score, vmaf_score = get_score(
        origy4m, target_y4m, target_json
    )

    butteraugli_score = get_butteraugli(origpng, target_png)
    dssim_score = get_dssim(origpng, target_png)
    ssimulacra_score = get_ssimulacra(origpng, target_png)

    target_file_size = os.path.getsize(target)

    try:
        os.remove(origy4m)
        os.remove(target_dec)
        os.remove(target_json)
        os.remove(target_y4m)
    except FileNotFoundError:
        pass

    return (
        target_file_size,
        encode_time,
        decode_time,
        ssim_score,
        msssim_score,
        ciede2000_score,
        psnrhvs_score,
        vmaf_score,
        butteraugli_score,
        dssim_score,
        ssimulacra_score,
    )

def process_image(args):
    [format, format_recipe, subset_name, origpng] = args

    result_file = (
        "results/"
        + subset_name
        + "/"
        + format
        + "/lossy/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "."
        + format
        + ".out"
    )
    if os.path.isfile(result_file) and not os.stat(result_file).st_size == 0:
        return

    try:
        isfloat = (
            isinstance(format_recipe["quality_start"], float)
            or isinstance(format_recipe["quality_end"], float)
            or isinstance(format_recipe["quality_step"], float)
        )

        if isfloat:
            start = float(format_recipe["quality_start"])
            end = float(format_recipe["quality_end"])
            step = float(format_recipe["quality_step"])
        else:
            start = int(format_recipe["quality_start"])
            end = int(format_recipe["quality_end"])
            step = int(format_recipe["quality_step"])
    except ValueError:
        print("There was an error parsing the format recipe.")
        return

    if (
        not "encode_extension" in format_recipe
        or not "decode_extension" in format_recipe
        or not "encode_cmd" in format_recipe
        or not "lossless_cmd" in format_recipe
        or not "decode_cmd" in format_recipe
    ):
        print("There was an error parsing the format recipe.")
        return

    orig_file_size = os.path.getsize(origpng)
    width = get_img_width(origpng)
    height = get_img_height(origpng)
    pixels = width * height

    # Lossless
    print("Processing image {}, quality lossless".format(os.path.basename(origpng)))

    origppm = os.path.splitext(origpng)[0] + ".ppm";
    ppm_file_size = os.path.getsize(origppm)

    path = (
        "results/"
        + subset_name
        + "/"
        + format
        + "/lossless/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "."
        + format
        + ".out"
    )
    create_dir(path)
    file = open(path, "w")

    file.write(
        "file_name:orig_file_size:compressed_file_size:pixels:bpp:compression_ratio:encode_time:decode_time\n"
    )

    results = get_lossless_results(subset_name, origpng, format, format_recipe)
    bpp = results[0] * 8 / pixels
    compression_ratio = ppm_file_size / results[0]
    file.write(
        "%s:%d:%d:%d:%f:%f:%f:%f\n"
        % (
            os.path.splitext(os.path.basename(origpng))[0],
            ppm_file_size,
            results[0],
            pixels,
            bpp,
            compression_ratio,
            results[1],
            results[2],
        )
    )

    file.close()

    if "png" in format:
        return

    # Lossy
    path = (
        "results/"
        + subset_name
        + "/"
        + format
        + "/lossy/"
        + os.path.splitext(os.path.basename(origpng))[0]
        + "."
        + format
        + ".out"
    )
    create_dir(path)
    file = open(path, "w")

    file.write(
        "file_name:quality:orig_file_size:compressed_file_size:pixels:bpp:compression_ratio:encode_time:decode_time:ssim_score:msssim_score:ciede2000_score:psnrhvs_score:vmaf_score:butteraugli_score:dssim_score:ssimulacra_score\n"
    )
    if isfloat:
        quality_list = list(np.arange(start, end, step))
    else:
        quality_list = list(range(start, end, step))

    i = 0
    while i < len(quality_list):
        quality = quality_list[i]
        print(
            "Processing image {}, quality {}".format(os.path.basename(origpng), quality)
        )
        results = get_lossy_results(
            subset_name, origpng, width, height, format, format_recipe, quality
        )
        bpp = results[0] * 8 / pixels
        compression_ratio = orig_file_size / results[0]
        file.write(
            "%s:%f:%d:%d:%d:%f:%f:%f:%f:%f:%f:%f:%f:%f:%f:%f:%f\n"
            % (
                os.path.splitext(os.path.basename(origpng))[0],
                quality,
                orig_file_size,
                results[0],
                pixels,
                bpp,
                compression_ratio,
                results[1],
                results[2],
                results[3],
                results[4],
                results[5],
                results[6],
                results[7],
                results[8],
                results[9],
                results[10],
            )
        )
        i += 1

    file.close()

def main(argv):
    if sys.version_info[0] < 3 and sys.version_info[1] < 5:
        raise Exception("Python 3.5 or a more recent version is required.")

    data = {}
    try:
        with open("recipes.json") as json_file:
            data = json.load(json_file)
    except FileNotFoundError:
        raise Exception("Could not find recipes.json")

    supported_formats = list(data["recipes"].keys())

    if len(argv) != 4:
        print(
            "rd_collect.py: Generate compressed images from Y4Ms and calculate quality and speed metrics for a given format"
        )
        print("Arg 1: format to test {}".format(supported_formats))
        print("Arg 2: name of the subset to test (e.g. 'subset1')")
        print("Arg 3: path to the subset to test (e.g. 'subset1/')")
        return

    format = argv[1]
    subset_name = argv[2]
    if format not in supported_formats:
        print(
            "Image format not supported. Supported formats are: {}.".format(
                supported_formats
            )
        )
        return

    pool = Pool(processes=16)
    pool.map(
        process_image,
        [
            (format, data["recipes"][format], subset_name, origpng)
            for origpng in glob.glob(argv[3] + "/*.png")
        ],
    )

if __name__ == "__main__":
    main(sys.argv)

max_stars_repo_name: eclipseo/image-comparison-sources

max_issues_repo_name: eclipseo/image-comparison-sources

max_forks_repo_name: eclipseo/image-comparison-sources


text:
# Copyright 2022 The TensorFlow 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.

"""Tests for panoptic_deeplab_heads.py."""

from absl.testing import parameterized
import numpy as np
import tensorflow as tf

from official.vision.beta.projects.panoptic_maskrcnn.modeling.heads import panoptic_deeplab_heads

class PanopticDeeplabHeadsTest(parameterized.TestCase, tf.test.TestCase):

  @parameterized.parameters(
      (2, (2,), (48,)),
      (3, (2,), (48,)),
      (2, (2,), (48,)),
      (2, (2,), (48,)),
      (3, (2,), (48,)),
      (3, (2,), (48,)),
      (4, (4, 3), (64, 32)),
      (4, (3, 2), (64, 32)))
  def test_forward(self, level, low_level, low_level_num_filters):
    backbone_features = {
        '3': np.random.rand(2, 128, 128, 16),
        '4': np.random.rand(2, 64, 64, 16),
        '5': np.random.rand(2, 32, 32, 16),
    }
    decoder_features = {
        '3': np.random.rand(2, 128, 128, 64),
        '4': np.random.rand(2, 64, 64, 64),
        '5': np.random.rand(2, 32, 32, 64),
        '6': np.random.rand(2, 16, 16, 64),
    }

    backbone_features['2'] = np.random.rand(2, 256, 256, 16)
    decoder_features['2'] = np.random.rand(2, 256, 256, 64)
    num_classes = 10
    semantic_head = panoptic_deeplab_heads.SemanticHead(
        num_classes=num_classes,
        level=level,
        low_level=low_level,
        low_level_num_filters=low_level_num_filters)

    instance_head = panoptic_deeplab_heads.InstanceHead(
        level=level,
        low_level=low_level,
        low_level_num_filters=low_level_num_filters)

    semantic_outputs = semantic_head((backbone_features, decoder_features))
    instance_outputs = instance_head((backbone_features, decoder_features))

    if str(level) in decoder_features:
      h, w = decoder_features[str(low_level[-1])].shape[1:3]
      self.assertAllEqual(
          semantic_outputs.numpy().shape,
          [2, h, w, num_classes])
      self.assertAllEqual(
          instance_outputs['instance_centers_heatmap'].numpy().shape,
          [2, h, w, 1])
      self.assertAllEqual(
          instance_outputs['instance_centers_offset'].numpy().shape,
          [2, h, w, 2])

  def test_serialize_deserialize(self):
    semantic_head = panoptic_deeplab_heads.SemanticHead(num_classes=2, level=3)
    instance_head = panoptic_deeplab_heads.InstanceHead(level=3)

    semantic_head_config = semantic_head.get_config()
    instance_head_config = instance_head.get_config()

    new_semantic_head = panoptic_deeplab_heads.SemanticHead.from_config(
        semantic_head_config)
    new_instance_head = panoptic_deeplab_heads.InstanceHead.from_config(
        instance_head_config)

    self.assertAllEqual(semantic_head.get_config(),
                        new_semantic_head.get_config())
    self.assertAllEqual(instance_head.get_config(),
                        new_instance_head.get_config())

if __name__ == '__main__':
  tf.test.main()

max_stars_repo_name: Lufeifeina/models

max_issues_repo_name: Lufeifeina/models

max_forks_repo_name: Lufeifeina/models


text:
"""
Plot full psychometric functions as a function of choice history,
and separately for 20/80 and 80/20 blocks
"""

import pandas as pd
import numpy as np
import sys, os, time
import matplotlib.pyplot as plt
import seaborn as sns
import datajoint as dj
from IPython import embed as shell # for debugging
from scipy.special import erf # for psychometric functions
from matplotlib import dates

## INITIALIZE A FEW THINGS
sns.set_style("darkgrid")
sns.set_context(context="poster")

# import wrappers etc
from ibl_pipeline import reference, subject, action, acquisition, data, behavior
from ibl_pipeline.utils import psychofit as psy
from dj_tools import *
figpath  = os.path.join(os.path.expanduser('~'), 'Data/Figures_IBL')

# ================================= #
##  GET DATA
# ================================= #

b = (behavior.TrialSet) \
	* (acquisition.Session.proj(session_date='DATE(session_start_time)') & 'session_start_time > "2019-03-01"') \
	* (subject.Subject() & 'subject_birth_date between "2018-09-01" and "2019-02-01"') * subject.SubjectLab() \
	* action.Weighing.proj(weighing_date='DATE(weighing_time)')
bdat = pd.DataFrame(b.fetch(order_by='subject_nickname, session_start_time'))

print(bdat['subject_nickname'].unique())
bdat['date_march'] = pd.to_timedelta(pd.to_datetime(bdat.session_date)).dt.total_seconds().astype(int)

g = sns.lmplot(x="date_march", y="n_trials", hue="sex",
	col="lab_name", units="subject_nickname",
	col_wrap=4, lowess=True, data=bdat)

# fig = sns.FacetGrid(bdat, hue="sex", col="lab_name", col_wrap=4)
# fig.map(sns.lineplot, "session_date", "n_trials", estimator=None, units="subject_nickname").add_legend()
g.set_xticklabels("", rotation=45)
g.savefig(os.path.join(figpath, "trialcounts_malevsfemale.pdf"))
g.savefig(os.path.join(figpath, "trialcounts_malevsfemale.png"))
plt.close("all")

# ================================= #
# WEIGHT VS TRIALCOUNTS
# ================================= #

weight_info = action.Weighing.proj('weight', session_date='DATE(weighing_time)')
weight_matched = (dj.U('subject_uuid', 'session_date') & weight_info) * weight_info
b = (behavior.TrialSet) \
    * (acquisition.Session.proj(session_date='DATE(session_start_time)') & 'session_start_time > "2019-03-01"') \
    * (subject.Subject() & 'subject_birth_date between "2018-09-01" and "2019-02-01"') * subject.SubjectLab() \
    * weight_matched
bdat = pd.DataFrame(b.fetch(order_by='subject_nickname, session_start_time'))

g = sns.lmplot(x="weight", y="n_trials", hue="sex",
	col="lab_name", units="subject_nickname",
	col_wrap=4, data=bdat, sharex=False,sharey=False, truncate=True, ci=None)
g.savefig(os.path.join(figpath, "trialcounts_weight_malevsfemale.png"))

max_stars_repo_name: ineslaranjeira/analysis

max_issues_repo_name: ineslaranjeira/analysis

max_forks_repo_name: ineslaranjeira/analysis

reason: import numpy,from scipy


text:
# -*- coding: utf-8 -*-
"""
Created on Sat Mar 13 10:15:25 2021

@author: arc
"""
import numpy as np
import pandas as pd
import joblib, pickle
from sklearn.linear_model import LogisticRegression
from os.path import dirname, join, abspath

class heartAttackModel():
    def __init__(self,model_file='model_heart_LG.sav'):
        model_file = join(dirname(abspath(__file__)), model_file)
        with open(model_file,'rb') as model_file:
            self.classifier  = pickle.load(model_file)
            self.data = None
    # reading is an array to be passed from GUI
    def predict_outcome(self,reading):
        self.data = reading
        if self.data is not None:
            pred = self.classifier.predict(self.data)
            return pred
        else:
            return None
    def predict_proba(self,reading):
        self.data = reading
        if self.data is not None:
            prob = self.classifier.predict_proba(self.data)[:,1]
            return prob
        else:
            return None
    def predict(self, dataframe):
        data_np = np.array(dataframe)
        data_np = np.reshape(data_np,(-1,13))
        print(dataframe)
        outcome = self.predict_outcome(data_np)
        probability = self.predict_proba(data_np)
        arr = list()
        arr.append(outcome[0])
        arr.append(probability[0])
        return arr

max_stars_repo_name: arc-arnob/Reddit-Clone

max_issues_repo_name: arc-arnob/Reddit-Clone

max_forks_repo_name: arc-arnob/Reddit-Clone


text:
import numpy as np
import sys
sys.path.append(os.path.join('..')) # Include the home dir in our path to be able to import modules in notebooks that are not child of this dir?
import cv2
from utils.imutils import jimshow

def translate(img, x = 0, y = 0):
    height, width = img.shape[:2]

    m = np.float64([[1, 0, x],
                    [0, 1, y]])
    return cv2.warpAffine(img, m, (width, height)) # Translates img

def main():

if __name__ == "__main__":
    main()

max_stars_repo_name: NormalReedus/cds-visual-analytics

max_issues_repo_name: NormalReedus/cds-visual-analytics

max_forks_repo_name: NormalReedus/cds-visual-analytics