1948e4b1ba
Per request the full portable environment is now in git: a fresh clone's tools/ runs everything with zero installation. Only regenerable release zips and the dev venv stay ignored.
248 lines
6.9 KiB
Python
248 lines
6.9 KiB
Python
import asyncio
|
|
import os
|
|
from collections.abc import (
|
|
Awaitable,
|
|
Callable,
|
|
Mapping,
|
|
MutableMapping,
|
|
MutableSequence,
|
|
Sequence,
|
|
)
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any, TypeVar, Protocol, TYPE_CHECKING
|
|
|
|
from typing_extensions import Self
|
|
|
|
# NB: have to import Object from here instead of winrt.system to avoid circular import issues.
|
|
from winrt._winrt import add_dll_directory, remove_dll_directory, Object
|
|
|
|
# Unfortunately, we can't import at runtime because of circular imports.
|
|
if TYPE_CHECKING:
|
|
from winrt.windows.foundation import HResult, AsyncStatus
|
|
|
|
|
|
class _DllCookie:
|
|
def __init__(self, cookie: int) -> None:
|
|
self.cookie = cookie
|
|
|
|
def close(self):
|
|
if self.cookie:
|
|
remove_dll_directory(self.cookie)
|
|
self.cookie = None
|
|
|
|
def __del__(self):
|
|
self.close()
|
|
|
|
|
|
def register_dll_search_path(module_path: str) -> _DllCookie:
|
|
"""
|
|
Register a module's directory as a DLL search path.
|
|
|
|
Args:
|
|
module_path: The path to a module file (i.e. ``__file__``)
|
|
|
|
Returns:
|
|
An cookie object that will remove the search path when closed.
|
|
"""
|
|
return _DllCookie(add_dll_directory(os.fspath(Path(module_path).parent.resolve())))
|
|
|
|
|
|
# NB: The types implemented in C cannot inherit from abc.ABC since Python 3.12
|
|
# so we have to implement the protocols like this instead.
|
|
# https://github.com/python/cpython/issues/103968#issuecomment-1589928055
|
|
|
|
|
|
def mixin_sequence(typ: type) -> None:
|
|
"""
|
|
Adds missing Python mapping methods to types that implement IVectorView and
|
|
registers the type as a Sequence.
|
|
"""
|
|
# mixin methods
|
|
if not hasattr(typ, "index"):
|
|
typ.index = Sequence.index # type: ignore
|
|
|
|
if not hasattr(typ, "count"):
|
|
typ.count = Sequence.count # type: ignore
|
|
|
|
if not hasattr(typ, "__contains__"):
|
|
typ.__contains__ = Sequence.__contains__ # type: ignore
|
|
|
|
if not hasattr(typ, "__iter__"):
|
|
typ.__iter__ = Sequence.__iter__ # type: ignore
|
|
|
|
if not hasattr(typ, "__reversed__"):
|
|
typ.__reversed__ = Sequence.__reversed__ # type: ignore
|
|
|
|
Sequence.register(typ)
|
|
|
|
|
|
def mixin_mutable_sequence(typ: type) -> None:
|
|
"""
|
|
Adds missing Python mapping methods to types that implement IVector and
|
|
registers the type as a MutableSequence.
|
|
"""
|
|
mixin_sequence(typ)
|
|
|
|
if not hasattr(typ, "insert") and hasattr(typ, "insert_at"):
|
|
|
|
def insert(self: Any, index: int, value: object) -> None:
|
|
"""
|
|
Alias for ``insert_at`` for compatibility with Python Sequence protocol.
|
|
"""
|
|
self.insert_at(index, value)
|
|
|
|
typ.insert = insert # type: ignore
|
|
|
|
# mixin methods
|
|
if not hasattr(typ, "append"):
|
|
typ.append = MutableSequence.append # type: ignore
|
|
|
|
if not hasattr(typ, "clear"):
|
|
typ.clear = MutableSequence.clear # type: ignore
|
|
|
|
if not hasattr(typ, "extend"):
|
|
typ.extend = MutableSequence.extend # type: ignore
|
|
|
|
if not hasattr(typ, "reverse"):
|
|
typ.reverse = MutableSequence.reverse # type: ignore
|
|
|
|
if not hasattr(typ, "pop"):
|
|
typ.pop = MutableSequence.pop # type: ignore
|
|
|
|
if not hasattr(typ, "remove"):
|
|
typ.remove = MutableSequence.remove # type: ignore
|
|
|
|
if not hasattr(typ, "__iadd__"):
|
|
typ.__iadd__ = MutableSequence.__iadd__ # type: ignore
|
|
|
|
MutableSequence.register(typ)
|
|
|
|
|
|
def mixin_mapping(typ: type) -> None:
|
|
"""
|
|
Adds missing Python mapping methods to types that implement IMapView and
|
|
registers the type as a Mapping.
|
|
"""
|
|
# mixin methods
|
|
if not hasattr(typ, "keys"):
|
|
typ.keys = Mapping.keys # type: ignore
|
|
|
|
if not hasattr(typ, "items"):
|
|
typ.items = Mapping.items # type: ignore
|
|
|
|
if not hasattr(typ, "values"):
|
|
typ.values = Mapping.values # type: ignore
|
|
|
|
if not hasattr(typ, "get"):
|
|
typ.get = Mapping.get # type: ignore
|
|
|
|
if not hasattr(typ, "__contains__"):
|
|
typ.__contains__ = Mapping.__contains__ # type: ignore
|
|
|
|
# HACK: Version check works around inheritance anomaly caused by hacky
|
|
# metaclass inheritance implementation in runtime.cpp. This works as long
|
|
# as we don't change the projection to implement rich comparison methods
|
|
# on mapping types.
|
|
|
|
if (
|
|
typ.__eq__ == object.__eq__
|
|
or typ.__eq__ == Object.__eq__
|
|
or sys.version_info < (3, 12)
|
|
):
|
|
typ.__eq__ = Mapping.__eq__ # type: ignore
|
|
|
|
if (
|
|
typ.__ne__ == object.__ne__
|
|
or typ.__ne__ == Object.__ne__
|
|
or sys.version_info < (3, 12)
|
|
):
|
|
typ.__ne__ = Mapping.__ne__ # type: ignore
|
|
|
|
Mapping.register(typ)
|
|
|
|
|
|
def mixin_mutable_mapping(typ: type) -> None:
|
|
"""
|
|
Adds missing Python mapping methods to types that implement IMap and
|
|
registers the type as a MutableMapping.
|
|
"""
|
|
mixin_mapping(typ)
|
|
|
|
# mixin methods
|
|
if not hasattr(typ, "clear"):
|
|
typ.clear = MutableMapping.clear # type: ignore
|
|
|
|
if not hasattr(typ, "pop"):
|
|
typ.pop = MutableMapping.pop # type: ignore
|
|
# private-name-mangled attribute :-(
|
|
typ._MutableMapping__marker = MutableMapping._MutableMapping__marker # type: ignore
|
|
|
|
if not hasattr(typ, "popitem"):
|
|
typ.popitem = MutableMapping.popitem # type: ignore
|
|
|
|
if not hasattr(typ, "setdefault"):
|
|
typ.setdefault = MutableMapping.setdefault # type: ignore
|
|
|
|
if not hasattr(typ, "update"):
|
|
typ.update = MutableMapping.update # type: ignore
|
|
|
|
MutableMapping.register(typ)
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
# MyPy wants covariant T but Pylance wants invariant. Invariant seems correct.
|
|
class AsyncOp(Protocol[T]): # type: ignore [misc]
|
|
"""
|
|
Protocol that matches both IAsyncAction and IAsyncOperation.
|
|
"""
|
|
|
|
def cancel(self) -> None: ...
|
|
@property
|
|
def error_code(self) -> "HResult": ...
|
|
@property
|
|
def status(self) -> "AsyncStatus": ...
|
|
def get_results(self) -> T: ...
|
|
@property
|
|
def completed(self) -> Callable[[Self, "AsyncStatus"], None]: ...
|
|
@completed.setter
|
|
def completed(self, value: Callable[[Self, "AsyncStatus"], None]) -> None: ...
|
|
|
|
|
|
def wrap_async(op: AsyncOp[T]) -> Awaitable[T]:
|
|
"""
|
|
Wraps a WinRT async operation in a Python Awaitable.
|
|
|
|
This is used by the runtime to implement ``__await__`` on async operations.
|
|
|
|
Args:
|
|
op: The WinRT async operation to wrap.
|
|
|
|
Returns:
|
|
An asyncio Awaitable that will be completed when the WinRT operation
|
|
completes.
|
|
"""
|
|
|
|
async def wait() -> T:
|
|
loop = asyncio.get_running_loop()
|
|
event = asyncio.Event()
|
|
|
|
def on_complete(op: AsyncOp[T], status: "AsyncStatus") -> None:
|
|
loop.call_soon_threadsafe(event.set)
|
|
|
|
op.completed = on_complete
|
|
|
|
try:
|
|
await event.wait()
|
|
except asyncio.CancelledError:
|
|
op.cancel()
|
|
# REVISIT: what if it is cancelled again?
|
|
await event.wait()
|
|
raise
|
|
|
|
return op.get_results()
|
|
|
|
return wait()
|