text:
#!/usr/bin/env python
import pickle

from unittest import TestCase, main

from cogent3.core import sequence
from cogent3.core.moltype import (
    DNA,
    PROTEIN,
    RNA,
    STANDARD_CODON,
    AlphabetError,
    AlphabetGroup,
    CodonAlphabet,
    CoreObjectGroup,
    DnaStandardPairs,
    IUPAC_DNA_ambiguities,
    IUPAC_DNA_ambiguities_complements,
    IUPAC_DNA_chars,
    IUPAC_RNA_ambiguities,
    IUPAC_RNA_ambiguities_complements,
    IUPAC_RNA_chars,
    MolType,
    RnaStandardPairs,
    available_moltypes,
    get_moltype,
    make_matches,
    make_pairs,
)
from cogent3.data.molecular_weight import DnaMW, RnaMW

__author__ = "Gavin Huttley, Peter Maxwell, and Rob Knight"
__copyright__ = "Copyright 2007-2020, The Cogent Project"
__credits__ = ["Rob Knight", "Gavin Huttley", "Peter Maxwell"]
__license__ = "BSD-3"
__version__ = "2020.12.21a"
__maintainer__ = "Gavin Huttley"
__email__ = "gavin.huttley@anu.edu.au"
__status__ = "Production"

# ind some of the standard alphabets to reduce typing
from numpy.testing import assert_allclose

RnaBases = RNA.alphabets.base
DnaBases = DNA.alphabets.base
AminoAcids = PROTEIN.alphabets.base

# the following classes are to preserve compatibility for older test code
# that assumes mixed-case is OK.
RnaMolType = MolType(
    seq_constructor=sequence.RnaSequence,
    motifset=IUPAC_RNA_chars,
    ambiguities=IUPAC_RNA_ambiguities,
    label="rna_with_lowercase",
    mw_calculator=RnaMW,
    complements=IUPAC_RNA_ambiguities_complements,
    pairs=RnaStandardPairs,
    add_lower=True,
    preserve_existing_moltypes=True,
    make_alphabet_group=True,
)
DnaMolType = MolType(
    seq_constructor=sequence.DnaSequence,
    motifset=IUPAC_DNA_chars,
    ambiguities=IUPAC_DNA_ambiguities,
    label="dna_with_lowercase",
    mw_calculator=DnaMW,
    complements=IUPAC_DNA_ambiguities_complements,
    pairs=DnaStandardPairs,
    add_lower=True,
    preserve_existing_moltypes=True,
    make_alphabet_group=True,
)
ProteinMolType = PROTEIN

class make_matches_tests(TestCase):
    """Tests of the make_matches top-level function"""

    def test_init_empty(self):
        """make_matches should init ok with no parameters"""
        self.assertEqual(make_matches(), {})

    def test_init_monomers(self):
        """make_matches with only monomers should produce {(i,i):True}"""
        m = make_matches("")
        self.assertEqual(m, {})
        m = make_matches("qaz")
        self.assertEqual(m, {("q", "q"): True, ("a", "a"): True, ("z", "z"): True})

    def test_init_gaps(self):
        """make_matches with only gaps should match all gaps to each other"""
        m = make_matches("", "~!")
        self.assertEqual(
            m, {("~", "~"): True, ("!", "!"): True, ("!", "~"): True, ("~", "!"): True}
        )

    def test_init_degen(self):
        """make_matches with only degen should work as expected"""
        m = make_matches(None, None, {"x": "ab", "y": "bc", "z": "cd", "n": "bcd"})
        self.assertEqual(
            m,
            {
                ("x", "x"): False,
                ("x", "y"): False,
                ("x", "n"): False,
                ("y", "x"): False,
                ("y", "y"): False,
                ("y", "z"): False,
                ("y", "n"): False,
                ("z", "y"): False,
                ("z", "z"): False,
                ("z", "n"): False,
                ("n", "x"): False,
                ("n", "y"): False,
                ("n", "z"): False,
                ("n", "n"): False,
            },
        )
        self.assertNotIn(("x", "z"), m)

    def test_init_all(self):
        """make_matches with everything should produce correct dict"""
        m = make_matches(
            "ABC", ("-", "~"), {"X": "AB", "Y": ("B", "C"), "N": list("ABC")}
        )
        exp = {
            ("-", "-"): True,
            ("~", "~"): True,
            ("-", "~"): True,
            ("~", "-"): True,
            ("A", "A"): True,
            ("B", "B"): True,
            ("C", "C"): True,
            ("A", "X"): False,
            ("X", "A"): False,
            ("B", "X"): False,
            ("X", "B"): False,
            ("B", "Y"): False,
            ("Y", "B"): False,
            ("C", "Y"): False,
            ("Y", "C"): False,
            ("A", "N"): False,
            ("N", "A"): False,
            ("B", "N"): False,
            ("N", "B"): False,
            ("C", "N"): False,
            ("N", "C"): False,
            ("X", "X"): False,
            ("Y", "Y"): False,
            ("N", "N"): False,
            ("X", "Y"): False,
            ("Y", "X"): False,
            ("X", "N"): False,
            ("N", "X"): False,
            ("Y", "N"): False,
            ("N", "Y"): False,
        }
        self.assertEqual(m, exp)

class make_pairs_tests(TestCase):
    """Tests of the top-level make_pairs factory function."""

    def setUp(self):
        """Define some standard pairs and other data"""
        self.pairs = {("U", "A"): True, ("A", "U"): True, ("G", "U"): False}

    def test_init_empty(self):
        """make_pairs should init ok with no parameters"""
        self.assertEqual(make_pairs(), {})

    def test_init_pairs(self):
        """make_pairs with just pairs should equal the original"""
        self.assertEqual(make_pairs(self.pairs), self.pairs)
        self.assertIsNot(make_pairs(self.pairs), self.pairs)

    def test_init_monomers(self):
        """make_pairs with pairs and monomers should equal just the pairs"""
        self.assertEqual(make_pairs(self.pairs, "ABCDEFG"), self.pairs)
        self.assertIsNot(make_pairs(self.pairs, "ABCDEFG"), self.pairs)

    def test_init_gaps(self):
        """make_pairs should add all combinations of gaps as weak pairs"""
        p = make_pairs(self.pairs, None, "-~")
        self.assertNotEqual(p, self.pairs)
        self.pairs.update(
            {("~", "~"): False, ("-", "~"): False, ("-", "-"): False, ("~", "-"): False}
        )
        self.assertEqual(p, self.pairs)

    def test_init_degen(self):
        """make_pairs should add in degenerate combinations as weak pairs"""
        p = make_pairs(self.pairs, "AUG", "-", {"R": "AG", "Y": "CU", "W": "AU"})
        self.assertNotEqual(p, self.pairs)
        self.pairs.update(
            {
                ("-", "-"): False,
                ("A", "Y"): False,
                ("Y", "A"): False,
                ("A", "W"): False,
                ("W", "A"): False,
                ("U", "R"): False,
                ("R", "U"): False,
                ("U", "W"): False,
                ("W", "U"): False,
                ("G", "Y"): False,
                ("G", "W"): False,
                ("R", "Y"): False,
                ("R", "W"): False,
                ("Y", "R"): False,
                ("Y", "W"): False,
                ("W", "R"): False,
                ("W", "Y"): False,
                ("W", "W"): False,
            }
        )
        self.assertEqual(p, self.pairs)

class CoreObjectGroupTests(TestCase):
    """Tests of the CoreObjectGroup class."""

    def test_init(self):
        """CoreObjectGroup should init with basic list of objects."""

        class o(object):
            def __init__(self, s):
                self.s = s

        base = o("base")
        c = CoreObjectGroup(base)
        self.assertIs(c.base, base)
        self.assertIs(c.degen, None)
        self.assertIs(c.base.degen, None)

        base, degen, gap, degengap = list(map(o, ["base", "degen", "gap", "degengap"]))
        c = CoreObjectGroup(base, degen, gap, degengap)
        self.assertIs(c.base, base)
        self.assertIs(c.base.degen, degen)
        self.assertIs(c.degen.gapped, degengap)

class AlphabetGroupTests(TestCase):
    """Tests of the AlphabetGroup class."""

    def test_init(self):
        """AlphabetGroup should initialize successfully"""
        chars = "AB"
        degens = {"C": "AB"}
        g = AlphabetGroup(chars, degens)
        self.assertEqual("".join(g.base), "AB")
        self.assertEqual("".join(g.degen), "ABC")
        self.assertEqual("".join(g.gapped), "AB-")
        self.assertEqual("".join(g.degen_gapped), "AB-C?")

