enemy management v2.0

app/models/attribute.py

@@ -2,6 +2,6 @@ from pydantic import BaseModel
 from typing import Optional, Union
 
 class Attribute(BaseModel):
-    value: Optional[Union[str,int]]
+    value: Optional[Union[str,int,list]]
     type: Optional[str]
     id: str

app/models/bundle.py

@@ -59,3 +59,13 @@ class Bundle(BaseModel):
 
     def ordered_items(self) -> List[Item]:
         return sorted(self.items, key=lambda item: item.position)
+
+    # are there enemys in the bundle?
+    def enemy_pair_count(self, pair: List[Item]) -> int:
+        pair_count = 0
+
+        for item in self.items:
+            if pair in item.enemy_pairs():
+                pair_count += 1
+
+        return pair_count

app/models/constraints/enemy_pair_constraint.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+from typing import List
+
+from models.constraints.generic_constraint import *
+
+class EnemyPairConstraint(GenericConstraint):
+    @classmethod
+    def create(cls: Type[_T], pair: List[int]) -> _T:
+        return cls(
+            minimum=0,
+            maximum=0,
+            reference_attribute=Attribute(
+                value=pair,
+                type='enemy_pair',
+                id='enemy_pair'
+            )
+        )
+
+    def build(self, problem_handler: Problem, **_) -> None:
+        logging.info('Enemy Pair Constraint Generating...')
+
+        pair = self.reference_attribute.value
+        problem_handler.problem += lpSum(
+            [
+                bundle.enemy_pair_count(pair) * problem_handler.solver_bundles_var[bundle.id]
+                for bundle in problem_handler.bundles
+            ] + [
+                (pair in item.enemy_pairs()).real * problem_handler.solver_items_var[item.id]
+                for item in problem_handler.items
+            ]
+        ) <= 1, f'Enemy Pair constraint for pair: {pair}'

app/models/item.py

@@ -1,5 +1,6 @@
+import logging
 from pydantic import BaseModel, validator
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 from models.attribute import Attribute
 
@@ -61,3 +62,15 @@ class Item(BaseModel):
             total += self.irf(solver_run.irt_model, target.theta)
 
         return total
+
+    def enemy_pairs(self, sort: bool = True) -> List[List[int]]:
+        pairs = []
+
+        for enemy_id in self.enemies:
+            pair = [self.id, enemy_id]
+
+            if sort: pair.sort()
+
+            pairs.append(pair)
+
+        return pairs

app/models/problem.py

@@ -43,59 +43,64 @@ class Problem(BaseModel):
 
     def solve(self, solver_run: SolverRun, enemy_ids: List[int] = []) -> LpProblem:
         logging.info('solving problem...')
+        self.problem.solve()
+
+        # NOTICE: Legacy enemies implementation
+        # leaving this in, just in case the current impl fails to function
+        # and we need an immediate solution
         # if we allow enemies, go through the normal solving process
-        if solver_run.allow_enemies:
+        # if solver_run.allow_enemies:
-            logging.info('enemes allowed, so just solving')
+        #     logging.info('enemes allowed, so just solving')
-            self.problem.solve()
+        #     self.problem.solve()
-        # otherwise begin the process of filtering enemies
+        # # otherwise begin the process of filtering enemies
-        else:
+        # else:
-            self.problem.solve()
+        #     self.problem.solve()
 
-            # however, if the solve was infeasible, kick it back
+        #     # however, if the solve was infeasible, kick it back
-            # to the normal process
+        #     # to the normal process
-            if LpStatus[self.problem.status] == 'Infeasible':
+        #     if LpStatus[self.problem.status] == 'Infeasible':
-                return self.problem
+        #         return self.problem
-            # otherwise continue
+        #     # otherwise continue
-            else:
+        #     else:
-                # get items from solution
+        #         # get items from solution
-                solved_items, _ = service_helper.solution_items(self.problem.variables(), solver_run)
+        #         solved_items, _ = service_helper.solution_items(self.problem.variables(), solver_run)
 
-                # sacred items will remain the same (with new items added each run) between solve attempts
+        #         # sacred items will remain the same (with new items added each run) between solve attempts
-                # but new enemies will be appended
+        #         # but new enemies will be appended
-                sacred_ids, new_enemy_ids = sanctify(solved_items)
+        #         sacred_ids, new_enemy_ids = sanctify(solved_items)
 
-                # the current solve run found new enemies
+        #         # the current solve run found new enemies
-                if new_enemy_ids:
+        #         if new_enemy_ids:
-                    logging.info('enemies found, adding constraints...')
+        #             logging.info('enemies found, adding constraints...')
 
-                    # append the new enemies to the enemies_id list
+        #             # append the new enemies to the enemies_id list
-                    enemy_ids = list(set(enemy_ids+new_enemy_ids))
+        #             enemy_ids = list(set(enemy_ids+new_enemy_ids))
 
-                    # remove old enemy/sacred constraints
+        #             # remove old enemy/sacred constraints
-                    if 'Exclude_enemy_items' in self.problem.constraints.keys(): self.problem.constraints.pop('Exclude_enemy_items')
+        #             if 'Exclude_enemy_items' in self.problem.constraints.keys(): self.problem.constraints.pop('Exclude_enemy_items')
-                    if 'Include_sacred_items' in self.problem.constraints.keys(): self.problem.constraints.pop('Include_sacred_items')
+        #             if 'Include_sacred_items' in self.problem.constraints.keys(): self.problem.constraints.pop('Include_sacred_items')
 
