Switch from None to raising exceptions in HCV.

Author: donkirkby

.idea/pyvdrm.iml

@@ -5,7 +5,7 @@
 from functools import reduce, total_ordering
 from pyparsing import (Literal, nums, Word, Forward, Optional, Regex,
                        infixNotation, delimitedList, opAssoc, ParseException)
-from pyvdrm.drm import AsiExpr, AsiBinaryExpr, AsiUnaryExpr, DRMParser
+from pyvdrm.drm import AsiExpr, AsiBinaryExpr, DRMParser
 from pyvdrm.vcf import MutationSet
 
 
@@ -79,7 +79,7 @@ class Negate(AsiExpr):
     def __call__(self, mutations):
         child_score = self.children[0](mutations)
         if child_score is None:
-            return Score(True, []) # TODO: propagate negative residues
+            return Score(True, [])  # TODO: propagate negative residues
         return Score(not child_score.score, child_score.residues)
 
 

pyvdrm/asi2.py

@@ -8,6 +8,10 @@ class AsiParseError(Exception):
     pass
 
 
+class MissingPositionError(Exception):
+    pass
+
+
 class DRMParser(metaclass=ABCMeta):
     """abstract class for DRM rule parsers/evaluators"""
 

pyvdrm/drm.py

@@ -4,46 +4,22 @@
 
 from functools import reduce, total_ordering
 from pyparsing import (Literal, nums, Word, Forward, Optional, Regex,
-                       infixNotation, delimitedList, opAssoc, alphas, ParseException)
-from pyvdrm.drm import AsiExpr, AsiBinaryExpr, AsiUnaryExpr, DRMParser
+                       infixNotation, delimitedList, opAssoc, ParseException)
+
+from pyvdrm.drm import MissingPositionError
+from pyvdrm.drm import AsiExpr, AsiBinaryExpr, DRMParser
 from pyvdrm.vcf import MutationSet
 
+
 def update_flags(fst, snd):
     for k in snd:
         if k in fst:
             fst[k].append(snd[k])
         else:
-            fst[k] = snd[k] # this chould be achieved with a defaultdict
+            fst[k] = snd[k]  # this could be achieved with a defaultdict
     return fst
 
 
-def maybe_foldl(func, noneable):
-    """Safely fold a function over a potentially empty list of
-    potentially null values"""
-    if noneable is None:
-        return None
-    clean = [x for x in noneable if x is not None]
-    if not clean:
-        return None
-    return reduce(func, clean)
-
-
-def maybe_map(func, noneable):
-    if noneable is None:
-        return None
-    r_list = []
-    for x in noneable:
-        if x is None:
-            continue
-        result = func(x)
-        if result is None:
-            continue
-        r_list.append(result)
-    if not r_list:
-        return None
-    return r_list
-
-
 @total_ordering
 class Score(object):
     """Encapsulate a score and the residues that support it"""
@@ -65,13 +41,15 @@ def __init__(self, score, residues, flags=None):
 
     def __add__(self, other):
         flags = update_flags(self.flags, other.flags)
-        return Score(self.score + other.score, self.residues | other.residues,
-                flags)
+        return Score(self.score + other.score,
+                     self.residues | other.residues,
+                     flags)
 
     def __sub__(self, other):
         flags = update_flags(self.flags, other.flags)
-        return Score(self.score - other.score, self.residues | other.residues,
-                flags)
+        return Score(self.score - other.score,
+                     self.residues | other.residues,
+                     flags)
 
     def __repr__(self):
         return "Score({!r}, {!r})".format(self.score, self.residues)
@@ -166,11 +144,11 @@ def __call__(self, mutations):
         if len(self.children) == 4:
             operation, _, flag, _ = self.children
             flags[flag] = []
-            score = 0 # should be None
+            score = 0  # should be None
 
         elif len(self.children) == 3:
             operation, minus, score = self.children
-            if minus != '-': # this is parsing the expression twice, refactor
+            if minus != '-':  # this is parsing the expression twice, refactor
                 raise ValueError
             score = -1 * int(score)
 