class MolTypeTests(TestCase):
    """Tests of the MolType class. Should support same API as old Alphabet."""

    def test_to_regex(self):
        """returns a valid regex"""
        seq = "ACYGR"
        regular_expression = DNA.to_regex(seq=seq)
        self.assertEqual(regular_expression, "AC[CT]G[AG]")
        # raises an exception if a string is already a regex, or invalid
        with self.assertRaises(ValueError):
            DNA.to_regex("(?:GAT|GAC)(?:GGT|GGC|GGA|GGG)(?:GAT|GAC)(?:CAA|CAG)")

    def test_pickling(self):
        pkl = pickle.dumps(DNA)
        got = pickle.loads(pkl)
        self.assertIsInstance(got, type(DNA))
        self.assertEqual(DNA.ambiguities, got.ambiguities)

    def test_get_moltype(self):
        """correctly return a moltype by name"""
        for label in ("dna", "rna", "protein", "protein_with_stop"):
            mt = get_moltype(label)
            self.assertEqual(mt.label, label)
            mt = get_moltype(label.upper())
            self.assertEqual(mt.label, label)

        mt = get_moltype(DNA)
        self.assertEqual(mt.label, "dna")
        with self.assertRaises(ValueError):
            _ = get_moltype("blah")

    def test_available_moltypes(self):
        """available_moltypes returns table with correct columns and rows"""
        available = available_moltypes()
        self.assertEqual(available.shape, (7, 3))
        self.assertEqual(available[1, "Number of states"], 4)
        self.assertEqual(available["dna", "Number of states"], 4)
        txt = repr(available)
        self.assertIn("'dna'", txt)

    def test_init_minimal(self):
        """MolType should init OK with just monomers"""
        a = MolType("Abc")
        self.assertIn("A", a.alphabet)
        self.assertNotIn("a", a.alphabet)  # case-sensitive
        self.assertIn("b", a.alphabet)
        self.assertNotIn("B", a.alphabet)
        self.assertNotIn("x", a.alphabet)

    def test_init_everything(self):
        """MolType should init OK with all parameters set"""
        k = dict.fromkeys
        a = MolType(
            k("Abc"),
            ambiguities={"d": "bc"},
            gaps=k("~"),
            complements={"b": "c", "c": "b"},
            pairs={},
            add_lower=False,
        )
        for i in "Abcd~":
            self.assertIn(i, a)
        self.assertEqual(a.complement("b"), "c")
        self.assertEqual(a.complement("AbcAA"), "AcbAA")
        self.assertEqual(a.first_degenerate("AbcdA"), 3)
        self.assertEqual(a.first_gap("a~c"), 1)
        self.assertEqual(a.first_invalid("Abcx"), 3)

    def test_strip_degenerate(self):
        """MolType strip_degenerate should remove any degenerate bases"""
        s = RnaMolType.strip_degenerate
        self.assertEqual(s("UCAG-"), "UCAG-")
        self.assertEqual(s("NRYSW"), "")
        self.assertEqual(s("USNG"), "UG")

    def test_strip_bad(self):
        """MolType strip_bad should remove any non-base, non-gap chars"""
        s = RnaMolType.strip_bad
        self.assertEqual(s("UCAGwsnyrHBND-D"), "UCAGwsnyrHBND-D")
        self.assertEqual(s("@#^*($@!#&()!@QZX"), "")
        self.assertEqual(s("aaa ggg ---!ccc"), "aaaggg---ccc")

    def test_strip_bad_and_gaps(self):
        """MolType strip_bad_and_gaps should remove gaps and bad chars"""
        s = RnaMolType.strip_bad_and_gaps
        self.assertEqual(s("UCAGwsnyrHBND-D"), "UCAGwsnyrHBNDD")
        self.assertEqual(s("@#^*($@!#&()!@QZX"), "")
        self.assertEqual(s("aaa ggg ---!ccc"), "aaagggccc")

    def test_complement(self):
        """MolType complement should correctly complement sequence"""
        self.assertEqual(RnaMolType.complement("UauCG-NR"), "AuaGC-NY")
        self.assertEqual(DnaMolType.complement("TatCG-NR"), "AtaGC-NY")
        self.assertEqual(RnaMolType.complement(""), "")
        self.assertRaises(TypeError, ProteinMolType.complement, "ACD")
        # if it wasn't a string, result should be a list
        self.assertEqual(RnaMolType.complement(list("UauCG-NR")), list("AuaGC-NY"))
        self.assertEqual(RnaMolType.complement(("a", "c")), ("u", "g"))
        # constructor should fail for a dict
        self.assertRaises(ValueError, RnaMolType.complement, {"a": "c"})

    def test_rc(self):
        """MolType rc should correctly reverse-complement sequence"""
        self.assertEqual(RnaMolType.rc("U"), "A")
        self.assertEqual(RnaMolType.rc(""), "")
        self.assertEqual(RnaMolType.rc("R"), "Y")
        self.assertEqual(RnaMolType.rc("UauCG-NR"), "YN-CGauA")
        self.assertEqual(DnaMolType.rc("TatCG-NR"), "YN-CGatA")
        self.assertRaises(TypeError, ProteinMolType.rc, "ACD")
        # if it wasn't a string, result should be a list
        self.assertEqual(RnaMolType.rc(list("UauCG-NR")), list("YN-CGauA"))
        self.assertEqual(RnaMolType.rc(("a", "c")), ("g", "u"))
        # constructor should fail for a dict
        self.assertRaises(ValueError, RnaMolType.rc, {"a": "c"})

    def test_contains(self):
        """MolType contains should return correct result"""
        for i in "UCAGWSMKRYBDHVN-" + "UCAGWSMKRYBDHVN-".lower():
            self.assertIn(i, RnaMolType)
        for i in "x!@#$%^&ZzQq":
            self.assertNotIn(i, RnaMolType)

        a = MolType(dict.fromkeys("ABC"), add_lower=True)
        for i in "abcABC":
            self.assertIn(i, a)
        self.assertNotIn("x", a)
        b = MolType(dict.fromkeys("ABC"), add_lower=False)
        for i in "ABC":
            self.assertIn(i, b)
        for i in "abc":
            self.assertNotIn(i, b)

    def test_iter(self):
        """MolType iter should iterate over monomer order"""
        self.assertEqual(list(RnaMolType), ["U", "C", "A", "G", "u", "c", "a", "g"])
        a = MolType("ZXCV")
        self.assertEqual(list(a), ["Z", "X", "C", "V"])

    def test_is_gapped(self):
        """MolType is_gapped should return True if gaps in seq"""
        g = RnaMolType.is_gapped
        self.assertFalse(g(""))
        self.assertFalse(g("ACGUCAGUACGUCAGNRCGAUcaguaguacYRNRYRN"))
        self.assertTrue(g("-"))
        self.assertTrue(g("--"))
        self.assertTrue(g("CAGUCGUACGUCAGUACGUacucauacgac-caguACUG"))
        self.assertTrue(g("CA--CGUAUGCA-----g"))
        self.assertTrue(g("CAGU-"))

    def test_is_gap(self):
        """MolType is_gap should return True if char is a gap"""
        g = RnaMolType.is_gap
        # True for the empty string
        self.assertFalse(g(""))
        # True for all the standard and degenerate symbols
        s = "ACGUCAGUACGUCAGNRCGAUcaguaguacYRNRYRN"
        self.assertFalse(g(s))
        for i in s:
            self.assertFalse(g(i))
        # should be true for a single gap
        self.assertTrue(g("-"))
        # note that it _shouldn't_ be true for a run of gaps: use a.is_gapped()
        self.assertFalse(g("--"))

    def test_is_degenerate(self):
        """MolType is_degenerate should return True if degen symbol in seq"""
        d = RnaMolType.is_degenerate
        self.assertFalse(d(""))
        self.assertFalse(d("UACGCUACAUGuacgucaguGCUAGCUA---ACGUCAG"))
        self.assertTrue(d("N"))
        self.assertTrue(d("R"))
        self.assertTrue(d("y"))
        self.assertTrue(d("GCAUguagcucgUCAGUCAGUACgUgcasCUAG"))
        self.assertTrue(d("ACGYAUGCUGYEWEWNFMNfuwbybcwuybcjwbeiwfub"))

    def test_is_valid(self):
        """MolType is_valid should return True if any unknown symbol in seq"""
        v = RnaMolType.is_valid
        self.assertFalse(v(3))
        self.assertFalse(v(None))
        self.assertTrue(v("ACGUGCAUGUCAYCAYGUACGcaugacyugc----RYNCYRNC"))
        self.assertTrue(v(""))
        self.assertTrue(v("a"))
        self.assertFalse(v("ACIUBHFWUIXZKLNJUCIHBICNSOWMOINJ"))
        self.assertFalse(v("CAGUCAGUCACA---GACCAUG-_--cgau"))

    def test_is_strict(self):
        """MolType is_strict should return True if all symbols in Monomers"""
        s = RnaMolType.is_strict
        self.assertFalse(s(3))
        self.assertFalse(s(None))
        self.assertTrue(s(""))
        self.assertTrue(s("A"))
        self.assertTrue(s("UAGCACUgcaugcauGCAUGACuacguACAUG"))
        self.assertFalse(s("CAGUCGAUCA-cgaucagUCGAUGAC"))
        self.assertFalse(s("ACGUGCAUXCAGUCAG"))

    def test_first_gap(self):
        """MolType first_gap should return index of first gap symbol, or None"""
        g = RnaMolType.first_gap
        self.assertEqual(g(""), None)
        self.assertEqual(g("a"), None)
        self.assertEqual(g("uhacucHuhacUIUIhacan"), None)
        self.assertEqual(g("-abc"), 0)
        self.assertEqual(g("b-ac"), 1)
        self.assertEqual(g("abcd-"), 4)

    def test_first_degenerate(self):
        """MolType first_degenerate should return index of first degen symbol"""
        d = RnaMolType.first_degenerate
        self.assertEqual(d(""), None)
        self.assertEqual(d("a"), None)
        self.assertEqual(d("UCGACA--CU-gacucaguacgua"), None)
        self.assertEqual(d("nCAGU"), 0)
        self.assertEqual(d("CUGguagvAUG"), 7)
        self.assertEqual(d("ACUGCUAacgud"), 11)

    def test_first_invalid(self):
        """MolType first_invalid should return index of first invalid symbol"""
        i = RnaMolType.first_invalid
        self.assertEqual(i(""), None)
        self.assertEqual(i("A"), None)
        self.assertEqual(i("ACGUNVBuacg-wskmWSMKYRryNnN--"), None)
        self.assertEqual(i("x"), 0)
        self.assertEqual(i("rx"), 1)
        self.assertEqual(i("CAGUNacgunRYWSwx"), 15)

    def test_first_non_strict(self):
        """MolType first_non_strict should return index of first non-strict symbol"""
        s = RnaMolType.first_non_strict
        self.assertEqual(s(""), None)
        self.assertEqual(s("A"), None)
        self.assertEqual(s("ACGUACGUcgaucagu"), None)
        self.assertEqual(s("N"), 0)
        self.assertEqual(s("-"), 0)
        self.assertEqual(s("x"), 0)
        self.assertEqual(s("ACGUcgAUGUGCAUcaguX"), 18)
        self.assertEqual(s("ACGUcgAUGUGCAUcaguX-38243829"), 18)

    def test_disambiguate(self):
        """MolType disambiguate should remove degenerate bases"""
        d = RnaMolType.disambiguate
        self.assertEqual(d(""), "")
        self.assertEqual(d("AGCUGAUGUA--CAGU"), "AGCUGAUGUA--CAGU")
        self.assertEqual(d("AUn-yrs-wkmCGwmrNMWRKY", "strip"), "AU--CG")
        self.assertEqual(d(tuple("AUn-yrs-wkmCGwmrNMWRKY"), "strip"), tuple("AU--CG"))
        s = "AUn-yrs-wkmCGwmrNMWRKY"
        t = d(s, "random")
        u = d(s, "random")
        for i, j in zip(s, t):
            if i in RnaMolType.degenerates:
                self.assertIn(j, RnaMolType.degenerates[i])
            else:
                self.assertEqual(i, j)
        self.assertNotEqual(t, u)
        self.assertEqual(d(tuple("UCAG"), "random"), tuple("UCAG"))
        self.assertEqual(len(s), len(t))
        self.assertIs(RnaMolType.first_degenerate(t), None)
        # should raise exception on unknown disambiguation method
        self.assertRaises(NotImplementedError, d, s, "xyz")

    def test_degap(self):
        """MolType degap should remove all gaps from sequence"""
        g = RnaMolType.degap
        self.assertEqual(g(""), "")
        self.assertEqual(g("GUCAGUCgcaugcnvuincdks"), "GUCAGUCgcaugcnvuincdks")
        self.assertEqual(g("----------------"), "")
        self.assertEqual(g("gcuauacg-"), "gcuauacg")
        self.assertEqual(g("-CUAGUCA"), "CUAGUCA")
        self.assertEqual(g("---a---c---u----g---"), "acug")
        self.assertEqual(g(tuple("---a---c---u----g---")), tuple("acug"))

    def test_gap_indices(self):
        """MolType gap_indices should return correct gap positions"""
        g = RnaMolType.gap_indices
        self.assertEqual(g(""), [])
        self.assertEqual(g("ACUGUCAGUACGHFSDKJCUICDNINS"), [])
        self.assertEqual(g("GUACGUIACAKJDC-SDFHJDSFK"), [14])
        self.assertEqual(g("-DSHFUHDSF"), [0])
        self.assertEqual(g("UACHASJAIDS-"), [11])
        self.assertEqual(
            g("---CGAUgCAU---ACGHc---ACGUCAGU---"),
            [0, 1, 2, 11, 12, 13, 19, 20, 21, 30, 31, 32],
        )
        a = MolType({"A": 1}, gaps=dict.fromkeys("!@#$%"))
        g = a.gap_indices
        self.assertEqual(g(""), [])
        self.assertEqual(g("!!!"), [0, 1, 2])
        self.assertEqual(g("!@#$!@#$!@#$"), list(range(12)))
        self.assertEqual(g("cguua!cgcuagua@cguasguadc#"), [5, 14, 25])

    def test_gap_vector(self):
        """MolType gap_vector should return correct gap positions"""
        g = RnaMolType.gap_vector
        self.assertEqual(g(""), [])
        self.assertEqual(g("ACUGUCAGUACGHFSDKJCUICDNINS"), [False] * 27)
        self.assertEqual(
            g("GUACGUIACAKJDC-SDFHJDSFK"),
            list(map(bool, list(map(int, "000000000000001000000000")))),
        )
        self.assertEqual(g("-DSHFUHDSF"), list(map(bool, list(map(int, "1000000000")))))
        self.assertEqual(
            g("UACHASJAIDS-"), list(map(bool, list(map(int, "000000000001"))))
        )
        self.assertEqual(
            g("---CGAUgCAU---ACGHc---ACGUCAGU---"),
            list(map(bool, list(map(int, "111000000001110000011100000000111")))),
        )
        a = MolType({"A": 1}, gaps=dict.fromkeys("!@#$%"))
        g = a.gap_vector
        self.assertEqual(g(""), [])
        self.assertEqual(g("!!!"), list(map(bool, [1, 1, 1])))
        self.assertEqual(g("!@#$!@#$!@#$"), [True] * 12)
        self.assertEqual(
            g("cguua!cgcuagua@cguasguadc#"),
            list(map(bool, list(map(int, "00000100000000100000000001")))),
        )

    def test_gap_maps(self):
        """MolType gap_maps should return dicts mapping gapped/ungapped pos"""
        empty = ""
        no_gaps = "aaa"
        all_gaps = "---"
        start_gaps = "--abc"
        end_gaps = "ab---"
        mid_gaps = "--a--b-cd---"
        gm = RnaMolType.gap_maps
        self.assertEqual(gm(empty), ({}, {}))
        self.assertEqual(gm(no_gaps), ({0: 0, 1: 1, 2: 2}, {0: 0, 1: 1, 2: 2}))
        self.assertEqual(gm(all_gaps), ({}, {}))
        self.assertEqual(gm(start_gaps), ({0: 2, 1: 3, 2: 4}, {2: 0, 3: 1, 4: 2}))
        self.assertEqual(gm(end_gaps), ({0: 0, 1: 1}, {0: 0, 1: 1}))
        self.assertEqual(
            gm(mid_gaps), ({0: 2, 1: 5, 2: 7, 3: 8}, {2: 0, 5: 1, 7: 2, 8: 3})
        )

    def test_count_gaps(self):
        """MolType count_gaps should return correct gap count"""
        c = RnaMolType.count_gaps
        self.assertEqual(c(""), 0)
        self.assertEqual(c("ACUGUCAGUACGHFSDKJCUICDNINS"), 0)
        self.assertEqual(c("GUACGUIACAKJDC-SDFHJDSFK"), 1)
        self.assertEqual(c("-DSHFUHDSF"), 1)
        self.assertEqual(c("UACHASJAIDS-"), 1)
        self.assertEqual(c("---CGAUgCAU---ACGHc---ACGUCAGU---"), 12)
        a = MolType({"A": 1}, gaps=dict.fromkeys("!@#$%"))
        c = a.count_gaps
        self.assertEqual(c(""), 0)
        self.assertEqual(c("!!!"), 3)
        self.assertEqual(c("!@#$!@#$!@#$"), 12)
        self.assertEqual(c("cguua!cgcuagua@cguasguadc#"), 3)

    def test_count_degenerate(self):
        """MolType count_degenerate should return correct degen base count"""
        d = RnaMolType.count_degenerate
        self.assertEqual(d(""), 0)
        self.assertEqual(d("GACUGCAUGCAUCGUACGUCAGUACCGA"), 0)
        self.assertEqual(d("N"), 1)
        self.assertEqual(d("NRY"), 3)
        self.assertEqual(d("ACGUAVCUAGCAUNUCAGUCAGyUACGUCAGS"), 4)

    def test_possibilites(self):
        """MolType possibilities should return correct # possible sequences"""
        p = RnaMolType.possibilities
        self.assertEqual(p(""), 1)
        self.assertEqual(p("ACGUgcaucagUCGuGCAU"), 1)
        self.assertEqual(p("N"), 4)
        self.assertEqual(p("R"), 2)
        self.assertEqual(p("H"), 3)
        self.assertEqual(p("nRh"), 24)
        self.assertEqual(p("AUGCnGUCAg-aurGauc--gauhcgauacgws"), 96)

    def test_MW(self):
        """MolType MW should return correct molecular weight"""
        r = RnaMolType.mw
        p = ProteinMolType.mw
        self.assertEqual(p(""), 0)
        self.assertEqual(r(""), 0)
        assert_allclose(p("A"), 89.09)
        assert_allclose(r("A"), 375.17)
        assert_allclose(p("AAA"), 231.27)
        assert_allclose(r("AAA"), 1001.59)
        assert_allclose(r("AAACCCA"), 2182.37)

    def test_can_match(self):
        """MolType can_match should return True if all positions can match"""
        m = RnaMolType.can_match
        self.assertTrue(m("", ""))
        self.assertTrue(m("UCAG", "UCAG"))
        self.assertFalse(m("UCAG", "ucag"))
        self.assertTrue(m("UCAG", "NNNN"))
        self.assertTrue(m("NNNN", "UCAG"))
        self.assertTrue(m("NNNN", "NNNN"))
        self.assertFalse(m("N", "x"))
        self.assertFalse(m("N", "-"))
        self.assertTrue(m("UCAG", "YYRR"))
        self.assertTrue(m("UCAG", "KMWS"))

    def test_can_mismatch(self):
        """MolType can_mismatch should return True on any possible mismatch"""
        m = RnaMolType.can_mismatch
        self.assertFalse(m("", ""))
        self.assertTrue(m("N", "N"))
        self.assertTrue(m("R", "R"))
        self.assertTrue(m("N", "r"))
        self.assertTrue(m("CGUACGCAN", "CGUACGCAN"))
        self.assertTrue(m("U", "C"))
        self.assertTrue(m("UUU", "UUC"))
        self.assertTrue(m("UUU", "UUY"))
        self.assertFalse(m("UUU", "UUU"))
        self.assertFalse(m("UCAG", "UCAG"))
        self.assertFalse(m("U--", "U--"))

    def test_must_match(self):
        """MolType must_match should return True when no possible mismatches"""
        m = RnaMolType.must_match
        self.assertTrue(m("", ""))
        self.assertFalse(m("N", "N"))
        self.assertFalse(m("R", "R"))
        self.assertFalse(m("N", "r"))
        self.assertFalse(m("CGUACGCAN", "CGUACGCAN"))
        self.assertFalse(m("U", "C"))
        self.assertFalse(m("UUU", "UUC"))
        self.assertFalse(m("UUU", "UUY"))
        self.assertTrue(m("UU-", "UU-"))
        self.assertTrue(m("UCAG", "UCAG"))

    def test_can_pair(self):
        """MolType can_pair should return True if all positions can pair"""
        p = RnaMolType.can_pair
        self.assertTrue(p("", ""))
        self.assertFalse(p("UCAG", "UCAG"))
        self.assertTrue(p("UCAG", "CUGA"))
        self.assertFalse(p("UCAG", "cuga"))
        self.assertTrue(p("UCAG", "NNNN"))
        self.assertTrue(p("NNNN", "UCAG"))
        self.assertTrue(p("NNNN", "NNNN"))
        self.assertFalse(p("N", "x"))
        self.assertFalse(p("N", "-"))
        self.assertTrue(p("-", "-"))
        self.assertTrue(p("UCAGU", "KYYRR"))
        self.assertTrue(p("UCAG", "KKRS"))
        self.assertTrue(p("U", "G"))

        d = DnaMolType.can_pair
        self.assertFalse(d("T", "G"))

    def test_can_mispair(self):
        """MolType can_mispair should return True on any possible mispair"""
        m = RnaMolType.can_mispair
        self.assertFalse(m("", ""))
        self.assertTrue(m("N", "N"))
        self.assertTrue(m("R", "Y"))
        self.assertTrue(m("N", "r"))
        self.assertTrue(m("CGUACGCAN", "NUHCHUACH"))
        self.assertTrue(m("U", "C"))
        self.assertTrue(m("U", "R"))
        self.assertTrue(m("UUU", "AAR"))
        self.assertTrue(m("UUU", "GAG"))
        self.assertFalse(m("UUU", "AAA"))
        self.assertFalse(m("UCAG", "CUGA"))
        self.assertTrue(m("U--", "--U"))

        d = DnaMolType.can_pair
        self.assertTrue(d("TCCAAAGRYY", "RRYCTTTGGA"))

    def test_must_pair(self):
        """MolType must_pair should return True when no possible mispairs"""
        m = RnaMolType.must_pair
        self.assertTrue(m("", ""))
        self.assertFalse(m("N", "N"))
        self.assertFalse(m("R", "Y"))
        self.assertFalse(m("A", "A"))
        self.assertFalse(m("CGUACGCAN", "NUGCGUACG"))
        self.assertFalse(m("U", "C"))
        self.assertFalse(m("UUU", "AAR"))
        self.assertFalse(m("UUU", "RAA"))
        self.assertFalse(m("UU-", "-AA"))
        self.assertTrue(m("UCAG", "CUGA"))

        d = DnaMolType.must_pair
        self.assertTrue(d("TCCAGGG", "CCCTGGA"))
        self.assertTrue(d("tccaggg", "ccctgga"))
        self.assertFalse(d("TCCAGGG", "NCCTGGA"))

