Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use _multiprocess method in Molotov

Best Python code snippet using molotov_python

mfx_publisher.py

Source:mfx_publisher.py

1import sys2import zmq3import json4from time import sleep5class MFXPublisher(object):6    """7    Publishes real-time market data to client strategies  8    """9    def __init__(self,10                _host='127.0.0.1',       11                _protocol='tcp', 12                _pub_port=42069,13                _sleep_delay=1,14                _multiprocess=True):   15        # ZeroMQ Host16        self._host = _host17        18        # Connection Protocol19        self._protocol = _protocol20        21        # TCP Connection URL Template22        self._url = self._protocol + "://" + self._host + ":"23        # SUB Socket Port24        self._pub_port = _pub_port25        self._multiprocess = _multiprocess26        27        if self._multiprocess == False:28            self.create_context()29            print("[SUCCESS] MetaFarmerX Market Feeds Live")30        # set the frequency of data updates (s)31        self._sleep_delay = _sleep_delay 32        # Market Data Client Configuration33        self._PUBLISH_MARKET_DATA = True34        # Run MetaFarmerX35        self._RUN = True36    ##########################################################################37    def create_context(self):38        """39        Creates ZeroMQ Context40        """41        # ZeroMQ Context42        self._zmq_context = zmq.Context()43        # Create Sockets44        self._pub_socket = self._zmq_context.socket(zmq.PUB)45        self._pub_socket.setsockopt(zmq.SNDHWM, 1)46        self._pub_socket.setsockopt(zmq.LINGER, 0)47        # Connect PUB Socket to send market data48        self._pub_socket.bind(self._url + str(self._pub_port)) 49    ##########################################################################    50    def set_delay(self, _frequency):51        """52        Set new frequency of market updates53        """54        self._sleep_delay = _frequency55      56    ##########################################################################    57    def remote_send(self, _socket, _data):58        """59        Send string on socket60        """61        try:62            _socket.send_json(_data)63        except zmq.error.Again:64            print("\nResource timeout... trying again")65            sleep(self._sleep_delay)66      67    ##########################################################################    68    def update_data(self):69        """70        Retreive market data71        """72        self.fetch_data()73        self.transform_data()74      75    ##########################################################################76    def on_tick(self):77        """78        Publish market data on PUB socket79        """80        if self._PUBLISH_MARKET_DATA:81            # update market data82            self.update_data()83            # send data to client through PUB socket84            self.remote_send(self._pub_socket, self.tranformed_data)85    ##########################################################################86    def run(self):87        """88        Run MetaFarmerX89        """90        if self._multiprocess:91            self.create_context()92            print("[SUCCESS] MetaFarmerX Market Feeds Live")93        while self._RUN:94            # blocking delay95            sleep(self._sleep_delay)...

evaluation.py

Source:evaluation.py

1"""The genetic operation of evaluation.2This module defines the a basic form of the evaluation phase of bingo3evolutionary algorithms.4"""5from multiprocessing import Pool6class Evaluation:7    """Base phase for calculating fitness of a population.8    A base class for the fitness evaluation of populations of genetic9    individuals (list of chromosomes) in bingo.  All individuals in the10    population are evaluated with a fitness function unless their fitness has11    already been set.12    Parameters13    ----------14    fitness_function : FitnessFunction15        The function class that is used to calculate fitnesses of individuals16        in the population.17    redundant : bool18        Whether to re-evaluate individuals that have been evaluated previously.19        Default False.20    multiprocess : int or bool21        Number of processes to use in parallel evaluation22        or False for serial evaluation.23        Default False.24    Attributes25    ----------26    fitness_function : FitnessFunction27        The function class that is used to calculate fitnesses of individuals28        in the population.29    eval_count : int30        the number of fitness function evaluations that have occurred31    """32    def __init__(self, fitness_function, redundant=False, multiprocess=False):33        self.fitness_function = fitness_function34        self._redundant = redundant35        self._multiprocess = multiprocess36    @property37    def eval_count(self):38        """int : the number of evaluations that have been performed"""39        return self.fitness_function.eval_count40    @eval_count.setter41    def eval_count(self, value):42        self.fitness_function.eval_count = value43    def __call__(self, population):44        """Evaluates the fitness of an individual45        Parameters46        ----------47        population : list of chromosomes48                     population for which fitness should be calculated49        """50        if self._multiprocess:51            self._multiprocess_eval(population)52        else:53            self._serial_eval(population)54    def _serial_eval(self, population):55        for indv in population:56            if self._redundant or not indv.fit_set:57                indv.fitness = self.fitness_function(indv)58    def _multiprocess_eval(self, population):59        num_procs = self._multiprocess if isinstance(self._multiprocess, int) \60            else None61        with Pool(processes=num_procs) as pool:62            results = []63            for i, indv in enumerate(population):64                if self._redundant or not indv.fit_set:65                    results.append(66                            pool.apply_async(_fitness_job,67                                             (indv, self.fitness_function, i)))68            for res in results:69                indv, i = res.get()70                population[i] = indv71def _fitness_job(individual, fitness_function, population_index):72    individual.fitness = fitness_function(individual)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.