refactor create constraints for targets

app/helpers/aws_helper.py

@@ -37,6 +37,5 @@ def get_object_tags(key: str, bucket: str) -> list:
     tags = s3.get_object_tagging(Bucket=bucket, Key=key)['TagSet']
     return tags
 
-
 def file_stream_upload(buffer: io.BytesIO, name: str, bucket: str, action: str = None):
     return s3.upload_fileobj(buffer, bucket, name, ExtraArgs={'Tagging': f'action={action}'})

app/helpers/solver_helper.py

@@ -1,133 +0,0 @@
-from pulp import lpSum, LpProblem
-from random import randint, sample
-
-import logging
-
-from helpers.common_helper import *
-
-from models.bundle import Bundle
-from models.solver_run import SolverRun
-from models.item import Item
-
-from lib.errors.item_generation_error import ItemGenerationError
-
-# should probably be factored out into a bundle class method or a method in the solver run
-def build_constraints(solver_run: SolverRun, problem: LpProblem,
-                      items: list[Item], bundles: list[Bundle], selected_items: list[Item], selected_bundles: list[Bundle], current_drift: int) -> LpProblem:
-    logging.info('Creating Constraints...')
-
-    try:
-        total_form_items = solver_run.total_form_items
-        constraints = solver_run.constraints
-
-        for constraint in constraints:
-            attribute = constraint.reference_attribute
-            min = constraint.minimum
-            max = constraint.maximum
-
-            if attribute.type == 'metadata':
-                logging.info('Metadata Constraint Generating...')
-
-                problem += lpSum(
-                    [
-                        len(bundle.items_with_attribute(attribute)) * bundles[bundle.id] for bundle in selected_bundles
-                    ] +
-                    [
-                        item.attribute_exists(attribute).real * items[item.id] for item in selected_items
-                    ]
-                ) >= round(total_form_items * (min / 100)), f'{attribute.id} - {attribute.value} - min'
-
-                problem += lpSum(
-                    [
-                        len(bundle.items_with_attribute(attribute)) * bundles[bundle.id] for bundle in selected_bundles
-                    ] +
-                    [
-                        item.attribute_exists(attribute).real * items[item.id] for item in selected_items
-                    ]
-                ) <= round(total_form_items * (max / 100)), f'{attribute.id} - {attribute.value} - max'
-            elif attribute.type == 'bundle':
-                logging.info('Bundles Constraint Generating...')
-                # TODO: account for many different bundle types, since the id condition in L33 could yield duplicates
-                if selected_bundles != None and len(selected_bundles) > 0:
-                    # make sure the total bundles used in generated form is limited between min-max set
-                    problem += lpSum([
-                        bundles[bundle.id] for bundle in selected_bundles
-                    ]) == randint(int(constraint.minimum),
-                                  int(constraint.maximum))
-
-        logging.info('Constraints Created...')
-
-        # Behold our very own Elastic constraints!
-        for tif_target in solver_run.objective_function.tif_targets:
-            problem += lpSum([
-                bundle.tif(solver_run.irt_model, tif_target.theta)
-                * bundles[bundle.id]
-                for bundle in selected_bundles
-            ] + [
-                item.iif(solver_run, tif_target.theta) *
-                items[item.id]
-                for item in selected_items
-            ]) >= tif_target.minimum(
-            ), f'Min TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%'
-            problem += lpSum([
-                bundle.tif(solver_run.irt_model, tif_target.theta)
-                * bundles[bundle.id]
-                for bundle in selected_bundles
-            ] + [
-                item.iif(solver_run, tif_target.theta) *
-                items[item.id]
-                for item in selected_items
-            ]) <= tif_target.maximum(
-            ), f'Max TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%'
-
-        for tcc_target in solver_run.objective_function.tcc_targets:
-            problem += lpSum([
-                bundle.trf(solver_run.irt_model, tcc_target.theta)
-                * bundles[bundle.id]
-                for bundle in selected_bundles
-            ] + [
-                item.irf(solver_run, tcc_target.theta) *
-                items[item.id]
-                for item in selected_items
-            ]) >= tcc_target.minimum(
-            ), f'Min TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%'
-            problem += lpSum([
-                bundle.trf(solver_run.irt_model, tcc_target.theta)
-                * bundles[bundle.id]
-                for bundle in selected_bundles
-            ] + [
-                item.irf(solver_run, tcc_target.theta) *
-                items[item.id]
-                for item in selected_items
-            ]) <= tcc_target.maximum(
-            ), f'Max TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%'
-
-        return problem
-    except ValueError as error:
-        logging.error(error)
-        raise ItemGenerationError(
-            "Bundle min and/or max larger than bundle amount provided",
-            error.args[0])
-
-# should probably be factored out into a bundle class method or a method in the solver run
-def get_random_bundles(total_form_items: int,
-                       bundles: list[Bundle],
-                       min: int,
-                       max: int,
-                       found_bundles=False) -> list[Bundle]:
-    selected_bundles = None
-    total_bundle_items = 0
-    total_bundles = randint(min, max)
-    logging.info(f'Selecting Bundles (total of {total_bundles})...')
-
-    while found_bundles == False:
-        selected_bundles = sample(bundles, total_bundles)
-        total_bundle_items = sum(bundle.count for bundle in selected_bundles)
-
-        if total_bundle_items <= total_form_items:
-            found_bundles = True
-
-    if found_bundles == True:
-        return selected_bundles
-    else:
-        return get_random_bundles(total_form_items, total_bundles - 1, bundles)

