def _execute_fixed()

in src/spg/PatternGenerators/basePatternGenerator.py [0:0]

• 27 lines of code
• 12 McCabe index (conditional complexity)

    def _execute_fixed(self):
        """ Triggers the control loop for the pattern generators which need patterns to have the same contents per frame,
            and are independent of all the other led walls on the stage.

            The user needs to implement generator method which is executed in a separate thread for the first frame
            whilst subsequent frames are duplicated with the replicator method. This can also be overridden however the
            base class implementation is often enough

        :return: A dict of led wall name to frame and filepath values
        """
        generator_threads = []
        replicator_threads = []
        results = {}

        # We loop over all the walls in the spd config
        for self.led_wall in self.spg.walls.values():

            results[self.led_wall.name] = {}
            self.get_and_create_pattern_output_folder(
                sub_folder="_".join([self.led_wall.name, self.name]))

            # We add a hook to perform custom logic before we start a new sequence
            self.sequence_start()
            first_frame = True

            for self.frame in range(self.number_of_frames()):
                # We add a hook to perform custom logic before we start a frame
                self.frame_start()

                # We get a dictionary of arguments to pass to our custom generator function
                kwargs = self.get_kwargs()

                if first_frame:
                    # We create a new thread which executes our generator, which we start and keep track of
                    thread = _ThreadedFunction(self.generator, self.frame, kwargs, results)
                    thread.start()
                    generator_threads.append(thread)
                else:
                    # We create a new thread which executes our replicator, which we do not start and keep track of
                    thread = _ThreadedFunction(self.replicator, self.frame, kwargs, results)
                    replicator_threads.append(thread)

                # We add a hook to perform custom logic as we end a frame
                self.frame_end()
                first_frame = False

            # We add a hook to perform custom logic as we end a sequence
            self.sequence_end()

        # We wait for the first frame to be finished
        [thread.join() for thread in generator_threads]

        # We start the replicator threads and wait for them to all finish
        [thread.start() for thread in replicator_threads]

        [thread.join() for thread in replicator_threads]

        return results