class RnaMolTypeTests(TestCase):
    """Spot-checks of alphabet functionality applied to RNA alphabet."""

    def test_contains(self):
        """RnaMolType should __contain__ the expected symbols."""
        keys = "ucagrymkwsbhvdn?-"
        for k in keys:
            self.assertIn(k, RnaMolType)
        for k in keys.upper():
            self.assertIn(k, RnaMolType)
        self.assertNotIn("X", RnaMolType)

    def test_degenerate_from_seq(self):
        """RnaMolType degenerate_from_seq should give correct results"""
        d = RnaMolType.degenerate_from_seq
        # check monomers
        self.assertEqual(d("a"), "a")
        self.assertEqual(d("C"), "C")
        # check seq of monomers
        self.assertEqual(d("aaaaa"), "a")
        # check some 2- to 4-way cases
        self.assertEqual(d("aaaaag"), "r")
        self.assertEqual(d("ccgcgcgcggcc"), "s")
        self.assertEqual(d("accgcgcgcggcc"), "v")
        self.assertEqual(d("aaaaagcuuu"), "n")
        # check some cases with gaps
        self.assertEqual(d("aa---aaagcuuu"), "?")
        self.assertEqual(d("aaaaaaaaaaaaaaa-"), "?")
        self.assertEqual(d("----------------"), "-")
        # check mixed case example
        self.assertEqual(d("AaAAaa"), "A")
        # check example with degenerate symbols in set
        self.assertEqual(d("RS"), "V")
        self.assertEqual(d("RN-"), "?")
        # check that it works for proteins as well
        p = ProteinMolType.degenerate_from_seq
        self.assertEqual(p("A"), "A")
        self.assertEqual(p("AAA"), "A")
        self.assertEqual(p("DN"), "B")
        self.assertEqual(p("---"), "-")
        self.assertEqual(p("ACD"), "X")
        self.assertEqual(p("ABD"), "X")
        self.assertEqual(p("ACX"), "X")
        self.assertEqual(p("AC-"), "?")

    def test_get_css_styles(self):
        """generates css style mapping and css"""
        css, styles = DNA.get_css_style()
        # styles should consist of core characters appended by DNA.label
        states = list(DNA) + ["ambig", "terminal_ambig"]
        got = {styles[b] for b in states}
        expect = {b + "_dna" for b in states}
        self.assertEqual(got, expect)
        for state in expect:
            self.assertTrue(state in "\n".join(css))

        # check subset of protein
        css, styles = PROTEIN.get_css_style()
        states = list(PROTEIN) + ["ambig", "terminal_ambig"]
        got = {styles[b] for b in states}
        expect = {b + "_protein" for b in states}
        self.assertEqual(got, expect)

    def test_get_css_no_label(self):
        """should not fail when moltype has no label"""
        dna = get_moltype("dna")
        orig_label, dna.label = dna.label, None
        _ = dna.get_css_style()
        dna.label = orig_label

