Top

vania module

from .fair_distributor import FairDistributor

__all__ = ['FairDistributor']

Classes

class FairDistributor

This class represents a general fair distributor. Objects of this class allow solving fair distribution problems.

class FairDistributor:

    """
    This class represents a general fair distributor.
    Objects of this class allow solving fair distribution problems.
    """

    def __init__(self, targets=[], objects=[], weights=[]):
        """
        This constructor is simply a way to simplify the flow of solving a problem.
        """
        self.set_data(targets, objects, weights)

    def set_data(self, targets, objects, weights):
        """
        This method is a general setter for the data associated with the problem.
        Data is being validated by the validator method.
        """
        self._targets = targets
        self._objects = objects
        self._weights = weights

    def validate(self):
        """
        This method validates the current data.
        """
        try:
            self._validate()
            return True
        except ValueError:
            return False

    def _validate(self):
        if len(self._weights) != len(self._targets):
            raise ValueError(
                "The amount of weight lines should match the amount of targets")

        # All values should be positive
        s = list(
            dropwhile(lambda x: len(x) == len(self._objects), self._weights))
        if len(s) != 0:
            raise ValueError(
                "The amount of weight columns should match the amount of objects")
        for lines in self._weights:
            for i in lines:
                if i < 0:
                    raise ValueError(
                        "All weights must be positive")

    def distribute(self, fairness=True, output=None):
        """
        This method is responsible for actually solving the linear programming problem.
        It uses the data in the instance variables.
        The optional parameter **fairness** indicates if the solution should minimize individual effort. By default the solution will enforce this condition.
        An error will be raised in case the data has inconsistencies.
        """
        # Validate the problem variables
        self._validate()

        # State problem
        problem = pulp.LpProblem("Fair Distribution Problem", pulp.LpMinimize)

        # Prepare variables
        targets_objects = {}
        for (t, target) in enumerate(self._targets):
            for (o, object) in enumerate(self._objects):
                variable = pulp.LpVariable(
                    'x' + str(t) + str(o), lowBound=0, cat='Binary')
                position = {'target': t, 'object': o}
                targets_objects['x' + str(t) + str(o)] = (variable, position)

        # Generate linear expression for self._weights (Summation #1)
        weights = [(variable, self._weights[weight_position['target']][weight_position['object']])
                   for (variable, weight_position) in targets_objects.values()]
        weight_expression = pulp.LpAffineExpression(weights)

        # Generate linear expression for effort distribution (Summation #2)
        weight_diff_vars = []
        if fairness:
            total_targets = len(self._targets)
            for (t, target) in enumerate(self._targets):
                weight_diff_aux_variable = pulp.LpVariable(
                    'weight_diff_'+str(t), lowBound=0)
                weight_diff_vars.append(weight_diff_aux_variable)

                negative_effort_diff_weights = []
                positive_effort_diff_weights = []
                negative_factor = -1 * total_targets
                positive_factor = 1 * total_targets
                for (o, object) in enumerate(self._objects):
                    id = 'x' + str(t) + str(o)
                    negative_effort_diff_weights.append(
                        (targets_objects[id][0], negative_factor * self._weights[t][o]))
                    positive_effort_diff_weights.append(
                        (targets_objects[id][0], positive_factor * self._weights[t][o]))

                negative_effort_diff = pulp.LpAffineExpression(
                    negative_effort_diff_weights) + weight_expression
                positive_effort_diff = pulp.LpAffineExpression(
                    positive_effort_diff_weights) - weight_expression

                problem += negative_effort_diff <= weight_diff_aux_variable, 'abs negative effort diff ' + \
                    str(t)
                problem += positive_effort_diff <= weight_diff_aux_variable, 'abs positive effort diff ' + \
                    str(t)

        # Constraints - Each task must be done
        for (o, object) in enumerate(self._objects):
            constraints = []
            for (t, target) in enumerate(self._targets):
                constraints.append(targets_objects['x' + str(t) + str(o)][0])
            problem += pulp.lpSum(constraints) == 1, 'Task ' + \
                str(o) + ' must be done'

        # Set objective function
        problem += weight_expression + \
            pulp.lpSum(weight_diff_vars), "obj"

        if output:
            problem.writeLP(output)

        problem.solve()

        # Build output
        data = {}
        for v in filter(lambda x: x.varValue > 0, problem.variables()):
            if v.name not in targets_objects:
                continue
            position = targets_objects[v.name][1]
            target = self._targets[position['target']]
            object = self._objects[position['object']]

            if target not in data:
                data[target] = []

            data[target].append(object)

        return data

