API Reference¶

def configure_search(self) -> Search:
    """Configures the search object.

    Returns:
        Search: A configured search object.
    """
    search = self.search()
    search = search.source(self.search_source)
    search = search.params(scroll=self.scan_scroll, size=self.pool_size)
    return search

def create_translation_queue(self) -> JoinableQueue:
    """Creates a queue that can translate documents in parallel.

    Returns:
        JoinableQueue: A queue for parallel document translation.
    """
    return JoinableQueue(self.pool_size)

def init_interpreter(self) -> Any:
    """Initializes the interpreter.

    Returns:
        Any: The initialized interpreter.
    """
    pack_dir = path.join(self.data_dir, 'packs', self.interpreter_name)
    interpreters = (Apertium, Argos,)
    Interpreter = next(
        i for i in interpreters if i.name.lower() == self.interpreter_name.lower())
    return Interpreter(
        self.source_language,
        self.target_language,
        self.intermediary_language,
        pack_dir)

def instantiate_interpreter(self) -> Any:
    """Instantiates the interpreter.

    Returns:
        Any: An instance of the interpreter.
    """
    if not hasattr(self, 'interpreter'):
        with self.print_done(f'Instantiating {self.interpreter_name} interpreter'):
            self.interpreter = self.init_interpreter()
    return self.interpreter

@contextmanager
def print_done(self, string: str) -> ContextManager:
    """Prints a string and yields.

    Args:
        string (str): The string to be printed.

    Returns:
        ContextManager: A context manager.
    """
    logger.info(string)
    if self.stdout_loglevel > 20:
        string = '\r%s...' % string
        self.print_flush(string)
        try:
            yield
            print('{0} \033[92mdone\033[0m'.format(string))
        except (FatalTranslationException, ElasticsearchException, Full) as error:
            logger.error(error, exc_info=True)
            print('{0} \033[91merror\033[0m'.format(string))
            sys.exit(1)
    else:
        yield

def print_flush(self, string: str) -> None:
    """Prints and flushes a string to stdout.

    Args:
        string (str): The string to be printed.
    """
    sys.stdout.write('\r{0}'.format(string))
    sys.stdout.flush()

def process_document(
        self,
        translation_queue: JoinableQueue,
        hit: Any,
        progress: Progress,
        task: Any,
        shared_fatal_error: Manager) -> None:
    """Processes a document.

    Args:
        translation_queue (JoinableQueue): A queue for parallel document translation.
        hit (Any): The document to be translated.
        index (int): The index of the document.
        progress (Progress): A progress object.
        task (Any): The current task.
        shared_fatal_error (Manager): A shared manager for fatal errors.
    """
    translation_queue.put((self, hit), True, self.pool_timeout)
    progress.advance(task)
    if shared_fatal_error.value:
        raise FatalTranslationException(shared_fatal_error.value)

def search(self) -> Search:
    """Executes a search query.

    Returns:
        Search: The search result.
    """
    using = self.create_client()
    search = Search(index=self.index, using=using)
    if self.query_string:
        search = search.query("query_string", query=self.query_string)
    return search

def start(self) -> None:
    """Starts or plans the translation process."""
    if self.plan:
        self.start_later()
    else:
        self.start_now()

def start_later(self):
    """Plan the translation process."""
    self.instantiate_interpreter()
    total = self.search().count()
    desc = f"Planning translation for {total} document{'s'[:total^1]}"
    with self.print_done(desc):
        search = self.configure_search()
        for hit in search.scan():
            logger.info(f'Planned translation for doc {hit.meta.id}')
            translate_document_task.delay(self.options, hit.meta.to_dict())

def start_now(self) -> None:
    """Starts the translation process."""
    self.instantiate_interpreter()
    total = self.search().count()
    desc = 'Translating %s document(s)' % total
    with self.print_done(desc):
        search = self.configure_search()
        translation_queue = self.create_translation_queue()
        with self.with_shared_fatal_error() as shared_fatal_error:
            self.translate_documents_in_pool(
                search, translation_queue, shared_fatal_error, total)

def translate_documents_in_pool(
        self,
        search: Search,
        translation_queue: JoinableQueue,
        shared_fatal_error: Manager,
        total: int) -> None:
    """Translates documents.

    Args:
        search (Search): A search object.
        translation_queue (JoinableQueue): A queue for parallel document translation.
        shared_fatal_error (Manager): A shared manager for fatal errors.
        total (int): The total number of documents.
    """
    with Pool(self.pool_size, translation_worker, (translation_queue, shared_fatal_error)):
        with Progress(disable=self.no_progressbar, transient=True) as progress:
            task = progress.add_task(
                f"Translating {total} document{'s'[:total^1]}", total=total)
            for hit in search.scan():
                self.process_document(
                    translation_queue, hit, progress, task, shared_fatal_error)
            translation_queue.join()

@contextmanager
def with_shared_fatal_error(self) -> ContextManager:
    """Creates a context manager for managing shared fatal errors.

    Returns:
        ContextManager: A context manager.
    """
    with Manager() as manager:
        yield manager.Value('b', None)

def download_and_install_package(self, package):
    """
    Downloads and installs a language package.

    This method locks the download using a file lock based on the package's source and target language codes.
    If the package is not installed, it is downloaded and installed.

    Args:
        package (Package): The package to download and install.

    Raises:
        Timeout: If a lock on the download path cannot be acquired within the timeout duration.
    """
    try:
        temp_dir = Path(tempfile.gettempdir())
        lock_path = temp_dir / f'{package.from_code}_{package.to_code}.lock'

        with FileLock(lock_path, timeout=600).acquire(timeout=600):
            if self.is_package_installed(package):
                return
            download_path = package.download()
            logger.info('Installing Argos package %s', package)
            return argospackage.install_from_path(download_path)
    except Timeout as exc:
        raise ArgosPackageDownloadLockTimeout(
            f'Another instance of the program is downloading the package {package}. Please try again later.') from exc

def download_necessary_languages(self):
    """
    Downloads necessary language packages if they are not installed already.

    This method performs the following steps:
    1. Updates the package index.
    2. Finds the necessary package.
    3. Downloads and installs the package.

    Raises:
        ArgosPairNotAvailable: If the necessary language package could not be found.
        Timeout: If a lock on the download path cannot be acquired within the timeout duration.
    """
    self.update_package_index()
    necessary_package = self.find_necessary_package()
    self.download_and_install_package(necessary_package)

def find_necessary_package(self):
    """
    Finds the necessary language package.

    This method loops over the available packages to find the necessary package based on the source and target language codes.
    If the package cannot be found, it raises an ArgosPairNotAvailable exception.

    Returns:
        Package: The necessary language package.

    Raises:
        ArgosPairNotAvailable: If the necessary language package could not be found.
    """
    for package in argospackage.get_available_packages():
        if package.from_code == self.source_alpha_2 and package.to_code == self.target_alpha_2:
            return package
    raise ArgosPairNotAvailable

def is_package_installed(self, package):
    """
    Checks if a package is installed.

    Args:
        package (Package): The package to check.

    Returns:
        bool: True if the package is installed, False otherwise.
    """
    return package in argospackage.get_installed_packages()

def translate(self, text_input):
    """
    Translates the input text from the source language to the target language.

    Args:
        text_input (str): The input text in the source language.

    Returns:
        str: The translated text in the target language.
    """
    return self.translation.translate(text_input)

def update_package_index(self):
    """
    Updates the package index.
    """
    argospackage.update_package_index()