class _AlphabetTestCase(TestCase):
    def assertEqualSeqs(self, a, b):
        """For when we don't care about the type, just the elements"""
        self.assertEqual(list(a), list(b))

    def assertEqualSets(self, a, b):
        self.assertEqual(set(a), set(b))

class DNAAlphabet(_AlphabetTestCase):
    def setUp(self):
        self.alpha = DNA.alphabet

    def test_exclude(self):
        """Nucleotide alphabet testing excluding gap motif"""
        self.assertEqualSeqs(self.alpha, ["T", "C", "A", "G"])

    def test_include(self):
        """Nucleotide alphabet testing including gap motif"""
        self.assertEqualSets(self.alpha.with_gap_motif(), ["A", "C", "G", "T", "-"])

    def test_usesubset(self):
        """testing using a subset of motifs."""
        self.assertEqualSets(self.alpha.with_gap_motif(), ["A", "C", "G", "T", "-"])
        alpha = self.alpha.get_subset(motif_subset=["A"])
        self.assertEqualSets(alpha, ["A"])
        # self.assertRaises(AlphabetError, self.alpha.get_subset, ['A','C'])
        alpha = DNA.alphabet
        self.assertEqualSets(alpha, ["T", "C", "A", "G"])
        alpha = alpha.get_subset(motif_subset=["A", "T", "G"])
        self.assertEqualSets(alpha, ["A", "G", "T"])

class DinucAlphabet(_AlphabetTestCase):
    def setUp(self):
        self.alpha = DNA.alphabet.with_gap_motif().get_word_alphabet(2)

    def test_exclude(self):
        """Dinucleotide alphabet testing excluding gap motif"""
        expected = [
            "-A",
            "-C",
            "-G",
            "-T",
            "A-",
            "AA",
            "AC",
            "AG",
            "AT",
            "C-",
            "CA",
            "CC",
            "CG",
            "CT",
            "G-",
            "GA",
            "GC",
            "GG",
            "GT",
            "T-",
            "TA",
            "TC",
            "TG",
            "TT",
        ]

        self.assertEqualSets(self.alpha.get_subset(["--"], excluded=True), expected)

    def test_include(self):
        """Dinucleotide alphabet testing including gap motif"""
        expected = [
            "--",
            "-A",
            "-C",
            "-G",
            "-T",
            "A-",
            "AA",
            "AC",
            "AG",
            "AT",
            "C-",
            "CA",
            "CC",
            "CG",
            "CT",
            "G-",
            "GA",
            "GC",
            "GG",
            "GT",
            "T-",
            "TA",
            "TC",
            "TG",
            "TT",
        ]
        self.assertEqualSets(self.alpha, expected)

    def test_usesubset(self):
        """testing using a subset of motifs."""
        alpha = self.alpha.get_subset(motif_subset=["AA", "CA", "GT"])
        self.assertEqualSeqs(alpha, ["AA", "CA", "GT"])

        self.assertRaises(
            AlphabetError, alpha.get_subset, motif_subset=["AA", "CA", "GT", "TT"]
        )

    def test_usesubsetbyfreq(self):
        """testing using a subset of motifs by using motif probs."""
        motif_freqs = {
            "--": 0,
            "-A": 0.0,
            "-C": 0,
            "-G": 0,
            "-T": 0,
            "A-": 0,
            "AA": 1.0,
            "AC": 0.0,
            "AG": 0,
            "AT": 0,
            "C-": 0,
            "CA": 1,
            "CC": 0,
            "CG": 0,
            "CT": 0,
            "G-": 0,
            "GA": 0,
            "GC": 0,
            "GG": 0,
            "GT": 1,
            "T-": 0,
            "TA": 0,
            "TC": 0,
            "TG": 0,
            "TT": 0,
        }

        alpha = self.alpha.get_subset(motif_freqs)
        self.assertEqualSets(alpha, ["AA", "CA", "GT"])

    def test_strand_symmetric_motifs(self):
        """construction of strand symmetric motif sets"""
        # fails for a moltype with no strand complement
        with self.assertRaises(TypeError):
            PROTEIN.strand_symmetric_motifs()

        got = DNA.strand_symmetric_motifs(motif_length=1)
        expect = set([("A", "T"), ("C", "G")])
        self.assertEqual(got, expect)
        got = RNA.strand_symmetric_motifs(motif_length=1)
        expect = set([("A", "U"), ("C", "G")])
        self.assertEqual(got, expect)
        got = DNA.strand_symmetric_motifs(motif_length=2)
        self.assertEqual(len(got), 8)
        got = DNA.strand_symmetric_motifs(motif_length=3)
        self.assertEqual(len(got), 32)

class TestCodonAlphabet(_AlphabetTestCase):
    def setUp(self):
        self.alpha = STANDARD_CODON

    def test_ambiguous_gaps(self):
        alpha = self.alpha.with_gap_motif()
        self.assertEqual(len(alpha.resolve_ambiguity("AT?")), 4)
        self.assertRaises(Exception, alpha.resolve_ambiguity, "at-")
        self.assertEqual(len(alpha.resolve_ambiguity("???")), 62)
        self.assertEqual(len(alpha.resolve_ambiguity("---")), 1)

        alpha = self.alpha
        self.assertEqual(len(alpha.resolve_ambiguity("AT?")), 4)
        self.assertRaises(Exception, alpha.resolve_ambiguity, "at-")
        self.assertEqual(len(alpha.resolve_ambiguity("???")), 61)
        self.assertRaises(Exception, alpha.resolve_ambiguity, "---")

    def test_exclude(self):
        """testing excluding gap motif"""
        alpha = self.alpha
        expected = [
            "AAA",
            "AAC",
            "AAG",
            "AAT",
            "ACA",
            "ACC",
            "ACG",
            "ACT",
            "AGA",
            "AGC",
            "AGG",
            "AGT",
            "ATA",
            "ATC",
            "ATG",
            "ATT",
            "CAA",
            "CAC",
            "CAG",
            "CAT",
            "CCA",
            "CCC",
            "CCG",
            "CCT",
            "CGA",
            "CGC",
            "CGG",
            "CGT",
            "CTA",
            "CTC",
            "CTG",
            "CTT",
            "GAA",
            "GAC",
            "GAG",
            "GAT",
            "GCA",
            "GCC",
            "GCG",
            "GCT",
            "GGA",
            "GGC",
            "GGG",
            "GGT",
            "GTA",
            "GTC",
            "GTG",
            "GTT",
            "TAC",
            "TAT",
            "TCA",
            "TCC",
            "TCG",
            "TCT",
            "TGC",
            "TGG",
            "TGT",
            "TTA",
            "TTC",
            "TTG",
            "TTT",
        ]
        self.assertEqualSets(alpha, expected)

    def test_include(self):
        """testing including gap motif"""
        alpha = self.alpha.with_gap_motif()
        expected = [
            "---",
            "AAA",
            "AAC",
            "AAG",
            "AAT",
            "ACA",
            "ACC",
            "ACG",
            "ACT",
            "AGA",
            "AGC",
            "AGG",
            "AGT",
            "ATA",
            "ATC",
            "ATG",
            "ATT",
            "CAA",
            "CAC",
            "CAG",
            "CAT",
            "CCA",
            "CCC",
            "CCG",
            "CCT",
            "CGA",
            "CGC",
            "CGG",
            "CGT",
            "CTA",
            "CTC",
            "CTG",
            "CTT",
            "GAA",
            "GAC",
            "GAG",
            "GAT",
            "GCA",
            "GCC",
            "GCG",
            "GCT",
            "GGA",
            "GGC",
            "GGG",
            "GGT",
            "GTA",
            "GTC",
            "GTG",
            "GTT",
            "TAC",
            "TAT",
            "TCA",
            "TCC",
            "TCG",
            "TCT",
            "TGC",
            "TGG",
            "TGT",
            "TTA",
            "TTC",
            "TTG",
            "TTT",
        ]
        self.assertEqualSets(alpha, expected)

    def test_constructing_from_func(self):
        """the CodonAlphabet function should support genetic code names as well"""
        alpha_int = CodonAlphabet(1)
        alpha_name = CodonAlphabet("Standard Nuclear")
        self.assertEqual(alpha_int, alpha_name)
        alpha_int = CodonAlphabet(2)
        alpha_name = CodonAlphabet("Vertebrate Mitochondrial")
        self.assertEqual(alpha_int, alpha_name)

if __name__ == "__main__":
    main()

reason: from numpy


text:
#!/usr/bin/env python3

import numpy as np

import pyboof as pb

original = pb.load_single_band('../data/example/outdoors01.jpg', np.uint8)

# Let BoofCV decide on the type of image to store the gradient as
deriv_dtype = pb.gradient_dtype(pb.get_dtype(original))

# Declare the gradient images
derivX = pb.create_single_band(original.getWidth(), original.getHeight(), deriv_dtype)
derivY = pb.create_single_band(original.getWidth(), original.getHeight(), deriv_dtype)

# Compute the results for a few operators and visualize
pb.gradient(original, derivX, derivY, pb.GradientType.SOBEL)
buffered_sobel = pb.swing.colorize_gradient(derivX, derivY)

pb.gradient(original, derivX, derivY, pb.GradientType.PREWITT)
buffered_prewitt = pb.swing.colorize_gradient(derivX, derivY)

pb.gradient(original, derivX, derivY, pb.GradientType.SCHARR)
buffered_pscharr = pb.swing.colorize_gradient(derivX, derivY)

pb.gradient(original, derivX, derivY, pb.GradientType.THREE)
buffered_three = pb.swing.colorize_gradient(derivX, derivY)

pb.gradient(original, derivX, derivY, pb.GradientType.TWO0)
buffered_two0 = pb.swing.colorize_gradient(derivX, derivY)

pb.gradient(original, derivX, derivY, pb.GradientType.TWO1)
buffered_two1 = pb.swing.colorize_gradient(derivX, derivY)

# display the results in a single window as a list
image_list = [(original, "original"),
              (buffered_sobel, "sobel"),
              (buffered_prewitt, "prewitt"),
              (buffered_pscharr, "scharr"),
              (buffered_three, "three"),
              (buffered_two0, "two0"),
              (buffered_two1, "two1")]
pb.swing.show_list(image_list, title="Gradients")

input("Press any key to exit")

ext: py

max_stars_repo_name: lessthanoptimal/PyBoof

max_issues_repo_name: lessthanoptimal/PyBoof

max_forks_repo_name: lessthanoptimal/PyBoof

reason: import numpy


text:
import itertools

import networkx
from matplotlib import pyplot

from soap.common import base_dispatcher, cached_property, Flyweight
from soap.expression import (
    InputVariable, External, InputVariableTuple, AccessExpr, UpdateExpr,
    operators
)
from soap.semantics import is_numeral

