Completed
Push — master ( bebaaf...af8e87 )
by Wojtek
02:27
created

Optimizer   A

Complexity

Total Complexity 10

Size/Duplication

Total Lines 37
Duplicated Lines 0 %

Test Coverage

Coverage 100%

Importance

Changes 2
Bugs 0 Features 0
Metric Value
wmc 10
c 2
b 0
f 0
dl 0
loc 37
ccs 23
cts 23
cp 1
rs 10

4 Methods

Rating   Name   Duplication   Size   Complexity  
B optimize_process() 0 17 5
A set_custom_optimizing_order() 0 7 3
A __init__() 0 4 1
A run_pso() 0 3 1
1
"""Basic optimizer."""
2 1
import networkx as nx
0 ignored issues
show
Configuration introduced by
The import networkx could not be resolved.

This can be caused by one of the following:

1. Missing Dependencies

This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands.

# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version.

2. Missing __init__.py files

This error could also result from missing __init__.py files in your module folders. Make sure that you place one file in each sub-folder.

Loading history...
3
4 1
from grortir.externals.pyswarm.pso import pso
5 1
from grortir.main.model.core.optimization_status import OptimizationStatus
6
7
8 1
class Optimizer(object):
9
    """Optimizer is object which optimize process."""
10
11 1
    def __init__(self, process):
12 1
        self.process = process
13 1
        self.ordered_stages = nx.topological_sort(self.process)
14 1
        self.swarm_size = 40
15
16 1
    def set_custom_optimizing_order(self, ordered_stages):
17
        """Set custom order."""
18 1
        if set(self.ordered_stages) == set(ordered_stages) and len(
19
                self.ordered_stages) == len(ordered_stages):
20 1
            self.ordered_stages = ordered_stages
21
        else:
22 1
            raise ValueError("List of stages must contain all stages.")
23
24 1
    def optimize_process(self):
25
        """Optimize process.
26
        Returns:
27
            True if success, False otherwise."""
28 1
        for stage in self.ordered_stages:
29 1
            if not self.process.predecessors(stage):
30 1
                self.run_pso(stage)
31 1
            elif len(self.process.predecessors(stage)) == 1:
32 1
                predecessor = self.process.predecessors(stage)[0]
33 1
                stage.input_vector = predecessor.get_output_of_stage()
34 1
                self.run_pso(stage)
35
            else:
36 1
                raise AttributeError('Incorrect process structure.')
37
38 1
            if stage.optimization_status != OptimizationStatus.success:
39 1
                return False
40 1
        return True
41
42 1
    def run_pso(self, stage):
43
        """Run pso with predefined parameters."""
44
        pso(stage, swarmsize=self.swarm_size)
45