| # I used this shell.nix to build LineageOS 13.0 for my maguro (Samsung Galaxy Nexus GSM) phone | |
| # The build instructions for normal Linuxes are here: https://wiki.lineageos.org/devices/maguro/build | |
| # For NixOS, follow those instructions but skip anything related to installing packages | |
| # Detailed instructions: | |
| # cd into an empty directory of your choice | |
| # copy this file there | |
| # in nix-shell: | |
| # $ repo init -u https://github.com/LineageOS/android.git -b cm-13.0 | |
| # $ repo sync | |
| # $ source build/envsetup.sh |
| class UniversalQueue: | |
| def __init__(self, *args, **kwargs): | |
| self._queue = trio.Queue(*args, **kwargs) | |
| self._portal = trio.BlockingTrioPortal() | |
| async def trio_get(self): | |
| return await self._queue.trio_get() | |
| def thread_get(self): | |
| return self._portal.run(self._queue.trio_get) |
| from kivy.factory import Factory as F | |
| from kivy.app import App | |
| from kivy.lang import Builder | |
| KV = ''' | |
| <ButtonLayout>: | |
| on_press: print("here or there") | |
| Label: |
| import math | |
| import trio | |
| class AsyncQueue: | |
| def __init__(self): | |
| self.history = [] | |
| self.inport, self.outport = trio.open_memory_channel(math.inf) |
| import trio | |
| import sys | |
| import time | |
| import httpx | |
| from outcome import Error | |
| import traceback | |
| # Can't use PySide2 currently because of | |
| # https://bugreports.qt.io/projects/PYSIDE/issues/PYSIDE-1313 | |
| from PyQt5 import QtCore, QtWidgets |
| class EnterTimeout: | |
| def __init__(self, async_cm, timeout): | |
| self._async_cm = async_cm | |
| self._timeout = timeout | |
| async def __aenter__(self): | |
| with fail_after(self.timeout): | |
| return await self._async_cm.__aenter__() | |
| async def __aexit__(self, *args): |
| import trio | |
| import weakref | |
| import cachetools | |
| # Note: this assumes you only have one trio thread running at once... | |
| # In the unlikely event that you have multiple, you should put the locks and cache | |
| # into thread-local or run-local storage. | |
| def trio_lru_cache(fn): | |
| cache = cachetools.LRUCache() | |
| # Trick: this contains a lock for each unique cache key in use, |
| """ | |
| Example of running an embedded IPython shell inside an already-running trio loop with working autoawait (it's handy | |
| to be able to start an interactive REPL with your application environment fully initialized). This is a full solution | |
| that works around https://github.com/ipython/ipython/issues/680 (see | |
| https://gist.github.com/mikenerone/3640fdd450b4ca55ee8df4d4da5a7165 for how simple it *could* be). This bug should be | |
| fixed IMO (using atexit is a questionable design choice in the first place given the embedding feature of IPython | |
| IMO). As it is now, the entire IPythonAtExitContext context manager exists only to work around the problem, | |
| otherwise it would result in an error on process exit when IPython's atexit-registered method calls fail to save the | |
| input history. |
| # Rough draft of a Queue object that can be used simultaneously simultaneously from | |
| # sync threads + *multiple* trio threads, all at once. | |
| # | |
| # If you don't have multiple threads each doing their own separate calls to trio.run, | |
| # then don't use this; there are simpler solutions. This was mostly an exercise to | |
| # figure out if and how this could be done. | |
| # | |
| # Also note: completely untested, probably has bugs | |
| from collections import OrderedDict |