class ExpressionDependencies(base_dispatcher()):
    """
    Find dependent variables for the corresponding expression
    """
    def generic_execute(self, expr):
        raise TypeError(
            'Do not know how to find dependencies in expression {!r}'
            .format(expr))

    def _execute_atom(self, expr):
        return [expr]

    execute_Label = execute_PartitionLabel = execute_numeral = _execute_atom

    def execute_Variable(self, expr):
        return [InputVariable(expr.name, expr.dtype)]

    def _execute_expression(self, expr):
        return list(expr.args)

    execute_UnaryArithExpr = execute_BinaryArithExpr = _execute_expression
    execute_UnaryBoolExpr = execute_BinaryBoolExpr = _execute_expression
    execute_SelectExpr = execute_Subscript = _execute_expression
    execute_AccessExpr = execute_UpdateExpr = _execute_expression

    def execute_FixExpr(self, expr):
        # is fixpoint expression, find external dependencies in init_state
        deps = []
        for v in expr.init_state.values():
            if not isinstance(v, External):
                continue
            deps.append(v.var)
        return deps

    def execute_External(self, expr):
        # external dependencies taken care of by FixExpr dependencies.
        return []

    execute_InputVariableTuple = _execute_expression
    execute_OutputVariableTuple = _execute_atom
    execute_MetaState = execute_LabelSemantics = _execute_atom

expression_dependencies = ExpressionDependencies()

class DependenceGraph(Flyweight):
    """Graph of dependences"""
    class _RootNode(object):
        def __str__(self):
            return '<root>'
        __repr__ = __str__

    deps_func = staticmethod(expression_dependencies)

    def __init__(self, env, out_vars):
        super().__init__()
        self.env = env
        new_out_vars = []
        for var in out_vars:
            if var not in new_out_vars:
                new_out_vars.append(var)
        self.out_vars = tuple(new_out_vars)
        self._root_node = self._RootNode()

    @cached_property
    def graph(self):
        graph = networkx.DiGraph()
        self._init_edges_recursive(graph, self._root_node)
        self._init_array_dependences(graph)
        return graph

    def add_dep_edges_one_to_many(self, graph, from_node, to_nodes):
        graph.add_edges_from(
            self.edge_attr(from_node, to_node) for to_node in to_nodes)

    def edge_attr(self, from_node, to_node):
        return from_node, to_node, {}

    def _init_edges_recursive(self, graph, out_vars):
        from soap.transformer.partition import PartitionLabel
        prev_nodes = graph.nodes()
        if out_vars == self._root_node:
            # terminal node
            self.add_dep_edges_one_to_many(
                graph, self._root_node, self.out_vars)
            deps = self.out_vars
        else:
            deps = []
            for var in out_vars:
                if is_numeral(var):
                    continue
                if isinstance(var, (InputVariable, PartitionLabel)):
                    continue
                if isinstance(var, InputVariableTuple):
                    expr = var
                else:
                    expr = self.env[var]
                local_deps = self.deps_func(expr)
                deps += local_deps
                self.add_dep_edges_one_to_many(graph, var, local_deps)
        if not deps:
            return
        new_deps = []
        for v in deps:
            if v not in prev_nodes:
                new_deps.append(v)
        self._init_edges_recursive(graph, new_deps)

    def _array_nodes(self, graph):
        from soap.transformer.partition import PartitionLabel

        def is_array_op(node):
            if isinstance(
                    node, (InputVariable, InputVariableTuple, PartitionLabel)):
                return False
            if is_numeral(node):
                return False
            if node == self._root_node:
                return False
            expr = self.env[node]
            return isinstance(expr, (AccessExpr, UpdateExpr))

        return (n for n in graph.nodes() if is_array_op(n))

    array_edge_attr = edge_attr

    def _init_array_dependences(self, graph):
        """
        Any updates to array must happen after access to the same array.  This
        method finds all such occurrences and add them to the dependence graph.
        """
        nodes = self._array_nodes(graph)
        for update_node, access_node in itertools.product(nodes, repeat=2):
            update = self.env[update_node]
            access = self.env[access_node]
            check = (
                update.op == operators.INDEX_UPDATE_OP and
                access.op == operators.INDEX_ACCESS_OP and
                access.var == update.var)
            if not check:
                continue
            graph.add_edge(*self.array_edge_attr(update_node, access_node))

    def nodes(self):
        try:
            return self._nodes
        except AttributeError:
            pass
        self._nodes = self.graph.nodes()
        return self._nodes

    def edges(self):
        try:
            return self._edges
        except AttributeError:
            pass
        self._edges = self.graph.edges()
        return self._edges

    def is_cyclic(self):
        return any(networkx.simple_cycles(self.graph))

    def input_vars(self):
        return (v for v in self.nodes() if isinstance(v, InputVariable))

    def _ignore_root_node(self, nodes):
        return (n for n in nodes if n != self._root_node)

    def dfs_preorder(self):
        # strange non-deterministic dependence bug when using networkx's
        # dfs_preorder method
        return reversed(list(self.dfs_postorder()))

    def dfs_postorder(self):
        return self._ignore_root_node(
            networkx.dfs_postorder_nodes(self.graph, self._root_node))

    def is_multiply_shared(self, node):
        return len(self.predecessors(node)) > 1

    def _draw(self, graph):
        pos = networkx.graphviz_layout(graph, prog='dot')
        networkx.draw_networkx_nodes(graph, pos, node_size=70)
        networkx.draw_networkx_edges(graph, pos)
        networkx.draw_networkx_labels(
            graph, pos, font_size=20, font_family='sans-serif')
        pyplot.axis('off')

    def show(self):
        self._draw(self.graph)
        pyplot.show()

class HierarchicalDependenceGraph(DependenceGraph):
    # FIXME broken
    def __init__(self, env, out_var, _parent_nodes=None):
        super().__init__(env, out_var)
        self._parent_nodes = _parent_nodes
        self._flat_edges = self.edges
        self.edges, self._subgraphs = self._partition({}, self.out_var)
        self._local_nodes = None

    @property
    def flat_edges(self):
        return self._flat_edges

    @property
    def subgraphs(self):
        return self._subgraphs

    def _partition(self, subgraphs, out_var):

        def local_nodes(var):
            def all_paths_converge_to_var(dep_var, var):
                if dep_var == var:
                    return True
                prev_vars = self.prev(dep_var)
                if not prev_vars:
                    return False
                return all(all_paths_converge_to_var(prev_var, var)
                           for prev_var in prev_vars)
            sphere = set()
            local_deps = self.dependencies(var)
            for dep_var in local_deps:
                if self._parent_nodes and dep_var not in self._parent_nodes:
                    continue
                if all_paths_converge_to_var(dep_var, var):
                    sphere.add(dep_var)
            return sphere

        edges = set()
        if isinstance(out_var, InputVariableTuple):
            out_vars = list(out_var)
        else:
            out_vars = [out_var]

        for var in out_vars:
            var_locals = local_nodes(var)

            if var == self.out_var:
                # root node is not considered for partitioning, since the
                # hierarchy itself is the partitioning of it.
                subgraph = var
                subgraph_deps = self.next(var)
            elif var_locals:
                # if has local hierarchy, treat locals as a subgraph, and
                # construct its dependencies
                var_locals.add(var)
                subgraph = subgraphs.get(var)
                if not subgraph:
                    local_env = {}
                    for k, v in self.env.items():
                        if k in var_locals:
                            local_env[k] = v
                    subgraph = HierarchicalDependenceGraph(
                        local_env, var, _parent_nodes=var_locals)
                    subgraphs[var] = subgraph
                subgraph_deps = subgraph.graph_dependencies()
            else:
                # no local hierarchy, standalone node
                subgraph = var
                subgraph_deps = self.next(var)

            # recursively find subgraphs in dependencies
            for dep in subgraph_deps:
                dep_edges, subgraphs = self._partition(subgraphs, dep)
                edges |= dep_edges

            # connect subgraph to its dependencies
            for dep in subgraph_deps:
                dep = subgraphs.get(dep, dep)
                edges.add((subgraph, dep))

        return edges, subgraphs

    def graph_dependencies(self):
        edges = self.flat_edges
        ends = {node for _, node in edges}
        starts = {node for node, _ in edges}
        return ends - starts

    @property
    def local_nodes(self):
        nodes = self._local_nodes
        if nodes:
            return nodes
        self._local_nodes = {node for node, _ in self.edges}
        return self._local_nodes

    def local_order(self):
        local_nodes = self.local_nodes
        # sometimes out_vars are not generated because not in edges?
        out_vars = self.out_var
        if not isinstance(out_vars, InputVariableTuple):
            out_vars = [out_vars]
        for var in out_vars:
            if not self.flat_contains(var):
                local_nodes.add(var)
        return self.order_by_dependencies(local_nodes)

    def flat_contains(self, node):
        nodes = self.local_nodes
        for each_node in nodes:
            if isinstance(each_node, HierarchicalDependenceGraph):
                if each_node.flat_contains(node):
                    return True
            elif node == each_node:
                return True
        return False

    def __eq__(self, other):
        try:
            return self.env == other.env
        except AttributeError:
            return False

    def __hash__(self):
        return hash(tuple(self.env.items()))

    def __str__(self):
        return '{{{}}}'.format(', '.join(str(n) for n in self.local_nodes))

    def __repr__(self):
        return '{cls}({nodes})'.format(
            cls=self.__class__.__name__, nodes=self.local_nodes)

max_stars_repo_name: gitter-badger/soap

max_issues_repo_name: gitter-badger/soap

max_forks_repo_name: gitter-badger/soap

reason: import networkx


text:
import unittest
from unittest.mock import patch
import greenbutton.greenbutton as gb

import pandas as pd
import numpy as np
import datetime
import os

THISDIR = os.path.dirname(os.path.abspath(__file__))

class TestStringMethods(unittest.TestCase):

    def test_has_sample_data(self):
        self.assertTrue(isinstance(gb.SAMPLE_DATA,pd.DataFrame))

    def test_create_box_plot_of_use_by_hour(self):
        ax = gb.boxplot_use_by_hour(gb.SAMPLE_DATA)
        self.assertIsNotNone(ax)
        self.assertNotIn('Start Hour',gb.SAMPLE_DATA.columns)
        self.assertEqual(168,len(ax.lines))

    def test_filter_by_time_of_day(self):
        test_df = gb.SAMPLE_DATA.copy()
        filtered_df = gb.filter_by_time_of_day(test_df,datetime.time(8,0,0),datetime.time(9,0,0))
        unique_filtered_df = filtered_df['Start Time'].dt.time.unique()
        self.assertTrue(np.isin(datetime.time(8,0),unique_filtered_df))
        self.assertTrue(np.isin(datetime.time(9,0),unique_filtered_df))
        self.assertTrue(2,len(unique_filtered_df))

    def test_can_load_data_from_file(self):
        # Check that loaded data is not just from hardcoded XMLFILE
        datafile = os.path.join(THISDIR,'data','testdata.xml')
        df = gb.dataframe_from_xml(datafile)
        self.assertEqual(pd.Timestamp(2019,10,1,0),df.loc[0]['Start Time'])

    def test_aggregate_use_by_day(self):
        pass

    def test_sample_data_is_immutable(self):
        # see: https://stackoverflow.com/questions/24928306/pandas-immutable-dataframe
        pass

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

max_stars_repo_name: JenDobson/greenbutton

max_issues_repo_name: JenDobson/greenbutton

max_forks_repo_name: JenDobson/greenbutton


text:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from collections import defaultdict

import numpy as np
from scipy.special import digamma, gammaln
# from tqdm import tqdm  # with .py script
from tqdm.notebook import tqdm  # with notebook