app/models/bundle_constraint.py

@@ -1,11 +1,6 @@
-import logging
-
 from random import randint
 
-from pulp import lpSum
+from models.constraint import *
-
-from models.constraint import Constraint
-from models.problem import Problem
 
 class BundleConstraint(Constraint):
     def build(self, problem_handler: Problem, _) -> None:

app/models/constraint.py

@@ -1,12 +1,14 @@
+import logging
+
+from pulp import lpSum
 from pydantic import BaseModel
+
 from helpers.common_helper import *
 
 from models.attribute import Attribute
+from models.problem import Problem
 
 class Constraint(BaseModel):
     reference_attribute: Attribute
     minimum: float
     maximum: float
-
-    def __init__(self, **data) -> None:
-        super().__init__(**data)

app/models/irt_target_constraint.py

@@ -1,51 +0,0 @@
-from __future__ import annotations
-from typing import TYPE_CHECKING, Optional
-
-from pulp import lpSum
-
-from models import Constraint, Problem, Attribute, Target, Item, Bundle
-
-if TYPE_CHECKING:
-    from models.solver_run import SolverRun
-
-class IrtTargetConstraint(Constraint):
-    reference_attribute: Optional[Attribute]
-    minimum: Optional[float]
-    maximum: Optional[float]
-    target: Target
-    target_type: str
-
-    def build(self, problem_handler: Problem, solver_run: SolverRun):
-        problem_handler.problem += lpSum([
-            self.bundle_irt_function(bundle, solver_run.irt_model, self.target.theta)
-            * problem_handler.solver_bundles_var[bundle.id]
-            for bundle in problem_handler.bundles
-        ] + [
-            self.item_irt_function(item, solver_run.irt_model, self.target.theta) *
-            problem_handler.solver_items_var[item.id]
-            for item in problem_handler.items
-        ]) >= self.target.minimum(
-        ), f'Min {self.target_type} theta({self.target.theta}) at target {self.target.value}'
-
-        problem_handler.problem += lpSum([
-            self.bundle_irt_function(bundle, solver_run.irt_model, self.target.theta)
-            * problem_handler.solver_bundles_var[bundle.id]
-            for bundle in problem_handler.bundles
-        ] + [
-            self.item_irt_function(item, solver_run.irt_model, self.target.theta) *
-            problem_handler.solver_items_var[item.id]
-            for item in problem_handler.items
-        ]) <= self.target.maximum(
-        ), f'Max {self.target_type} theta({self.target.theta}) at target {self.target.value}'
-
-    def item_irt_function(self, item: Item, irt_model: str, theta: float) -> float:
-        if self.target_type == 'tcc':
-            return item.irf(irt_model, theta)
-        elif self.target_type == 'tif':
-            return item.iif(irt_model, theta)
-
-    def bundle_irt_function(self, bundle: Bundle, irt_model: str, theta: float) -> float:
-        if self.target_type == 'tcc':
-            return bundle.trf(irt_model, theta)
-        elif self.target_type == 'tif':
-            return bundle.tif(irt_model, theta)

app/models/metadata_constraint.py

@@ -1,12 +1,7 @@
 from __future__ import annotations
 from typing import TYPE_CHECKING
 
-import logging
+from models.constraint import *
-
-from pulp import lpSum
-
-from models.constraint import Constraint
-from models.problem import Problem
 
 if TYPE_CHECKING:
     from models.solver_run import SolverRun

app/models/objective_function.py