Ancestors (in MRO)

Static methods

def __init__(

self, targets=[], objects=[], weights=[])

This constructor is simply a way to simplify the flow of solving a problem.

def __init__(self, targets=[], objects=[], weights=[]):
    """
    This constructor is simply a way to simplify the flow of solving a problem.
    """
    self.set_data(targets, objects, weights)

def distribute(

self, fairness=True, output=None)

This method is responsible for actually solving the linear programming problem. It uses the data in the instance variables. The optional parameter fairness indicates if the solution should minimize individual effort. By default the solution will enforce this condition. An error will be raised in case the data has inconsistencies.

def distribute(self, fairness=True, output=None):
    """
    This method is responsible for actually solving the linear programming problem.
    It uses the data in the instance variables.
    The optional parameter **fairness** indicates if the solution should minimize individual effort. By default the solution will enforce this condition.
    An error will be raised in case the data has inconsistencies.
    """
    # Validate the problem variables
    self._validate()
    # State problem
    problem = pulp.LpProblem("Fair Distribution Problem", pulp.LpMinimize)
    # Prepare variables
    targets_objects = {}
    for (t, target) in enumerate(self._targets):
        for (o, object) in enumerate(self._objects):
            variable = pulp.LpVariable(
                'x' + str(t) + str(o), lowBound=0, cat='Binary')
            position = {'target': t, 'object': o}
            targets_objects['x' + str(t) + str(o)] = (variable, position)
    # Generate linear expression for self._weights (Summation #1)
    weights = [(variable, self._weights[weight_position['target']][weight_position['object']])
               for (variable, weight_position) in targets_objects.values()]
    weight_expression = pulp.LpAffineExpression(weights)
    # Generate linear expression for effort distribution (Summation #2)
    weight_diff_vars = []
    if fairness:
        total_targets = len(self._targets)
        for (t, target) in enumerate(self._targets):
            weight_diff_aux_variable = pulp.LpVariable(
                'weight_diff_'+str(t), lowBound=0)
            weight_diff_vars.append(weight_diff_aux_variable)
            negative_effort_diff_weights = []
            positive_effort_diff_weights = []
            negative_factor = -1 * total_targets
            positive_factor = 1 * total_targets
            for (o, object) in enumerate(self._objects):
                id = 'x' + str(t) + str(o)
                negative_effort_diff_weights.append(
                    (targets_objects[id][0], negative_factor * self._weights[t][o]))
                positive_effort_diff_weights.append(
                    (targets_objects[id][0], positive_factor * self._weights[t][o]))
            negative_effort_diff = pulp.LpAffineExpression(
                negative_effort_diff_weights) + weight_expression
            positive_effort_diff = pulp.LpAffineExpression(
                positive_effort_diff_weights) - weight_expression
            problem += negative_effort_diff <= weight_diff_aux_variable, 'abs negative effort diff ' + \
                str(t)
            problem += positive_effort_diff <= weight_diff_aux_variable, 'abs positive effort diff ' + \
                str(t)
    # Constraints - Each task must be done
    for (o, object) in enumerate(self._objects):
        constraints = []
        for (t, target) in enumerate(self._targets):
            constraints.append(targets_objects['x' + str(t) + str(o)][0])
        problem += pulp.lpSum(constraints) == 1, 'Task ' + \
            str(o) + ' must be done'
    # Set objective function
    problem += weight_expression + \
        pulp.lpSum(weight_diff_vars), "obj"
    if output:
        problem.writeLP(output)
    problem.solve()
    # Build output
    data = {}
    for v in filter(lambda x: x.varValue > 0, problem.variables()):
        if v.name not in targets_objects:
            continue
        position = targets_objects[v.name][1]
        target = self._targets[position['target']]
        object = self._objects[position['object']]
        if target not in data:
            data[target] = []
        data[target].append(object)
    return data

def set_data(

self, targets, objects, weights)

This method is a general setter for the data associated with the problem. Data is being validated by the validator method.

def set_data(self, targets, objects, weights):
    """
    This method is a general setter for the data associated with the problem.
    Data is being validated by the validator method.
    """
    self._targets = targets
    self._objects = objects
    self._weights = weights

def validate(

self)

This method validates the current data.

def validate(self):
    """
    This method validates the current data.
    """
    try:
        self._validate()
        return True
    except ValueError:
        return False