# TODO: create object to other file and inheritance
class LDA:
    def __init__(self, users, items, n_topics=3, max_iterations=10):
        # TODO: add arg parser
        self.max_iterations = max_iterations
        self.finish_ratio = 1.0e-5
        self.llh = None
        # self.prev_llh = 100000.0
        self.perplexity = None
        self.prev_perplexity = 100000.0
        self.seed = 2020
        np.random.seed(seed=self.seed)

        # hyper parameters
        self.alpha = 0.1
        self.beta = 0.1

        self.users = users
        self.items = items
        self.n_users = len(set(users))
        self.n_items = len(set(items))
        self.n_data = len(users)
        self.n_topics = n_topics
        self.n_parameters = (self.n_topics - 1) \
            + (self.n_topics * self.n_users) \
            + (self.n_topics * self.n_items) \
            + 2  # alpha, beta

        # TODO: remove test
        print('n_data: {0} | n_users: {1} | n_items: {2} | \
            n_topics: {3}'.format(
            self.n_data,
            self.n_users,
            self.n_items,
            self.n_topics,
        ))

        self._generate_handle_data()
        self._generate_counter_matrixes()

    def _generate_handle_data(self):
        self.dict_user_items = defaultdict(list)
        for n in range(self.n_data):
            self.dict_user_items[self.users[n]].append(self.items[n])

    def _generate_counter_matrixes(self):
        # number of times user u and topic z co-occur
        self.matrix_n_uz = np.zeros((self.n_users, self.n_topics))
        # number of times topic z and item i co-occur
        self.matrix_n_zi = np.zeros((self.n_topics, self.n_items))
        # number of times user u occur
        self.matrix_n_u = np.zeros(self.n_users)
        # number of times topic z occur
        self.matrix_n_z = np.zeros(self.n_topics)
        # topic of each user and item pair
        self.topic_ui = list(map(lambda x: np.zeros(len(x)),
                                 [self.dict_user_items[i]
                                  for i in range(self.n_users)]))

        for userid in range(len(self.dict_user_items)):
            for j in range(len(self.dict_user_items[userid])):
                itemid = self.dict_user_items[userid][j]
                topic = np.random.randint(self.n_topics)

                self.matrix_n_uz[userid, topic] += 1
                self.matrix_n_zi[topic, itemid] += 1
                self.matrix_n_u[userid] += 1
                self.matrix_n_z[topic] += 1
                self.topic_ui[userid][j] = topic

    def train(self):
        count = 0
        for iteration in tqdm(range(self.max_iterations)):
            self._gibbs_sampling()
            self._update_alpha()
            self._update_beta()
            self._calc_loglikelihood()
            self._calc_perplexity()

            ratio = abs((self.perplexity - self.prev_perplexity)
                        / self.prev_perplexity)
            if self.perplexity > self.prev_perplexity:
                count += 1
            else:
                count = 0

            tqdm.write(" - iter : {0} | perplexity : {1} | ratio : {2} | \
                count : {3}".format(
                iteration + 1,
                self.perplexity,
                ratio,
                count,
            ))

            # (変化率が一定以下，もしくは2回以上最低値を連続で更新できなかった場合に収束)
            if (ratio < self.finish_ratio) or (count >= 2):
                break
            self.prev_perplexity = self.perplexity

        self._calc_perplexity()
        print("finish training! | perplexity : {0} | pz : {1}".format(
            self.perplexity,
            self.get_pz(),
        ))

    def _gibbs_sampling(self):
        for i in range(len(self.dict_user_items)):
            for j in range(len(self.dict_user_items[i])):
                itemid = self.dict_user_items[i][j]

                # decrease each counters
                topic = int(self.topic_ui[i][j])
                self.matrix_n_uz[i, topic] -= 1
                self.matrix_n_zi[topic, itemid] -= 1
                self.matrix_n_z[topic] -= 1

                # increase each counters
                topic = self._sampling_topic(i, itemid)
                self.topic_ui[i][j] = topic
                self.matrix_n_uz[i, topic] += 1
                self.matrix_n_zi[topic, itemid] += 1
                self.matrix_n_z[topic] += 1

    def _sampling_topic(self, userid, itemid):
        left = (self.matrix_n_uz[userid, :] + self.alpha) / \
            (self.matrix_n_u[userid] + self.alpha * self.n_topics)
        right = (self.matrix_n_zi[:, itemid] + self.beta) / \
            (self.matrix_n_z + self.beta * self.n_items)
        prob = right * left
        prob /= prob.sum()

        return np.random.multinomial(1, prob).argmax()

    def _update_alpha(self):
        """
        update hyperparameter alpha
        """
        numer = digamma(self.matrix_n_uz + self.alpha).sum() - \
            self.n_users * self.n_topics * digamma(self.alpha)
        denom = self.n_topics * digamma(
            self.matrix_n_u + self.alpha * self.n_topics).sum(
        ) - self.n_users * self.n_topics * digamma(
            self.alpha * self.n_topics)

        self.alpha = (self.alpha * numer) / denom

    def _update_beta(self):
        """
        update hyperparameter beta
        """
        numer = digamma(self.matrix_n_zi + self.beta).sum() - \
            self.n_topics * self.n_items * digamma(self.beta)
        denom = self.n_items * digamma(
            self.matrix_n_z + self.beta * self.n_items).sum(
        ) - self.n_topics * self.n_items * digamma(
            self.beta * self.n_items)

        self.beta = (self.beta * numer) / denom

    def _calc_loglikelihood(self):
        """
        calculate loglikelihood
        """
        # ref: https://gist.github.com/mblondel/542786#file-lda_gibbs-py-L91
        def log_multi_beta(alpha, K=None):
            if K is None:
                # alpha is assumed to be a vector
                return np.sum(gammaln(alpha)) - gammaln(np.sum(alpha))
            else:
                # alpha is assumed to be a scalar
                return K * gammaln(alpha) - gammaln(K * alpha)

        llh = 0
        for z in range(self.n_topics):
            llh += log_multi_beta(self.matrix_n_zi[z, :] + self.beta)
            llh -= log_multi_beta(self.beta, self.n_items)

        for m in range(self.n_users):
            llh += log_multi_beta(self.matrix_n_uz[m, :] + self.alpha)
            llh -= log_multi_beta(self.alpha, self.n_topics)

        self.llh = llh

    # ref:
    # https://github.com/satopirka/topic-model/blob/master/lda_with_gibbs_sampling.py#L86
    def _calc_perplexity(self):
        """
        calculate perplexity
        """
        item_dist4topics = self.get_phi()
        topic_dist4users = self.get_theta()
        sum_prob = 0.0

        # TODO: escape loop
        for i in range(len(self.dict_user_items)):
            for j in range(len(self.dict_user_items[i])):
                topic_dist4items = item_dist4topics[:,
                                                    self.dict_user_items[i][j]]
                sum_prob += np.log(np.dot(topic_dist4items,
                                          topic_dist4users[i]))

        self.perplexity = np.exp(- (1 / self.n_data) * sum_prob)

    # TODO: add function to compute Coherence
    # def _calc_coherence(self):

    def get_phi(self):
        """
        Compute phi = p(i|z).
        """
        # (z, i)
        numer = self.matrix_n_zi + self.beta
        # (z, ) -> (z, 1)
        denom = self.matrix_n_z[:, np.newaxis] + self.n_items * self.beta
        return numer / denom

    def get_theta(self):
        """
        Compute theta = p(u|z).
        """
        numer = self.matrix_n_uz + self.alpha  # (u, z)
        # (u,) -> (u, 1)
        denom = self.matrix_n_u[:, np.newaxis] + self.n_topics * self.alpha
        return numer / denom

    def get_pz(self):
        """
        Compute p(z).
        """
        pu_z = self.get_theta()
        prob = pu_z.sum(axis=0)
        prob /= prob.sum()
        return prob

    def get_aic(self):
        aic = ((-2) * self.llh) + (2 * self.n_parameters)
        return aic

    def get_bic(self):
        bic = ((-2) * self.llh) + \
            (self.n_parameters * np.log(self.n_data))
        return bic

    # TODO: add functions to calculate each probability for result analysis
    # def get_pz_u(self):
    #     PzPu_z = self.Pu_z * self.Pz
    #     Pz_u = PzPu_z / np.sum(PzPu_z, axis=1).reshape(self.n_users, 1)
    #     return Pz_u

    # def get_pz_i(self):
    #     PzPi_z = self.Pi_z * self.Pz
    #     Pz_i = PzPi_z / np.sum(PzPi_z, axis=1).reshape(self.n_items, 1)
    #     return Pz_i

    # TODO: make function to merge master(like index2name) and show top N items

max_stars_repo_name: jchanxtarov/plsa

max_issues_repo_name: jchanxtarov/plsa

max_forks_repo_name: jchanxtarov/plsa

reason: import numpy,from scipy


text:
import os
from tkinter.messagebox import NO
import torch
import torch.nn as nn
import numpy as np
import pdb
import cv2
import torch.nn.functional as F
import sys
sys.path.append('/home/jty/mvs/idn-solver/')
from utils import depth2normal

