diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml new file mode 100644 index 0000000..be4f9a9 --- /dev/null +++ b/.github/workflows/main.yml @@ -0,0 +1,24 @@ +name: main +on: push +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"] + steps: + - uses: actions/checkout@v3 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install pytest + + - name: Run tests + run: pytest -v pyvdrm/tests/ diff --git a/pyvdrm/asi2.py b/pyvdrm/asi2.py index 0f4d8da..778facb 100644 --- a/pyvdrm/asi2.py +++ b/pyvdrm/asi2.py @@ -4,7 +4,7 @@ from functools import reduce, total_ordering from pyparsing import (Literal, nums, Word, Forward, Optional, Regex, - infixNotation, delimitedList, opAssoc, ParseException) + infix_notation, DelimitedList, opAssoc, ParseException) from pyvdrm.drm import AsiExpr, AsiMultipleExpr, DRMParser, MissingPositionError from pyvdrm.vcf import MutationSet @@ -230,10 +230,10 @@ def parser(self, rule): integer = Word(nums) mutation = Optional(Regex(r'[A-Z]')) + integer + Regex(r'[diA-Z]+') - mutation.setParseAction(AsiMutations) + mutation.set_parse_action(AsiMutations) not_ = Literal('NOT').suppress() + mutation - not_.setParseAction(Negate) + not_.set_parse_action(Negate) residue = mutation | not_ # integer + l_par + not_ + Regex(r'[A-Z]+') + r_par @@ -244,40 +244,39 @@ def parser(self, rule): quantifier = exactly | atleast | notmorethan inequality = quantifier + integer - inequality.setParseAction(EqualityExpr) + inequality.set_parse_action(EqualityExpr) - select_quantifier = infixNotation(inequality, + select_quantifier = infix_notation(inequality, [(and_, 2, opAssoc.LEFT, AndExpr), (or_, 2, opAssoc.LEFT, OrExpr)]) - residue_list = l_par + delimitedList(residue) + r_par + residue_list = l_par + DelimitedList(residue) + r_par # so selectstatement.eval :: [Mutation] -> Maybe Bool selectstatement = select + select_quantifier + from_ + residue_list - selectstatement.setParseAction(SelectFrom) - + selectstatement.set_parse_action(SelectFrom) booleancondition = Forward() condition = residue | excludestatement | selectstatement - booleancondition << infixNotation(condition, + booleancondition << infix_notation(condition, [(and_, 2, opAssoc.LEFT, AndExpr), (or_, 2, opAssoc.LEFT, OrExpr)]) | condition scoreitem = booleancondition + mapper + Optional(Literal('-')) + integer - scoreitem.setParseAction(ScoreExpr) - scorelist = max_ + l_par + delimitedList(scoreitem) + r_par |\ - delimitedList(scoreitem) - scorelist.setParseAction(ScoreList) + scoreitem.set_parse_action(ScoreExpr) + scorelist = max_ + l_par + DelimitedList(scoreitem) + r_par |\ + DelimitedList(scoreitem) + scorelist.set_parse_action(ScoreList) scorecondition = Literal('SCORE FROM').suppress() +\ - l_par + delimitedList(scorelist) + r_par + l_par + DelimitedList(scorelist) + r_par - scorecondition.setParseAction(AsiScoreCond) + scorecondition.set_parse_action(AsiScoreCond) statement = booleancondition | scorecondition try: - return statement.parseString(rule) + return statement.parse_string(rule) except ParseException as ex: - ex.msg = 'Error in ASI2: ' + ex.markInputline() + ex.msg = 'Error in ASI2: ' + ex.mark_input_line() raise diff --git a/pyvdrm/hcvr.py b/pyvdrm/hcvr.py index a9a2c4a..1a8d3e8 100644 --- a/pyvdrm/hcvr.py +++ b/pyvdrm/hcvr.py @@ -3,8 +3,8 @@ """ from functools import reduce, total_ordering -from pyparsing import (Literal, nums, Word, Forward, Optional, Regex, - infixNotation, delimitedList, opAssoc, ParseException) +from pyparsing import (DelimitedList, Literal, nums, Word, Forward, Optional, Regex, + infix_notation, opAssoc, ParseException) from pyvdrm.drm import MissingPositionError from pyvdrm.drm import AsiExpr, AsiMultipleExpr, DRMParser @@ -274,7 +274,7 @@ def parser(self, rule): integer = Word(nums) residue = Optional(Regex(r'[A-Z]')) + integer + Regex(r'\!?[diA-Z]+') - residue.setParseAction(AsiMutations) + residue.set_parse_action(AsiMutations) # Syntax of expressions excludestatement = except_ + residue @@ -282,44 +282,44 @@ def parser(self, rule): quantifier = exactly | atleast | notmorethan tropical = max_ | min_ inequality = quantifier + integer - inequality.setParseAction(EqualityExpr) + inequality.set_parse_action(EqualityExpr) - select_quantifier = infixNotation(inequality, + select_quantifier = infix_notation(inequality, [(and_, 2, opAssoc.LEFT, AndExpr), (or_, 2, opAssoc.LEFT, OrExpr)]) - residue_list = l_par + delimitedList(residue) + r_par + residue_list = l_par + DelimitedList(residue) + r_par # so selectstatement.eval :: [Mutation] -> Maybe Bool selectstatement = select + select_quantifier + from_ + residue_list - selectstatement.setParseAction(SelectFrom) + selectstatement.set_parse_action(SelectFrom) - bool_ = (Literal('TRUE').suppress().setParseAction(BoolTrue) | - Literal('FALSE').suppress().setParseAction(BoolFalse)) + bool_ = (Literal('TRUE').suppress().set_parse_action(BoolTrue) | + Literal('FALSE').suppress().set_parse_action(BoolFalse)) booleancondition = Forward() condition = residue | excludestatement | selectstatement | bool_ - booleancondition << infixNotation(condition, + booleancondition << infix_notation(condition, [(and_, 2, opAssoc.LEFT, AndExpr), (or_, 2, opAssoc.LEFT, OrExpr)]) | condition score = Optional(Literal('-')) + integer | quote + Regex(r'[a-zA-Z0-9 _]+') + quote scoreitem = booleancondition + mapper + score - scoreitem.setParseAction(ScoreExpr) - scorelist = tropical + l_par + delimitedList(scoreitem) + r_par |\ - delimitedList(scoreitem) - scorelist.setParseAction(ScoreList) + scoreitem.set_parse_action(ScoreExpr) + scorelist = tropical + l_par + DelimitedList(scoreitem) + r_par |\ + DelimitedList(scoreitem) + scorelist.set_parse_action(ScoreList) scorecondition = Literal('SCORE FROM').suppress() +\ - l_par + delimitedList(scorelist) + r_par + l_par + DelimitedList(scorelist) + r_par - scorecondition.setParseAction(AsiScoreCond) + scorecondition.set_parse_action(AsiScoreCond) statement = booleancondition | scorecondition try: - return statement.parseString(rule) + return statement.parse_string(rule) except ParseException as ex: - ex.msg = 'Error in HCVR: ' + ex.markInputline() + ex.msg = 'Error in HCVR: ' + ex.mark_input_line() raise diff --git a/pyvdrm/tests/test_integration.py b/pyvdrm/tests/test_integration.py new file mode 100644 index 0000000..a2a2222 --- /dev/null +++ b/pyvdrm/tests/test_integration.py @@ -0,0 +1,196 @@ +""" +Integration tests for pyvdrm - testing end-to-end usage examples +""" +import unittest + +from pyvdrm.asi2 import ASI2 +from pyvdrm.hcvr import HCVR +from pyvdrm.vcf import Mutation, VariantCalls + + +class TestASI2Integration(unittest.TestCase): + """Test ASI2 algorithm with real-world examples""" + + def test_readme_example(self): + """Test the example similar to README""" + # Define a rule + rule = ASI2("SCORE FROM (MAX (100T => 20, 282N => 15))") + + # Evaluate against mutations + score = rule(VariantCalls("100d 282N")) + + self.assertEqual(15, score) + + def test_basic_scoring_rule(self): + """Test a basic scoring rule""" + rule = ASI2("SCORE FROM (65R => 20, 74V => 20, 184VI => 20)") + + # Test with one mutation + self.assertEqual(20, rule(VariantCalls("65R 74d 184d"))) + + # Test with multiple mutations + self.assertEqual(40, rule(VariantCalls("65R 74V 184d"))) + + # Test with all mutations + self.assertEqual(60, rule(VariantCalls("65R 74V 184V"))) + + def test_boolean_and_rule(self): + """Test boolean AND logic""" + rule = ASI2("100G AND 200T") + + self.assertTrue(rule(VariantCalls("100G 200T"))) + self.assertFalse(rule(VariantCalls("100G 200d"))) + self.assertFalse(rule(VariantCalls("100d 200T"))) + + def test_boolean_or_rule(self): + """Test boolean OR logic""" + rule = ASI2("100G OR 200T") + + self.assertTrue(rule(VariantCalls("100G 200d"))) + self.assertTrue(rule(VariantCalls("100d 200T"))) + self.assertTrue(rule(VariantCalls("100G 200T"))) + self.assertFalse(rule(VariantCalls("100d 200d"))) + + def test_select_atleast(self): + """Test SELECT ATLEAST operator""" + rule = ASI2("SELECT ATLEAST 2 FROM (41L, 67N, 70R)") + + self.assertTrue(rule(VariantCalls("41L 67N 70d"))) + self.assertTrue(rule(VariantCalls("41L 67N 70R"))) + self.assertFalse(rule(VariantCalls("41L 67d 70d"))) + + def test_max_operator(self): + """Test MAX operator in scoring""" + rule = ASI2("SCORE FROM (MAX (100P => 40, 100E => 30, 100H => 15))") + + # Should take the maximum score + self.assertEqual(30, rule(VariantCalls("100E"))) + self.assertEqual(40, rule(VariantCalls("100P"))) + self.assertEqual(15, rule(VariantCalls("100H"))) + + def test_mutation_from_sequence(self): + """Test creating mutations from aligned sequences""" + reference = "ACHE" + sample = "ICRE" + + calls = VariantCalls(reference=reference, sample=sample) + + # Should have mutations at positions 1 and 3 + self.assertEqual(4, len(calls)) # All positions present + + # Create a rule to check specific mutations + rule = ASI2("1I AND 3R") + self.assertTrue(rule(calls)) + + +class TestHCVRIntegration(unittest.TestCase): + """Test HCVR algorithm with real-world examples""" + + def test_basic_scoring(self): + """Test basic HCVR scoring""" + rule = HCVR("SCORE FROM (100G => 10, 200T => 20)") + + self.assertEqual(10, rule(VariantCalls("100G 200d"))) + self.assertEqual(20, rule(VariantCalls("100d 200T"))) + self.assertEqual(30, rule(VariantCalls("100G 200T"))) + + def test_negative_mutations(self): + """Test NOT operator (negative mutations)""" + rule = HCVR("SCORE FROM (100!G => 10)") + + # Score when NOT G + self.assertEqual(10, rule(VariantCalls("100T 200d"))) + + # No score when it IS G + self.assertEqual(0, rule(VariantCalls("100G 200d"))) + + def test_min_operator(self): + """Test MIN operator""" + rule = HCVR("SCORE FROM (MIN (100G => 40, 100E => 30, 100H => 15))") + + # Should take the minimum score + self.assertEqual(15, rule(VariantCalls("100H"))) + + def test_multiple_variants_at_position(self): + """Test mutations with multiple variants at a position""" + rule = HCVR("SCORE FROM (100GE => 20)") + + # Should match if either G or E + self.assertEqual(20, rule(VariantCalls("100G"))) + self.assertEqual(20, rule(VariantCalls("100E"))) + self.assertEqual(0, rule(VariantCalls("100T"))) + + +class TestMutationAPI(unittest.TestCase): + """Test the Mutation and MutationSet API""" + + def test_mutation_creation(self): + """Test creating mutations in different ways""" + # From string + m1 = Mutation("Q80K") + self.assertEqual("Q", m1.wildtype) + self.assertEqual(80, m1.pos) + self.assertEqual("K", m1.variant) + + # From parameters + m2 = Mutation(wildtype="Q", pos=80, variant="K") + self.assertEqual(m1, m2) + + # Without wildtype + m3 = Mutation("80K") + self.assertIsNone(m3.wildtype) + self.assertEqual(80, m3.pos) + + def test_variant_calls_from_text(self): + """Test creating VariantCalls from text""" + calls = VariantCalls("A1I H3R E4D") + + self.assertEqual(3, len(calls)) + self.assertIn("A1I", str(calls)) + self.assertIn("H3R", str(calls)) + self.assertIn("E4D", str(calls)) + + def test_variant_calls_from_sequences(self): + """Test creating VariantCalls from sequences""" + reference = "AHEC" + sample = "IRDC" + + calls = VariantCalls(reference=reference, sample=sample) + + # All positions should be present + self.assertEqual(4, len(calls)) + + # Convert to string and check + calls_str = str(calls) + self.assertIn("A1I", calls_str) + self.assertIn("H2R", calls_str) + self.assertIn("E3D", calls_str) + self.assertIn("C4C", calls_str) + + +class TestErrorHandling(unittest.TestCase): + """Test error handling and validation""" + + def test_invalid_mutation_format(self): + """Test that invalid mutation formats raise errors""" + with self.assertRaises(ValueError): + Mutation("!20A") + + with self.assertRaises(ValueError): + Mutation("Q20") # Missing variant + + def test_invalid_rule_syntax(self): + """Test that invalid rule syntax raises ParseException""" + from pyparsing import ParseException + + with self.assertRaises(ParseException): + ASI2("SCORE FROM ( 10R => 2;0 )") + + def test_sequence_length_mismatch(self): + """Test that mismatched sequence lengths raise error""" + with self.assertRaises(ValueError): + VariantCalls(reference="ACE", sample="ACHED") + + +if __name__ == '__main__': + unittest.main()