@@ -1,15 +1,18 @@
+from __future__ import annotations
+
 from pydantic import BaseModel
 from typing import Dict, List, AnyStr
 
-from models.target import Target
+from models.tif_target import TifTarget
+from models.tcc_target import TccTarget
 
 
 class ObjectiveFunction(BaseModel):
     # minimizing tif/tcc target value is only option currently
     # as we add more we can build this out to be more dynamic
     # likely with models representing each objective function type
-    tif_targets: List[Target]
+    tif_targets: List[TifTarget]
-    tcc_targets: List[Target]
+    tcc_targets: List[TccTarget]
     target_variance_percentage: int = 10
     objective: AnyStr = "minimize"
     weight: Dict = {'tif': 1, 'tcc': 1}
@@ -18,7 +21,7 @@ class ObjectiveFunction(BaseModel):
                                 limit: float or bool,
                                 all: bool = False,
                                 amount: float = 0.1,
-                                targets: list[Target] = []) -> bool:
+                                targets: list[TifTarget|TccTarget] = []) -> bool:
         if all:
             for target in self.tif_targets:
                 target.drift = round(target.drift + amount, 2)
@@ -44,5 +47,5 @@ class ObjectiveFunction(BaseModel):
 
         return minimum_drift
 
-    def all_targets(self) -> list[Target]:
+    def all_targets(self) -> list[TifTarget|TccTarget]:
         return self.tif_targets + self.tcc_targets

app/models/problem.py

@@ -51,9 +51,9 @@ class Problem(BaseModel):
         self.problem.solve()
         return self.problem
 
-    def generate(self, solution: Solution, solver_run: SolverRun):    
+    def generate(self, solution: Solution, solver_run: SolverRun) -> None:
 
-        # Form Constraints
+        # Objective Function
         self.problem += lpSum(
             [
                 bundle.count * self.solver_bundles_var[bundle.id]
@@ -76,56 +76,15 @@ class Problem(BaseModel):
             ]
         ) <= solver_run.total_form_items - 1, f'Ensuring uniqueness for form'
 
-    def generate_constraints(self, solver_run: SolverRun, current_drift: int):
+    def generate_constraints(self, solver_run: SolverRun, current_drift: int) -> None:
         logging.info('Creating Constraints...')
 
         try:
             for constraint in solver_run.constraints:
                 constraint.build(self, solver_run)
-            
-            for tif_target in solver_run.objective_function.tif_targets:
-                self.problem += lpSum([
-                    bundle.tif(solver_run.irt_model, tif_target.theta)
-                    * self.solver_bundles_var[bundle.id]
-                    for bundle in self.bundles
-                ] + [
-                    item.iif(solver_run.irt_model, tif_target.theta) *
-                    self.solver_items_var[item.id]
-                    for item in self.items
-                ]) >= tif_target.minimum(
-                ), f'Min TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%'
-                self.problem += lpSum([
-                    bundle.tif(solver_run.irt_model, tif_target.theta)
-                    * self.solver_bundles_var[bundle.id]
-                    for bundle in self.bundles
-                ] + [
-                    item.iif(solver_run.irt_model, tif_target.theta) *
-                    self.solver_items_var[item.id]
-                    for item in self.items
-                ]) <= tif_target.maximum(
-                ), f'Max TIF theta({tif_target.theta}) at target {tif_target.value} drift at {current_drift}%'
 
-            for tcc_target in solver_run.objective_function.tcc_targets:
+            for target in solver_run.objective_function.all_targets():
-                self.problem += lpSum([
+                target.build_constraint(self, solver_run)
-                    bundle.trf(solver_run.irt_model, tcc_target.theta)
-                    * self.solver_bundles_var[bundle.id]
-                    for bundle in self.bundles
-                ] + [
-                    item.irf(solver_run.irt_model, tcc_target.theta) *
-                    self.solver_items_var[item.id]
-                    for item in self.items
-                ]) >= tcc_target.minimum(
-                ), f'Min TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%'
-                self.problem += lpSum([
-                    bundle.trf(solver_run.irt_model, tcc_target.theta)
-                    * self.solver_bundles_var[bundle.id]
-                    for bundle in self.bundles
-                ] + [
-                    item.irf(solver_run.irt_model, tcc_target.theta) *
-                    self.solver_items_var[item.id]
-                    for item in self.items
-                ]) <= tcc_target.maximum(
-                ), f'Max TCC theta({tcc_target.theta}) at target {tcc_target.value} drift at {current_drift}%'
 
             logging.info('Constraints Created...')
         except ValueError as error:

