<commit_before># -*- coding: utf-8 -*-

"""Thread of dimensional synthesis."""

__author__ = "<NAME>"
__copyright__ = "Copyright (C) 2016-2021"
__license__ = "AGPL"
__email__ = "<EMAIL>"

from typing import cast, Dict, Any
from time import process_time
from platform import system, release, machine
from numpy.distutils.cpuinfo import cpu
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QWidget
from pyslvs.optimization import FMatch, FConfig
from pyslvs.metaheuristics import algorithm, AlgorithmType, AlgorithmConfig
from pyslvs_ui.info import logger
from pyslvs_ui.synthesis.thread import BaseThread


class DimensionalThread(BaseThread):
    """The QThread class to handle algorithm."""
    progress_update = Signal(int, str)
    result = Signal(dict)

    def __init__(
        self,
        algorithm: AlgorithmType,
        mech: Dict[str, Any],
        settings: Dict[str, Any],
        parent: QWidget
    ):
        super(DimensionalThread, self).__init__(parent)
        self.algorithm = algorithm
        self.mech = mech
        self.planar = FMatch(cast(FConfig, self.mech))
        self.settings = settings
        self.loop = 1

    def set_loop(self, loop: int) -> None:
        """Set the loop times."""
        self.loop = loop

    def run(self) -> None:
        """Start the algorithm loop."""
        for name, path in self.mech['target'].items():
            logger.debug(f"- [P{name}] ({len(path)})")
        t0 = process_time()
        for self.loop in range(self.loop):
            logger.info(f"Algorithm [{self.loop + 1}]: {self.algorithm}")
            if self.is_stop:
                # Cancel the remaining tasks
                logger.info("Canceled.")
                continue
            self.result.emit(self.__algorithm())
        logger.info(f"total cost time: {process_time() - t0:.02f} [s]")
        self.finished.emit()

    def __algorithm(self) -> Dict[str, Any]:
        """Get the algorithm result."""
        t0 = process_time()
        a = algorithm(self.algorithm)(
            self.planar,
            cast(AlgorithmConfig, self.settings),
            progress_fun=self.progress_update.emit,
            interrupt_fun=lambda: self.is_stop,
        )
        expression = a.run()
        tf = a.history()
        time_spend = process_time() - t0
        info = cpu.info[0]
        my_cpu = info.get("model name", info.get('ProcessorNameString', ''))
        last_gen = tf[-1][0]
        mechanism = {
            'algorithm': self.algorithm.value,
            'time': time_spend,
            'last_gen': last_gen,
            'last_fitness': tf[-1][1],
            'interrupted': str(last_gen) if self.is_stop else 'False',
            'settings': self.settings,
            'hardware_info': {
                'os': f"{system()} {release()} {machine()}",
                'cpu': my_cpu,
            },
            'time_fitness': tf,
            'callback': self.planar.callback,
        }
        mechanism.update(self.mech)
        mechanism['expression'] = expression
        logger.info(f"cost time: {time_spend:.02f} [s]")
        return mechanism<commit_msg>Fix MyPy error.<commit_after># -*- coding: utf-8 -*-

"""Thread of dimensional synthesis."""

__author__ = "<NAME>"
__copyright__ = "Copyright (C) 2016-2021"
__license__ = "AGPL"
__email__ = "<EMAIL>"

from typing import cast, Dict, List, Any
from time import process_time
from platform import system, release, machine
from numpy.distutils.cpuinfo import cpu
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QWidget
from pyslvs.optimization import FMatch, FConfig
from pyslvs.metaheuristics import algorithm, AlgorithmType, AlgorithmConfig
from pyslvs_ui.info import logger
from pyslvs_ui.synthesis.thread import BaseThread


class DimensionalThread(BaseThread):
    """The QThread class to handle algorithm."""
    progress_update = Signal(int, str)
    result = Signal(dict)

    def __init__(
        self,
        algorithm: AlgorithmType,
        mech: Dict[str, Any],
        settings: Dict[str, Any],
        parent: QWidget
    ):
        super(DimensionalThread, self).__init__(parent)
        self.algorithm = algorithm
        self.mech = mech
        self.planar = FMatch(cast(FConfig, self.mech))
        self.settings = settings
        self.loop = 1

    def set_loop(self, loop: int) -> None:
        """Set the loop times."""
        self.loop = loop

    def run(self) -> None:
        """Start the algorithm loop."""
        for name, path in self.mech['target'].items():
            logger.debug(f"- [P{name}] ({len(path)})")
        t0 = process_time()
        for self.loop in range(self.loop):
            logger.info(f"Algorithm [{self.loop + 1}]: {self.algorithm}")
            if self.is_stop:
                # Cancel the remaining tasks
                logger.info("Canceled.")
                continue
            self.result.emit(self.__algorithm())
        logger.info(f"total cost time: {process_time() - t0:.02f} [s]")
        self.finished.emit()

    def __algorithm(self) -> Dict[str, Any]:
        """Get the algorithm result."""
        t0 = process_time()
        a = algorithm(self.algorithm)(
            self.planar,
            cast(AlgorithmConfig, self.settings),
            progress_fun=self.progress_update.emit,
            interrupt_fun=lambda: self.is_stop,
        )
        expression = a.run()
        tf = a.history()
        time_spend = process_time() - t0
        info = cast(List[Dict[str, Any]], cpu.info)[0]
        my_cpu = info.get("model name", info.get('ProcessorNameString', ''))
        last_gen = tf[-1][0]
        mechanism = {
            'algorithm': self.algorithm.value,
            'time': time_spend,
            'last_gen': last_gen,
            'last_fitness': tf[-1][1],
            'interrupted': str(last_gen) if self.is_stop else 'False',
            'settings': self.settings,
            'hardware_info': {
                'os': f"{system()} {release()} {machine()}",
                'cpu': my_cpu,
            },
            'time_fitness': tf,
            'callback': self.planar.callback,
        }
        mechanism.update(self.mech)
        mechanism['expression'] = expression
        logger.info(f"cost time: {time_spend:.02f} [s]")
        return mechanism
{'fraction_non_alphanumeric': 0.12810707456978968, 'fraction_numerical': 0.0076481835564053535, 'mean_word_length': 2.2743870631194576, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1691126', 'n_tokens_mistral': 1894, 'n_tokens_neox': 1849, 'n_words': 454}