-                    # add constraint to not allow enemy items
+        #             # add constraint to not allow enemy items
-                    self.problem += lpSum([
+        #             self.problem += lpSum([
-                        len(bundle.find_items(enemy_ids)) * self.solver_bundles_var[bundle.id]
+        #                 len(bundle.find_items(enemy_ids)) * self.solver_bundles_var[bundle.id]
-                        for bundle in self.bundles
+        #                 for bundle in self.bundles
-                    ] + [
+        #             ] + [
-                        (item.id in enemy_ids) * self.solver_items_var[item.id]
+        #                 (item.id in enemy_ids) * self.solver_items_var[item.id]
-                        for item in self.items
+        #                 for item in self.items
-                    ]) == 0, 'Exclude enemy items'
+        #             ]) == 0, 'Exclude enemy items'
 
-                    # add constraint to use sacred items
+        #             # add constraint to use sacred items
-                    self.problem += lpSum([
+        #             self.problem += lpSum([
-                        len(bundle.find_items(sacred_ids)) * self.solver_bundles_var[bundle.id]
+        #                 len(bundle.find_items(sacred_ids)) * self.solver_bundles_var[bundle.id]
-                        for bundle in self.bundles
+        #                 for bundle in self.bundles
-                    ] + [
+        #             ] + [
-                        (item.id in sacred_ids) * self.solver_items_var[item.id]
+        #                 (item.id in sacred_ids) * self.solver_items_var[item.id]
-                        for item in self.items
+        #                 for item in self.items
-                    ]) == len(sacred_ids), 'Include sacred items'
+        #             ]) == len(sacred_ids), 'Include sacred items'
 
-                    # recursively solve until no enemies exist or infeasible
+        #             # recursively solve until no enemies exist or infeasible
-                    logging.info('recursively solving...')
+        #             logging.info('recursively solving...')
-                    self.solve(solver_run)
+        #             self.solve(solver_run)
 
         return self.problem
 

app/models/solver_run.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
-from typing import TYPE_CHECKING, List, Literal, Optional, Union, TypeVar
+from typing import TYPE_CHECKING, List, Literal, Optional, Tuple, Union, TypeVar
 from pulp import lpSum
 from pydantic import BaseModel
 
-
+import itertools
 import logging
 
 from models.item import Item
@@ -12,6 +12,7 @@ from models.constraints.metadata_constraint import MetadataConstraint
 from models.constraints.bundle_constraint import BundleConstraint
 from models.constraints.form_uniqueness_constraint import FormUniquenessConstraint
 from models.constraints.total_form_items_constraint import TotalFormItemsConstraint
+from models.constraints.enemy_pair_constraint import EnemyPairConstraint
 from models.irt_model import IRTModel
 from models.bundle import Bundle
 from models.objective_function import ObjectiveFunction
@@ -45,13 +46,6 @@ class SolverRun(BaseModel):
         # ideally we'd change the payload to determine what type it is
         constraints: [ConstraintType] = []
 
-        # total form items
-        constraints.append(TotalFormItemsConstraint.create(self.total_form_items))
-
-        # ensure form uniqueness
-        if self.advanced_options.ensure_form_uniqueness:
-            constraints.append(FormUniquenessConstraint.create(self.total_form_items - 1))
-
         # repackage to create appropriate constraint types
         for constraint in self.constraints:
             if constraint.reference_attribute.type == 'metadata':
@@ -80,6 +74,18 @@ class SolverRun(BaseModel):
         self.items = [item for item in self.items if item not in items]
         return True
 
+    def generate_constraints(self) -> None:
+        # total form items
+        self.constraints.append(TotalFormItemsConstraint.create(self.total_form_items))
+
+        # ensure form uniqueness
+        if self.advanced_options.ensure_form_uniqueness:
+            self.constraints.append(FormUniquenessConstraint.create(self.total_form_items - 1))
+
+        # enemies constraints
+        for pair in self.enemy_pairs():
+            self.constraints.append(EnemyPairConstraint.create(pair))
+
     def generate_bundles(self):
         logging.info('Generating Bundles...')
         # confirms bundle constraints exists
@@ -149,3 +155,13 @@ class SolverRun(BaseModel):
         else:
             return self.items
 
+    def enemy_pairs(self) -> List[List[int]]:
+        pairs = []
+
+        for item in self.items:
+            # add enemy pairs for item to pairs
+            pairs += item.enemy_pairs()
+
+        # remove duplicates
+        pairs.sort()
+        return list(k for k,_ in itertools.groupby(pairs))

app/services/form_generation_service.py

@@ -21,6 +21,7 @@ class FormGenerationService(Base):
         try:
             self.solver_run = self.create_solver_run_from_attributes()
             self.solver_run.generate_bundles()
+            self.solver_run.generate_constraints()
             self.solution = self.generate_solution()
             self.result = self.stream_to_s3_bucket()
         except ItemGenerationError as error: