Recovered lost code from other libraries

This commit is contained in:
Queue A
2026-06-03 06:36:09 +02:00
parent d66d2fd4ca
commit efdf7d6f0f
4 changed files with 80 additions and 24 deletions

31
bash/init_pyenv.sh Normal file
View File

@@ -0,0 +1,31 @@
init_pyenv(){
vtuple="(${1//./, })";
python3 -c "import sys; exit(sys.version_info <= ($vtuple))";
pyvok=$?;
if [ $pyvok -eq 1 ]; then
add-apt-repository -y ppa:deadsnakes/ppa
apt-get update
apt-get -y install "python$1" "python$1-venv"
"python$1" -m venv .env
else
python3 -m venv .env
fi;
source .env/bin/activate;
if [ -z "$2" ]; then
reqfile="requirements.txt";
else
reqfile="$2";
fi
if [ -z "$3" ]; then
piplog="/dev/null";
else
piplog="$3";
fi
pip install -r "$reqfile" >> "$piplog";
echo "PIP status: $?" >> "$piplog";
}

13
bash/thismux.sh Normal file
View File

@@ -0,0 +1,13 @@
thismux() {
dir="$(pwd)"
# Search upwards for `.run/thismux.sock`
while [ "$dir" != "/" ]; do
if [ -d "$dir/.run" ]; then
tmux -S "$dir/.run/thismux.sock" "$@"
return
fi
dir="$(dirname "$dir")"
done
echo "No .run found in any parent directory."
return 1
}

View File

@@ -1,5 +1,5 @@
#
# Last Change: 2025-10-25
# Last Change: 2026-04-04
#
import functools
import traceback
@@ -18,8 +18,7 @@ class AsyncOriented:
- They will track all their internal tasks if self.create_task is used,
and will perform cleanup on tasks that are done,
once the task counter has passed a certain number since the last cleanup,
or the cleanup function is explicitly called. (The cleanup is done in a sync manner)
once a task is finished, it'll remove itself from the tracked_tasks set.
Notable Fields / Methods:
@@ -39,7 +38,6 @@ class AsyncOriented:
- async self.cleanup()
"""
TRACKED_TASKS_CLEANUP_AFTER = 64
LOG_NAME = __name__
LOG_FMT = r"%(asctime)s [%(process)d] [%(levelname)s] %(message)s", r"[%Y-%m-%d %H:%M:%S %z]"
@@ -69,17 +67,14 @@ class AsyncOriented:
async def startup( self ):
assert not hasattr(self, "loop"), f"This AsyncOriented object's startup has already been called once."
self.loop = asyncio.get_running_loop()
self.tracked_tasks = []
self.tracked_tasks_next_cleanup = self.TRACKED_TASKS_CLEANUP_AFTER
self.tracked_tasks = set()
async def cleanup( self ):
self.garbage_collect_tasks()
msg = "Object's lifetime is over, cleanup is called."
for task in self.tracked_tasks:
task.cancel( msg = msg )
await asyncio.gather( *self.tracked_tasks, return_exceptions = True )
self.garbage_collect_tasks()
def wrap_coro_to_log_exceptions( self, coro, name = None ):
"""
@@ -113,27 +108,13 @@ class AsyncOriented:
if debug: coro = self.wrap_coro_to_log_exceptions( coro, name = name )
task = self.loop.create_task( coro, name = name, context = context )
self.tracked_tasks.append( task )
if self.tracked_tasks_next_cleanup <= len(self.tracked_tasks):
self.garbage_collect_tasks()
self.tracked_tasks.add( task )
task.add_done_callback( self.tracked_tasks.discard )
return task
def garbage_collect_tasks( self ):
task_count = len(self.tracked_tasks) #For the log
for t in list(self.tracked_tasks):
if t.done(): self.tracked_tasks.remove(t)
self.tracked_tasks_next_cleanup = self.TRACKED_TASKS_CLEANUP_AFTER + len(self.tracked_tasks)
self.log.debug(
f"Cleaned up {task_count - len(self.tracked_tasks)} of {task_count} tasks,"
f" next cleanup at: {self.tracked_tasks_next_cleanup}"
)

31
python/singelton.py Normal file
View File

@@ -0,0 +1,31 @@
#
# Last Change: 2026-03-18
#
import threading
class Singleton:
"""
A thread lock protected singleton,
makes any inherting class into a singlton.
The SINGLETON_LOCK object is shared between all inherited children,
unless they initiate their own using the 'decouple_lock_from_parent_class' method.
"""
SINGLETON_INSTANCE = None
SINGLETON_LOCK = threading.Lock()
@classmethod
def decouple_lock_from_parent_class( cls ):
if cls.SINGLETON_LOCK is not super().SINGLETON_LOCK: return
cls.SINGLETON_LOCK = threading.Lock()
def __new__( cls, *args, **kwargs ):
if cls.SINGLETON_INSTANCE: return cls.SINGLETON_INSTANCE
with cls.SINGLETON_LOCK:
if cls.SINGLETON_INSTANCE: return cls.SINGLETON_INSTANCE
cls.SINGLETON_INSTANCE = super().__new__( cls, *args, **kwargs )
return cls.SINGLETON_INSTANCE