# SPDX-FileCopyrightText: All Contributors to the PyTango project
# SPDX-License-Identifier: LGPL-3.0-or-later

# Imports
import functools
import asyncio
import inspect
import collections
import os
import types
import concurrent.futures
import warnings
import threading

from asyncio import futures, coroutines
from asyncio.tasks import ensure_future
from typing import Callable

# Tango imports
from tango.green import AbstractExecutor
from tango.utils import _is_coroutine_function, PyTangoThreadPoolExecutor

__all__ = (
    "AsyncioExecutor",
    "get_global_executor",
    "set_global_executor",
    "_switch_global_executor_to_thread",
)

_ALREADY_WARNED_FUNCTIONS = []


# Function removed from Python 3.11
# Taken from https://github.com/python/cpython/blob/3.10/Lib/asyncio/coroutines.py
# (without the _DEBUG part)
# Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
# 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020, 2021, 2022 Python Software Foundation;
# All Rights Reserved
def _coroutine(func):
    """Decorator to mark coroutines.
    If the coroutine is not yielded from before it is destroyed,
    an error message is logged.
    """
    if inspect.iscoroutinefunction(func):
        return func

    if inspect.isgeneratorfunction(func):
        coro = func
    else:

        @functools.wraps(func)
        def coro(*args, **kw):
            res = func(*args, **kw)
            if asyncio.isfuture(res) or inspect.isgenerator(res):
                res = yield from res
            else:
                # If 'res' is an awaitable, run it.
                try:
                    await_meth = res.__await__
                except AttributeError:
                    pass
                else:
                    if isinstance(res, collections.abc.Awaitable):
                        res = yield from await_meth()
            return res

    coro = types.coroutine(coro)
    wrapper = coro
    wrapper._is_coroutine = (
        asyncio.coroutines._is_coroutine
    )  # For iscoroutinefunction().
    return wrapper


# In Python 3.12 the legacy generator-based coroutines are not allowed for execution anymore
# Here we use modified run_coroutine_threadsafe method, which still can execute them
# Taken from https://github.com/python/cpython/blob/3.12/Lib/asyncio/tasks.py
# Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
# 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020, 2021, 2022, 2023 Python Software Foundation;
# All Rights Reserved
def run_coroutine_threadsafe(coro, loop):
    """Submit a coroutine object to a given event loop.

    Return a concurrent.futures.Future to access the result.
    """
    if not coroutines.iscoroutine(coro) and not types.GeneratorType:
        raise TypeError("A coroutine object is required")

    future = concurrent.futures.Future()

    def callback():
        try:
            futures._chain_future(ensure_future(coro, loop=loop), future)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            if future.set_running_or_notify_cancel():
                future.set_exception(exc)
            raise

    loop.call_soon_threadsafe(callback)
    return future


_PYTANGOTHREADPOOLEXECUTOR = None


def get_thread_pool_executor():
    global _PYTANGOTHREADPOOLEXECUTOR
    if _PYTANGOTHREADPOOLEXECUTOR is None:
        _PYTANGOTHREADPOOLEXECUTOR = PyTangoThreadPoolExecutor(
            thread_name_prefix="_PyTangoThreadPoolExecutor"
        )
    return _PYTANGOTHREADPOOLEXECUTOR


# Global executor
_MAIN_EXECUTOR = None
_THREAD_EXECUTORS = {}


def _switch_global_executor_to_thread():
    """
    internal PyTango function, use only if you sure, what you are doing!
    Used for correct behavior of TestDeviceContext
    Checks, that global executor belongs to the caller thread, and,
    if not - creates a new one and saves it as a new global
    """
    global _MAIN_EXECUTOR
    if _MAIN_EXECUTOR is not None and not _MAIN_EXECUTOR.in_executor_context():
        # we save current executor in the known subthread executors to be used later
        _THREAD_EXECUTORS[_MAIN_EXECUTOR.get_ident()] = _MAIN_EXECUTOR
        _MAIN_EXECUTOR = AsyncioExecutor()


def get_global_executor():
    global _MAIN_EXECUTOR
    if _MAIN_EXECUTOR is None:
        _MAIN_EXECUTOR = AsyncioExecutor()
    # the following patch is used for correct behavior of TestDeviceContext,
    # which has two different executors for main and device threads
    if not _MAIN_EXECUTOR.in_executor_context():
        ident = threading.get_ident(), os.getpid()
        if ident in _THREAD_EXECUTORS:
            return _THREAD_EXECUTORS[ident]

    return _MAIN_EXECUTOR


def set_global_executor(executor):
    global _MAIN_EXECUTOR
    _MAIN_EXECUTOR = executor


def _get_function_name(fn: Callable) -> str:
    if hasattr(fn, "__qualname__"):
        return fn.__qualname__
    elif hasattr(fn, "__name__"):
        return fn.__name__

    return f"{fn}"


# Asyncio executor
class AsyncioExecutor(AbstractExecutor):
    """Asyncio tango executor"""

    asynchronous = True
    default_wait = False

    def __init__(self, loop=None, subexecutor=None):
        super().__init__()
        if loop is None:
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
        self.loop = loop
        self.subexecutor = (
            subexecutor if subexecutor is not None else get_thread_pool_executor()
        )

    def delegate(self, fn, *args, **kwargs):
        """Return the given operation as an asyncio future."""
        callback = functools.partial(fn, *args, **kwargs)
        coro = self.loop.run_in_executor(self.subexecutor, callback)
        return asyncio.ensure_future(coro)

    def access(self, accessor, timeout=None):
        """Return a result from an asyncio future."""
        if self.loop.is_running():
            raise RuntimeError("Loop is already running")
        coro = asyncio.wait_for(accessor, timeout)
        return self.loop.run_until_complete(coro)

    def submit(self, fn, *args, **kwargs):
        """Submit an operation"""
        if _is_coroutine_function(fn):
            return run_coroutine_threadsafe(fn(*args, **kwargs), self.loop)
        else:
            # we leave this part of the code to support legacy servers
            name = _get_function_name(fn)
            if name not in _ALREADY_WARNED_FUNCTIONS:
                _ALREADY_WARNED_FUNCTIONS.append(name)
                warnings.warn(
                    f"Sync {name} function called: support of "
                    f"sync functions in PyTango's Asyncio mode is "
                    f"deprecated. Use 'async def' instead of 'def'.",
                    DeprecationWarning,
                )
            corofn = _coroutine(lambda: fn(*args, **kwargs))
            return run_coroutine_threadsafe(corofn(), self.loop)

    def execute(self, fn, *args, **kwargs):
        """Execute an operation and return the result."""
        if self.in_executor_context():
            if _is_coroutine_function(fn):
                return fn(*args, **kwargs)
            else:
                # we leave this part of the code to support legacy servers
                name = _get_function_name(fn)
                if name not in _ALREADY_WARNED_FUNCTIONS:
                    _ALREADY_WARNED_FUNCTIONS.append(name)
                    warnings.warn(
                        f"Sync {name} function called: support of "
                        f"sync functions in PyTango's Asyncio mode is "
                        f"deprecated. Use 'async def' instead of 'def'.",
                        DeprecationWarning,
                    )
                corofn = _coroutine(lambda: fn(*args, **kwargs))
                return corofn()
        future = self.submit(fn, *args, **kwargs)
        return future.result()