class Solver(nn.Module):
    def __init__(self, h, w, alpha1=10.0, alpha2=10.0, sigma1=20.0, sigma2=3.0, with_sommth=False):
        # check_offset: The checkerboard size to fetch depth
        # alpha1, alpha2: The weights of data term in depth update and normal update
        # sigma1, sigma2: The threshold value in color and distance weighting
        super(Solver, self).__init__()
        self.alpha1 = alpha1
        self.alpha2 = alpha2
        self.sigma1 = sigma1
        self.sigma2 = sigma2
        self.h, self.w = h, w
        yy, xx = torch.meshgrid(torch.arange(self.h), torch.arange(self.w))
        self.xy = torch.stack([xx.float(),yy.float()], 0)
        self.xy_homo = torch.cat([self.xy, torch.ones((1,h,w))], 0).cuda() # [3,h,w]
        self.xy_homo = self.xy_homo.reshape(3,-1).unsqueeze(0) # [1,3,-1]

        self.with_sommth = with_sommth

    def propagate_axis(self, depth, normal, rgb, conf, offset, K_inv):
        # depth & conf: [b, 1, h, w], rgb & normal: [b, 3, h, w], K_inv: [b, 3, 3]

        b = depth.shape[0]
        xy_homo_ = self.xy_homo.repeat(b,1,1) # [b, 3, h*w]
        xy_3d = torch.matmul(K_inv, xy_homo_).reshape(b, 3, self.h, self.w)
        nom = depth * torch.sum(normal * xy_3d, dim=1, keepdim=True)

        xy_homo_ref = xy_homo_.clone()
        # 点 i上存放点j在 i平面上的深度值
        _b, c, h, w = offset.shape
        offset = offset.reshape(_b, c, h * w)
        xy_homo_ref[:,:2,:] = xy_homo_ref[:,:2,:] + offset

        denom = torch.sum(normal * torch.matmul(K_inv, xy_homo_ref).reshape(b,3,self.h,self.w), dim=1, keepdim=True)
        d1_p =  (nom / (denom + 1e-18))
        # if self.with_sommth:
        #     # p = 0.7
        #     # from pdb  import set_trace; set_trace()
        #     d1_p = (1-conf )* d1_p + conf * depth
        d1_p =  d1_p.clamp(0.01, 10.0)
        # 点 i存档的变换后的点j的深度值变换回到 j原始储存点

        xy_homo_ref = xy_homo_.clone()
        xy_back = xy_homo_ref[:,:2,:] - offset
        xy_back = xy_back.reshape(-1, 2, self.h, self.w).permute(0,2,3,1)
        xy_back_mask = (((xy_back[..., 0] < 0) | (xy_back[..., 0] > self.w)) |  \
            ((xy_back[..., 1] < 0) | (xy_back[..., 1] > self.h))).unsqueeze(1)

        xy_back[..., 0] = (xy_back[..., 0] - (self.w-1)/2 ) / ((self.w-1)/2)
        xy_back[..., 1] = (xy_back[..., 1] - (self.h-1)/2 ) / ((self.h-1)/2)

        d1 = F.grid_sample(d1_p, xy_back ,align_corners=True)

        # 若是在索引之外的点用原始深度代替
        # from pdb import set_trace; set_trace()
        d1[xy_back_mask] = depth[xy_back_mask]

        # confidence 需要j点位置储存i点的
        conf1 = F.grid_sample(conf, xy_back, padding_mode='zeros',mode='bilinear',align_corners=True)
        # 若是在索引之外的点用conf为0
        conf1[xy_back_mask] = 0

        rgb1 = F.grid_sample(rgb, xy_back, padding_mode='zeros',mode='bilinear',align_corners=True)
        xy_back_mask = xy_back_mask.repeat_interleave(dim=1, repeats=3)
        rgb1[xy_back_mask] = 0

        return d1, conf1, rgb1

    def checkerboard_propagate(self, depth, normal, rgb, conf, K_inv, profiler=None, offsets=None):
        b, _, h, w = depth.shape
        if profiler is not None:
            profiler.report_process('before zeros')

        propagated_depth, propagated_conf, propagated_rgb = [], [], []
        distance = []
        if profiler is not None:
            profiler.report_process('checkerboard propagate prev')

        _b, _c, _h, _w = offsets.shape
        offsets = offsets.reshape( _b, -1, 2, _h, _w)
        nums_offsets = offsets.shape[1]

        for i in range(nums_offsets):
            offset = offsets[:, i]
            d1, conf1, rgb1 = self.propagate_axis(depth, normal, rgb, conf, offset, K_inv)
            propagated_depth.append(d1)
            propagated_conf.append(conf1)
            propagated_rgb.append(rgb1)
            distance.append(offset)

        propagated_depth, propagated_conf, propagated_rgb = torch.stack(propagated_depth, 1), torch.stack(propagated_conf, 1), torch.stack(propagated_rgb, 1)
        distance = torch.stack(distance, 1).abs().norm(dim=2, keepdim=True)

        return propagated_depth, propagated_conf, propagated_rgb, distance

    def propagate_axis_less(self, points, conf, offset):
        # Up, down, left and right directions
        b = points.shape[0]
        xy_homo_ = self.xy_homo.repeat(b,1,1) # [b, 3, h*w]
        xy_homo_ref = xy_homo_.clone()
        _b, c, h, w = offset.shape
        offset = offset.reshape(_b, c, h * w)
        xy_back = xy_homo_ref[:,:2,:] - offset
        xy_back = xy_back.reshape(-1, 2, self.h, self.w).permute(0,2,3,1)
        xy_back_mask = (((xy_back[..., 0] < 0) | (xy_back[..., 0] > self.w)) | \
            ((xy_back[..., 1] < 0) | (xy_back[..., 1] > self.h))).unsqueeze(1)

        xy_back[..., 0] = (xy_back[..., 0] - (self.w-1)/2 ) / ((self.w-1)/2)
        xy_back[..., 1] = (xy_back[..., 1] - (self.h-1)/2 ) / ((self.h-1)/2)

        # confidence 需要j点位置储存i点的
        conf1 = F.grid_sample(conf, xy_back, padding_mode='zeros',mode='bilinear',align_corners=True)
        # 若是在索引之外的点用conf为0
        conf1[xy_back_mask] = 0

        points1 = F.grid_sample(points, xy_back, padding_mode='zeros',mode='bilinear',align_corners=True)
        xy_back_mask = xy_back_mask.repeat_interleave(dim=1, repeats=3)
        points1[xy_back_mask] = 0

        return points1, conf1

    def checkerboard_propagate_less(self, points, conf, offsets=None):
        b, _, h, w = conf.shape
        propagated_conf, propagated_points = [], []

        _b, _c, _h, _w = offsets.shape
        offsets = offsets.reshape( _b, -1, 2, _h, _w)
        nums_offsets = offsets.shape[1]

        for i in range(nums_offsets):

            offset = offsets[:, i]
            points1, conf1 = self.propagate_axis_less(points, conf, offset)
            propagated_points.append(points1)
            propagated_conf.append(conf1)

        propagated_points = torch.stack(propagated_points, 1)
        propagated_conf = torch.stack(propagated_conf, 1)
        return propagated_points, propagated_conf

    def forward(self, depth, normal, image, conf, confN, K, profiler=None, offsets=None, affweights=None):
        # Update the depth and normal value.
        # depth: [B, 1, H, W], normal: [B, 3, H, W], image: [B, 3, H, W], conf: depth confidence [B, 1, H, W], confN: normal confidence
        # Update the depth
        b = depth.shape[0]
        K_inv = torch.inverse(K.cpu()).to(depth.get_device())
        checkerboard_depth, checkerboard_conf, checkerboard_rgb, checkerboard_dis = \
            self.checkerboard_propagate(depth, normal, image, conf*confN, K_inv, profiler=profiler, offsets=offsets)

        if affweights is None :
            colorweights = image.unsqueeze(1).repeat(1,checkerboard_rgb.shape[1],1,1,1) - checkerboard_rgb
            colorweights = torch.exp(-torch.sum(torch.pow(colorweights, 2), 2) / self.sigma1).unsqueeze(2).detach()
            spatialweights = torch.exp(-checkerboard_dis / self.sigma2).detach()
            affweights = colorweights * spatialweights

        if affweights.shape.__len__() == 4:
            affweights = affweights.unsqueeze(2)

        checkerboard_conf = checkerboard_conf * affweights

        # if self.with_sommth:
        #     # from pdb import set_trace; set_trace()
        #     denom =  2 * checkerboard_conf.sum(1) + self.alpha1 * conf
        # else:
        denom = checkerboard_conf.sum(1) + self.alpha1 * conf
        nom = (checkerboard_depth * checkerboard_conf).sum(1) + self.alpha1 * conf * depth
        updated_depth = nom / (denom + 1e-16)
        updated_depth = torch.clamp(updated_depth, 0.1, 10.0)

        # Update the normal
        # Using the plane fitting loss
        xy_homo_ = self.xy_homo.repeat(b,1,1)
        points = torch.matmul(K_inv, xy_homo_)
        points = updated_depth * points.reshape(b, 3, self.h, self.w)
        normal_1 = -normal / normal[:,2:,:,:] # [a,b,-1]

        checkerboard_points, checkerboard_dconf = self.checkerboard_propagate_less(points, conf, offsets=offsets)

        checkerboard_dconf = checkerboard_dconf * affweights
        residual = points.unsqueeze(1) - checkerboard_points
        checkerboard_dconf = checkerboard_dconf.squeeze(2) * conf

        A11 = self.alpha2 * confN + (checkerboard_dconf * (residual[:,:,0,:,:])**2).sum(1, keepdim=True)
        A12 = (checkerboard_dconf * residual[:,:,0,:,:] * residual[:,:,1,:,:]).sum(1, keepdim=True)
        A21 = (checkerboard_dconf * residual[:,:,0,:,:] * residual[:,:,1,:,:]).sum(1, keepdim=True)
        A22 = self.alpha2 * confN + (checkerboard_dconf * (residual[:,:,1,:,:])**2).sum(1, keepdim=True)
        b1 = self.alpha2 * confN * normal_1[:,0,:,:].unsqueeze(1) + (checkerboard_dconf * residual[:,:,2,:,:] * residual[:,:,0,:,:]).sum(1, keepdim=True)
        b2 = self.alpha2 * confN * normal_1[:,1,:,:].unsqueeze(1) + (checkerboard_dconf * residual[:,:,2,:,:] * residual[:,:,1,:,:]).sum(1, keepdim=True)
        det = A11 * A22 - A12 * A21
        n1 = (b1 * A22 - b2 * A12) / (det + 1e-6)
        n2 = (b2 * A11 - b1 * A21) / (det + 1e-6)
        n1 = torch.clamp(n1, -20.0, 20.0)
        n2 = torch.clamp(n2, -20.0, 20.0)
        filler = -1.0 * torch.ones((b,1,self.h,self.w)).to(n1.get_device())
        updated_normal = torch.cat([n1,n2,filler], dim=1)
        updated_normal = -updated_normal / torch.norm(updated_normal, p=2, dim=1, keepdim=True)

        return updated_depth, updated_normal

def normal2color(normal_map):
    """
    colorize normal map
    :param normal_map: range(-1, 1)
    :return:
    """
    tmp = normal_map / 2. + 0.5  # mapping to (0, 1)
    color_normal = (tmp * 255).astype(np.uint8)

    return color_normal

if __name__ == '__main__':
    print('Test case...')
    # A whole plain 3D plane filling the incorrect values
    img = cv2.imread('/6t/jty/scannet/scannet_nas/train/scene0000_00/color/0000.jpg')
    depth = np.load('/6t/jty/scannet/scannet_nas/train/scene0000_00/depth/0000.npy').astype(np.float32)
    depth_f = cv2.medianBlur(depth, 5)
    depth_f = depth_f / 5000.0
    K = np.array([[535.4, 0.0, 320.1], [0.0, 539.2, 247.6], [0.0, 0.0, 1.0]])
    K = K[np.newaxis, ...]
    depth_f = depth_f[np.newaxis, ...]
    normal = depth2normal(torch.from_numpy(depth_f).float().unsqueeze(0), torch.from_numpy(np.linalg.inv(K)).float())
    normal = normal + 1e-16
    normal_img = normal[0].numpy()
    cv2.imwrite('/home/jty/mvs/idn-solver/vis/test_normal.png', normal2color(normal_img).transpose(1,2,0))
    #

    solver = Solver(h=480, w=640, check_offsets=[1,2,3,5,7,20], alpha1=5.0, alpha2=5.0, sigma1=1000.0, sigma2=10.0)
    iter_ = 5
    conf = np.where(depth_f > 0.001, 1.0, 0.01)
    depth_var = torch.from_numpy(depth_f).unsqueeze(1).float().cuda()
    normal_var = normal.float().cuda()
    img_var = torch.from_numpy(img).float().cuda().permute(2,0,1).unsqueeze(0)
    conf_var = torch.from_numpy(conf).float().cuda().unsqueeze(-1).permute(0,3,1,2)
    K_var = torch.from_numpy(K).float().cuda()

    x, y = np.arange(-1,2), np.arange(-1,2)
    x, y = np.meshgrid(x,y)
    xy = np.stack([x,y],-1).reshape(-1,1)
    xy = torch.tensor(xy).cuda(depth_var.device)

    tmp = depth_var.new_ones(18, 480 *640) * xy
    check_offsets = [1,2,3,5,7,20]
    offsets = []
    for offset in check_offsets:
        offsets.append(offset * tmp)
    offsets = torch.cat(offsets, 0)
    c = offsets.shape[0]
    offsets = offsets.reshape(1, c, 480 , 640)

    for i in range(iter_):
        if i == 0:

            # (depth, normal, image, conf, confN, K, profiler=None):
            # depth: [B, 1, H, W], normal: [B, 3, H, W], image: [B, 3, H, W], conf: depth confidence [B, 1, H, W], confN: normal confidence
            updated_depth, updated_normal = solver(depth_var, normal_var, img_var, conf_var, conf_var, K_var, offsets=offsets)
        else:
            updated_depth, updated_normal = solver(updated_depth, updated_normal, img_var, conf_var, conf_var, K_var, offsets=offsets)
        # from pdb import set_trace; set_trace()
        cv2.imwrite('/home/jty/mvs/idn-solver/vis/test_depth'+str(i)+'.png', (updated_depth.detach().cpu().numpy()*100.0)[0,0].astype(np.uint8))
        cv2.imwrite('/home/jty/mvs/idn-solver/vis/test_normal'+str(i)+'.png', normal2color(updated_normal.detach().cpu().numpy())[0].transpose(1,2,0))