@@ -197,10 +175,22 @@ class ScoreList(AsiExpr):
     def __call__(self, mutations):
         operation, *rest = self.children
         if operation == 'MAX':
-            return maybe_foldl(max, [f(mutations) for f in rest])
-
-        # the default operation is sum
-        return maybe_foldl(lambda x, y: x+y, [f(mutations) for f in self.children])
+            terms = rest
+            func = max
+        else:
+            # the default operation is sum
+            terms = self.children
+            func = sum
+        scores = [f(mutations) for f in terms]
+        matched_scores = [score.score for score in scores if score.score]
+        residues = reduce(lambda x, y: x | y,
+                          (score.residues for score in scores))
+        flags = {}
+        for score in scores:
+            flags.update(score.flags)
+        return Score(bool(matched_scores) and func(matched_scores),
+                     residues,
+                     flags)
 
 
 class SelectFrom(AsiExpr):
@@ -215,14 +205,12 @@ def __call__(self, mutations):
         operation, *rest = self.children
         # the head of the arg list must be an equality expression
 
-        scored = list(maybe_map(lambda f: f(mutations), rest))
-        passing = len(scored) 
+        scored = [f(mutations) for f in rest]
+        passing = sum(bool(score.score) for score in scored)
 
-        if operation(passing):
-            return Score(True, maybe_foldl(
-                lambda x, y: x.residues.union(y.residues), scored))
-        else:
-            return None
+        return Score(operation(passing),
+                     reduce(lambda x, y: x | y,
+                            (item.residues for item in scored)))
 
 
 class AsiScoreCond(AsiExpr):
@@ -232,7 +220,7 @@ class AsiScoreCond(AsiExpr):
 
     def __call__(self, args):
         """Score conditions evaluate a list of expressions and sum scores"""
-        return maybe_foldl(lambda x, y: x+y, map(lambda x: x(args), self.children))
+        return sum((f(args) for f in self.children), Score(False, set()))
 
 
 class AsiMutations(object):
@@ -241,21 +229,26 @@ class AsiMutations(object):
     def __init__(self, _label=None, _pos=None, args=None):
         """Initialize set of mutations from a potentially ambiguous residue
         """
-        self.mutations = args and MutationSet(''.join(args))
+        self.mutations = MutationSet(''.join(args))
 
     def __repr__(self):
         if self.mutations is None:
             return "AsiMutations()"
         return "AsiMutations(args={!r})".format(str(self.mutations))
 
     def __call__(self, env):
+        is_found = False
         for mutation_set in env:
+            is_found |= mutation_set.pos == self.mutations.pos
             intersection = self.mutations.mutations & mutation_set.mutations
             if len(intersection) > 0:
                 return Score(True, intersection)
 
-        # the mutationset has no members in the environment
-        return None
+        if not is_found:
+            # Some required positions were not found in the environment.
+            raise MissingPositionError('Missing position {}.'.format(
+                self.mutations.pos))
+        return Score(False, set())
 
 
 class HCVR(DRMParser):
@@ -305,8 +298,8 @@ def parser(self, rule):
         selectstatement = select + select_quantifier + from_ + residue_list
         selectstatement.setParseAction(SelectFrom)
 
-        bool_ = Literal('TRUE').suppress().setParseAction(BoolTrue) |\
-                Literal('FALSE').suppress().setParseAction(BoolFalse)
+        bool_ = (Literal('TRUE').suppress().setParseAction(BoolTrue) |
+                 Literal('FALSE').suppress().setParseAction(BoolFalse))
 
         booleancondition = Forward()
         condition = residue | excludestatement | selectstatement | bool_

pyvdrm/hcvr.py

@@ -1,3 +1,4 @@
+import os
 import unittest
 
 from pyparsing import ParseException
@@ -128,7 +129,9 @@ def test_parse_exception_multiline(self):
 
 class TestActualRules(unittest.TestCase):
     def test_hivdb_rules_parse(self):
-        for line in open("pyvdrm/tests/HIVDB.rules"):
+        folder = os.path.dirname(__file__)
+        rules_file = os.path.join(folder, 'HIVDB.rules')
+        for line in open(rules_file):
             r = ASI2(line)
             self.assertEqual(line, r.rule)