app/models/solver_run.py

@@ -8,7 +8,6 @@ from models.item import Item
 from models.constraint import Constraint
 from models.metadata_constraint import MetadataConstraint
 from models.bundle_constraint import BundleConstraint
-# from models.irt_target_constraint import IrtTargetConstraint
 from models.irt_model import IRTModel
 from models.bundle import Bundle
 from models.objective_function import ObjectiveFunction
@@ -41,13 +40,6 @@ class SolverRun(BaseModel):
             elif constraint.reference_attribute.type == 'bundle':
                 constraints.append(BundleConstraint(reference_attribute=constraint.reference_attribute, minimum=constraint.minimum, maximum=constraint.maximum))
 
-        # constraints for tif and tcc targets
-        # for target in self.objective_function.tif_targets:
-        #     constraints.append(IrtTargetConstraint(target=target, target_type='tif'))
-
-        # for target in self.objective_function.tcc_targets:
-        #     constraints.append(IrtTargetConstraint(target=target, target_type='tcc'))
-
         self.constraints = constraints
 
     def get_item(self, item_id: int) -> Item or None:

app/models/target.py

@@ -1,6 +1,15 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
 from pydantic import BaseModel
 from typing import Optional
 
+from pulp import lpSum
+
+if TYPE_CHECKING:
+    from models.solver_run import SolverRun
+    from models.problem import Problem
+
 class Target(BaseModel):
     theta: float
     value: float

app/models/tcc_target.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from models.target import *
+
+if TYPE_CHECKING:
+    from models.problem import Problem
+
+class TccTarget(Target):
+    def build_constraint(self, problem_handler: Problem, solver_run: SolverRun):
+        problem_handler.problem += lpSum([
+            bundle.trf(solver_run.irt_model, self.theta)
+            * problem_handler.solver_bundles_var[bundle.id]
+            for bundle in problem_handler.bundles
+        ] + [
+            item.irf(solver_run.irt_model, self.theta) *
+            problem_handler.solver_items_var[item.id]
+            for item in problem_handler.items
+        ]) >= self.minimum(
+        ), f'Min TCC theta({self.theta}) at target {self.value} with a drift % of {self.drift}'
+
+        problem_handler.problem += lpSum([
+            bundle.trf(solver_run.irt_model, self.theta)
+            * problem_handler.solver_bundles_var[bundle.id]
+            for bundle in problem_handler.bundles
+        ] + [
+            item.irf(solver_run.irt_model, self.theta) *
+            problem_handler.solver_items_var[item.id]
+            for item in problem_handler.items
+        ]) <= self.maximum(
+        ), f'Max TCC theta({self.theta}) at target {self.value} with a drift % of {self.drift}'

app/models/tif_target.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from models.target import *
+
+if TYPE_CHECKING:
+    from models.problem import Problem
+
+class TifTarget(Target):
+    def build_constraint(self, problem_handler: Problem, solver_run: SolverRun):
+        problem_handler.problem += lpSum([
+            bundle.tif(solver_run.irt_model, self.theta)
+            * problem_handler.solver_bundles_var[bundle.id]
+            for bundle in problem_handler.bundles
+        ] + [
+            item.iif(solver_run.irt_model, self.theta) *
+            problem_handler.solver_items_var[item.id]
+            for item in problem_handler.items
+        ]) >= self.minimum(
+        ), f'Min TIF theta({self.theta}) at target {self.value} with a drift % of {self.drift}'
+
+        problem_handler.problem += lpSum([
+            bundle.tif(solver_run.irt_model, self.theta)
+            * problem_handler.solver_bundles_var[bundle.id]
+            for bundle in problem_handler.bundles
+        ] + [
+            item.iif(solver_run.irt_model, self.theta) *
+            problem_handler.solver_items_var[item.id]
+            for item in problem_handler.items
+        ]) <= self.maximum(
+        ), f'Max TIF theta({self.theta}) at target {self.value} with a drift % of {self.drift}'

app/services/form_generation_service.py

@@ -3,7 +3,7 @@ import json, random, io, logging
 from pulp import LpProblem, LpVariable, LpMinimize, LpStatus, lpSum
 
 from lib.application_configs import ApplicationConfigs
-from helpers import aws_helper, tar_helper, csv_helper, service_helper, solver_helper
+from helpers import aws_helper, tar_helper, csv_helper, service_helper
 from lib.errors.item_generation_error import ItemGenerationError
 
 from models.solver_run import SolverRun