text:
# Copyright (c) 2012, GPy authors (see AUTHORS.txt).
# Licensed under the BSD 3-clause license (see LICENSE.txt)

import numpy as np
from ..util.warping_functions import *
from ..core import GP
from .. import likelihoods
from GPy.util.warping_functions import TanhWarpingFunction_d
from GPy import kern

class WarpedGP(GP):
    def __init__(self, X, Y, kernel=None, warping_function=None, warping_terms=3):

        if kernel is None:
            kernel = kern.RBF(X.shape[1])

        if warping_function == None:
            self.warping_function = TanhWarpingFunction_d(warping_terms)
            self.warping_params = (np.random.randn(self.warping_function.n_terms * 3 + 1,) * 1)
        else:
            self.warping_function = warping_function

        self.scale_data = False
        if self.scale_data:
            Y = self._scale_data(Y)
        self.has_uncertain_inputs = False
        self.Y_untransformed = Y.copy()
        self.predict_in_warped_space = False
        likelihood = likelihoods.Gaussian()

        GP.__init__(self, X, self.transform_data(), likelihood=likelihood, kernel=kernel)
        self.link_parameter(self.warping_function)

    def _scale_data(self, Y):
        self._Ymax = Y.max()
        self._Ymin = Y.min()
        return (Y - self._Ymin) / (self._Ymax - self._Ymin) - 0.5

    def _unscale_data(self, Y):
        return (Y + 0.5) * (self._Ymax - self._Ymin) + self._Ymin

    def parameters_changed(self):
        self.Y[:] = self.transform_data()
        super(WarpedGP, self).parameters_changed()

        Kiy = self.posterior.woodbury_vector.flatten()

        grad_y = self.warping_function.fgrad_y(self.Y_untransformed)
        grad_y_psi, grad_psi = self.warping_function.fgrad_y_psi(self.Y_untransformed,
                                                                 return_covar_chain=True)
        djac_dpsi = ((1.0 / grad_y[:, :, None, None]) * grad_y_psi).sum(axis=0).sum(axis=0)
        dquad_dpsi = (Kiy[:, None, None, None] * grad_psi).sum(axis=0).sum(axis=0)

        warping_grads = -dquad_dpsi + djac_dpsi

        self.warping_function.psi.gradient[:] = warping_grads[:, :-1]
        self.warping_function.d.gradient[:] = warping_grads[0, -1]

    def transform_data(self):
        Y = self.warping_function.f(self.Y_untransformed.copy()).copy()
        return Y

    def log_likelihood(self):
        ll = GP.log_likelihood(self)
        jacobian = self.warping_function.fgrad_y(self.Y_untransformed)
        return ll + np.log(jacobian).sum()

    def plot_warping(self):
        self.warping_function.plot(self.Y_untransformed.min(), self.Y_untransformed.max())

    def predict(self, Xnew, which_parts='all', pred_init=None):
        # normalize X values
        # Xnew = (Xnew.copy() - self._Xoffset) / self._Xscale
        mu, var = GP._raw_predict(self, Xnew)

        # now push through likelihood
        mean, var = self.likelihood.predictive_values(mu, var)

        if self.predict_in_warped_space:
            mean = self.warping_function.f_inv(mean,  y=pred_init)
            var = self.warping_function.f_inv(var)

        if self.scale_data:
            mean = self._unscale_data(mean)

        return mean, var

if __name__ == '__main__':
    X = np.random.randn(100, 1)
    Y = np.sin(X) + np.random.randn(100, 1)*0.05

    m = WarpedGP(X, Y)

max_stars_repo_name: strongh/GPy

max_issues_repo_name: strongh/GPy

max_forks_repo_name: strongh/GPy


text:
#!/usr/bin/env python3
"""
ROS driver for the subway car photomosaic task,
original code authored by
Lauren Krieger and Bennedict Soesanto
lkrieg@uw.edu   /  ben2111@uw.edu
"""

import cv2
import rospy
import numpy as np
from sensor_msgs.msg import CompressedImage
from std_msgs.msg import Empty
from subway_car.Photomosaic import Photomosaic
from nautilus_utils.CamHolder import CamHolder

# Static Variables
cam1 = '/nautilus/cameras/stream'
# cam1 = 'out/img/compressed'
# cam2 = 'out/img/compressed2'
button = '/nautilus/controls/signal'

output_path = '/home/uwrov/Desktop/out.png'

# Mutable Variables
cameras = CamHolder()
output = Photomosaic()

def main():
    rospy.init_node('subway_car_runner')
    rospy.Subscriber(cam1, CompressedImage, update_frame, callback_args=(1))
#     rospy.Subscriber(cam2, CompressedImage, update_frame, callback_args=(2))
    rospy.Subscriber(button, Empty, snapshot_fn)

    rospy.on_shutdown(shutdown_fn)

    while not rospy.is_shutdown():
        if output.is_finished():
            export_output(output.get_output())
        rospy.sleep(.5)

def snapshot_fn(msg):
    cap = cameras.get_frame(1) if msg.data[0] == 'o' else cameras.get_frame(2)
    if cap is not None:
        print('adding frame', output.imageIndex)
        cv2.imwrite('/home/uwrov/Desktop/out' + str(output.imageIndex) + '.png', cap)

        output.add(cap)
    return

def update_frame(msg, target_cam):
    np_arr = np.fromstring(msg.data, np.uint8)
    image_np = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)
    # print(image_np)
    if target_cam == 1:
        cameras.update_frame(image_np, 1)
    else:
        cameras.update_frame(image_np, 2)

def export_output(output):
    cv2.imwrite(output_path, output.get_output())
    print("done!")
    rospy.signal_shutdown("finished executing")

def shutdown_fn():
    print("shutting down")

if __name__ == '__main__':
    main()


text:
import numpy as np
import warnings
from scipy import signal
from enum import Enum
from PIL import Image
import cv2
class Filter(Enum):
	UNIFORM = 0
	GAUSSIAN = 1
def _initial_check(GT, P):
	assert GT.shape == P.shape, "Supplied images have different sizes" + \
		str(GT.shape) + "and" + str(P.shape)
	if GT.dtype != P.dtype:
		msg = "Supplied images have different dtypes" + \
			str(GT.dtype) + "and" + str(P.dtype)
		warnings.warn(msg)
	if len(GT.shape) == 2:
		GT = GT[:, :, np.newaxis]
		P = P[:, :, np.newaxis]
	return GT.astype(np.float64), P.astype(np.float64)

def fspecial(fltr, ws, **kwargs):
	if fltr == Filter.UNIFORM:
		return np.ones((ws, ws))/ ws ** 2
	elif fltr == Filter.GAUSSIAN:
		x, y = np.mgrid[-ws//2 + 1:ws//2 + 1, -ws//2 + 1:ws//2 + 1]
		g = np.exp(-((x ** 2 + y ** 2)/(2.0 * kwargs['sigma'] ** 2)))
		g[ g < np.finfo(g.dtype).eps * g.max() ] = 0
		assert g.shape == (ws, ws)
		den = g.sum()
		if den !=0:
			g/=den
		return g
	return None

def _get_sums(GT,P,win,mode='same'):
	mu1,mu2 = (filter2(GT,win,mode),filter2(P,win,mode))
	return mu1 * mu1, mu2 * mu2, mu1 * mu2

def _get_sigmas(GT,P,win,mode='same',**kwargs):
	if 'sums' in kwargs:
		GT_sum_sq,P_sum_sq,GT_P_sum_mul = kwargs['sums']
	else:
		GT_sum_sq,P_sum_sq,GT_P_sum_mul = _get_sums(GT, P, win, mode)

	return filter2(GT * GT,win,mode)  - GT_sum_sq,\
			filter2(P * P,win,mode)  - P_sum_sq, \
			filter2(GT * P,win,mode) - GT_P_sum_mul

def filter2(img,fltr,mode='same'):
	return signal.convolve2d(img, np.rot90(fltr,2), mode=mode)

def _power_complex(a,b):
	return a.astype('complex') ** b

def _replace_value(array,value,replace_with):
	array[array == value] = replace_with
	return array

def _compute_bef(im, block_size=8):
	"""Calculates Blocking Effect Factor (BEF) for a given grayscale/one channel image

	C. Yim and A. C. Bovik, "Quality Assessment of Deblocked Images," in IEEE Transactions on Image Processing,
		vol. 20, no. 1, pp. 88-98, Jan. 2011.

	:param im: input image (numpy ndarray)
	:param block_size: Size of the block over which DCT was performed during compression
	:return: float -- bef.
	"""
	if len(im.shape) == 3:
		height, width, channels = im.shape
	elif len(im.shape) == 2:
		height, width = im.shape
		channels = 1
	else:
		raise ValueError("Not a 1-channel/3-channel grayscale image")

	if channels > 1:
		raise ValueError("Not for color images")

	h = np.array(range(0, width - 1))
	h_b = np.array(range(block_size - 1, width - 1, block_size))
	h_bc = np.array(list(set(h).symmetric_difference(h_b)))

	v = np.array(range(0, height - 1))
	v_b = np.array(range(block_size - 1, height - 1, block_size))
	v_bc = np.array(list(set(v).symmetric_difference(v_b)))

	d_b = 0
	d_bc = 0

	# h_b for loop
	for i in list(h_b):
		diff = im[:, i] - im[:, i + 1]
		d_b += np.sum(np.square(diff))

	# h_bc for loop
	for i in list(h_bc):
		diff = im[:, i] - im[:, i + 1]
		d_bc += np.sum(np.square(diff))

	# v_b for loop
	for j in list(v_b):
		diff = im[j, :] - im[j + 1, :]
		d_b += np.sum(np.square(diff))

	# V_bc for loop
	for j in list(v_bc):
		diff = im[j, :] - im[j+1, :]
		d_bc += np.sum(np.square(diff))

	# N code
	n_hb = height * (width / block_size) - 1
	n_hbc = (height * (width - 1)) - n_hb
	n_vb = width * (height / block_size) - 1
	n_vbc = (width * (height - 1)) - n_vb

	# D code
	d_b /= (n_hb + n_vb)
	d_bc /= (n_hbc + n_vbc)

	# Log
	if d_b > d_bc:
		t = log2(block_size) / log2(min(height, width))
	else:
		t = 0

	# BEF
	bef = t * (d_b - d_bc)

	return bef

def filter2(img,fltr,mode='same'):
	return signal.convolve2d(img, np.rot90(fltr,2), mode=mode)

def imresize(arr, size):
	arr_sw = cv2.resize(arr, size)
	return np.array(arr_sw)
if __name__ == "__main__":
	img = Image.open("/home/minhhoang/Desktop/sewar-master/sewar/tests/res/lena512color_constant.tiff")
	abcd = np.array(img.resize((123,123)))
	print(abcd.shape)

max_stars_repo_name: Mrmachiner/Image-Super-resolution

max_issues_repo_name: Mrmachiner/Image-Super-resolution

max_forks_repo_name: Mrmachiner/Image-Super-resolution


text:
from __future__ import print_function, division
import os
import pandas as pd
from skimage import io, transform, exposure, img_as_uint, img_as_float
import numpy as np
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("DataPath", help="Please provide the path of csv file.")
parser.add_argument('-o',"--OutputPath", help="Please provide the path of the output directory.")
args = parser.parse_args()

output_dir = ""

filepath = os.path.join(args.DataPath,)
if args.OutputPath:
	output_dir = args.OutputPath

labels = pd.read_csv(filepath,header=None)
labels.columns = ["building","barren_land","trees","grassland","road","water"]

labels_rows, labels_cols = labels.shape

output_path = os.path.join(output_dir,"annotations.csv")

labels.to_csv(output_path)