Skip to content

Instantly share code, notes, and snippets.

@ArtemGr
Last active Sep 24, 2022
Embed
What would you like to do?
llog
/__pycache__
/llog
#!/usr/bin/env python
# Install with
#
# python -mpip install --user --upgrade git+https://gist.github.com/bf91613a021a536c7ce16cdba9168604.git
import datetime
import math
import os
import sys
from inspect import currentframe, getframeinfo
def stdout2utf8():
'''https://stackoverflow.com/questions/3597480/how-to-make-python-3-print-utf8'''
#sys.stdout = open(1, 'w', encoding='utf-8', closefd=False)
sys.stdout.reconfigure(encoding='utf-8')
# rindex that does not throw
def rindexʹ(s, ss):
try:
return s.rindex(ss)
except ValueError:
return -1
def log(*args, sep=None, end=None, file=None, flush=False, back=1):
frame = currentframe()
for _ in range(back):
frame = frame.f_back
info = getframeinfo(frame)
name = info.filename
slash = max(rindexʹ(name, "/"), rindexʹ(name, "\\")) + 1
name = name[slash:]
if name.endswith(".py"):
name = name[:-3]
args = list(args)
args.insert(0, f"{name}:{frame.f_lineno}]")
args.insert(0, datetime.datetime.now().strftime("%H:%M:%S"))
print(*args, sep=sep, end=end, file=file, flush=flush)
LOGC_PAST = {}
def logc(line):
'''only log the `line` when it changes'''
frame = currentframe()
frame = frame.f_back
info = getframeinfo(frame)
name = info.filename
slash = max(rindexʹ(name, "/"), rindexʹ(name, "\\")) + 1
name = name[slash:]
if name.endswith(".py"):
name = name[:-3]
loc = f"{name}:{frame.f_lineno}"
if LOGC_PAST.get(loc) == line:
return False
tim = datetime.datetime.now().strftime("%H:%M:%S")
print(f"{tim} {loc}] {line}")
LOGC_PAST[loc] = line
return True
def floorʹ(v, extra=0):
'''drop more decimal places depending on whether the integer is large'''
if v == None:
return None
if math.isnan(v):
return v
av = abs(v)
if 10 <= av:
decimal_places = 0
elif 1 <= av:
decimal_places = 1
elif .1 <= av:
decimal_places = 2
elif .01 <= av:
decimal_places = 3
elif .001 <= av:
decimal_places = 4
elif .0001 <= av:
decimal_places = 5
else:
# See `format_float_positional` about printing these
decimal_places = 6
decimal_places += extra
if not decimal_places:
return int(v)
r = 10**decimal_places
return math.floor(v * r) / r
def floorᵃ(a, extra=0):
return list(map(lambda v: floorʹ(v, extra), a))
def bits2bedstead(ch):
'''
Row-major bits, 2x3, to [Bedstead](https://i.imgur.com/f3myFgM.png)
cf. https://youtu.be/5yoWxctJsYo graphics with teletext; Windows font rendering glitch
cf. https://github.com/saitoha/PySixel full-pixel color graphics on SIXEL-supported terminals
cf. https://stackoverflow.com/q/37288421/257568 termplotlib plotext uniplot terminalplot
'''
if 0b111111 < ch:
return chr(0xEE00)
if 32 <= ch:
return chr(0xEE40 + ch - 32)
if 0 < ch:
return chr(0xEE00 + ch)
return chr(0xEE00 + ch)
def bedstead2bits(ch):
'''[Bedstead](https://i.imgur.com/f3myFgM.png) to row-major bits, 2x3'''
ch = ord(ch) if len(ch) == 1 else 0
if 0xEE5F < ch: # Past G1
return 0
if 0xEE40 <= ch:
return ch - 0xEE40 + 32
if 0xEE00 <= ch:
return ch - 0xEE00
return 0 # Below G1
def replace_str(string, position, characters):
'''return a `string` with the `characters` at the given `position` replaced'''
# cf. https://stackoverflow.com/a/67361875/257568 fast replacement
return f"{string[:position]}{characters}{string[position+len(characters):]}"
class BedsteadMap:
def __init__(self, a, xs, ys, wofs=0, hofs=0, scale=True):
self.wofs = int(wofs)
self.hofs = int(hofs)
self.xsmin = float(min(xs))
self.xsmax = float(max(xs))
self.ysmin = float(min(ys))
self.ysmax = float(max(ys))
self.wh = (len(a[0]), len(a))
self.height = int(self.wh[1] * 3)
ε = 0.00006103515625 # f16: smallest positive normal number
self.coeff = float(self.height / max(ε, self.ysmax - self.ysmin))
self.scale = 1.0
if isinstance(scale, bool) and scale:
self.scale = (self.ysmax - self.ysmin) / (self.xsmax - self.xsmin) * (self.wh[0] * 2 / self.height)
self.width = min(self.wh[0] * 2, int((self.xsmax - self.xsmin) * self.coeff * self.scale + 1))
def zip(self, xs, ys):
for (x, y) in zip(xs, ys):
px = min(self.width - 1, int((float(x) - self.xsmin) * self.coeff * self.scale))
py = min(self.height - 1, int((self.ysmax - float(y)) * self.coeff))
ax = max(0, min(self.wh[0] - 1, px // 2 + self.wofs))
ay = max(0, min(self.wh[1] - 1, py // 3 + self.hofs))
bx = px % 2
by = py % 3
#print('x', x, 'y', y, 'px', px, 'py', py, 'ax', ax, 'ay', ay, 'bx', bx, 'by', by)
bit = 1 << (bx + by * 2)
yield bit, bx, by, px, py, ax, ay
def plot(a, xs, ys, wofs=0, hofs=0, scale=True):
'''
Example:
import shutil
wh = shutil.get_terminal_size ((111, 11))
a = [[' ' for x in range (wh.columns)] for y in range (3)]
map = plot (a, [1, 2, 3, 4, 5, 6, 7, 8, 9], [1, 2, 3, 4, 5, 4, 3, 2, 1])
for y in a: print (''.join (y) .rstrip())
'''
map = BedsteadMap(a, xs, ys, wofs, hofs, scale)
for bit, *_, ax, ay in map.zip(xs, ys):
bits = bedstead2bits(a[ay][ax])
bits |= bit
a[ay][ax] = bits2bedstead(bits)
return map
def has_color():
'''[CLICOLOR_FORCE](https://bixense.com/clicolors/) or `isatty`'''
return (os.getenv('CLICOLOR_FORCE', '0') != '0') or sys.stdout.isatty()
def c(color: int):
'''
https://en.wikipedia.org/wiki/ANSI_escape_code#8-bit
https://www.ditig.com/256-colors-cheat-sheet
if :func:`has_color`
'''
return f"\033[38;5;{color}m" if has_color() else ''
def esc0():
'''`\\033[0m` if :func:`has_color`'''
return '\033[0m' if has_color() else ''
if __name__ == '__main__':
import shutil
# clean the cache for library users
shutil.rmtree('__pycache__', ignore_errors=True)
for i in range(0, 2**6):
log(i, bits2bedstead(i), bedstead2bits(bits2bedstead(i)))
wh = shutil.get_terminal_size((111, 11))
a = [[' ' for x in range(wh.columns)] for y in range(3)]
plot(a, [.1, .15, .2, .3, .4, .5, .6, .7], [.1, .14, .15, .2, .25, .3, .25, .2], scale=False)
print('\n'.join(''.join(y).rstrip() for y in a))
#!/usr/bin/env python
import sys
def dump(path, size=False):
from llog import state
pv = state.State(path).dump(size=size)
try:
import yaml
print(yaml.dump(pv))
except ModuleNotFoundError:
import json
print(json.dumps(pv, ensure_ascii=False, indent=2))
if __name__ == '__main__':
size = False
for arg in sys.argv:
if arg == '--size':
size = True
if arg.startswith('--dump='):
dump(arg[7:], size)
exit(0)
print('Command line examples:')
print()
print(' python -mllog [--size] --dump=state.mdb')
class fragile(object):
'''
Fragile `with`, https://stackoverflow.com/a/23665658/257568
```
with fragile(open(path)) as f:
print 'before condition'
if condition:
raise fragile.Rollback
print 'after condition'
```
'''
class Rollback(Exception):
'''break out of the `with` statement with a failure'''
class Commit(Exception):
'''break out of the `with` statement with a success'''
def __init__(self, value):
self.value = value
def __enter__(self):
return self.value.__enter__()
def __exit__(self, etype, value, traceback):
if etype == self.Commit:
self.value.__exit__(None, None, None)
return True # consume the exception
error = self.value.__exit__(etype, value, traceback)
if etype == self.Rollback:
return True # consume the exception
return error
from setuptools import setup
import os
import shutil
if __name__ == "__main__":
if not os.path.exists('llog'):
os.mkdir('llog')
shutil.copyfile('__init__.py', 'llog/__init__.py')
shutil.copyfile('__main__.py', 'llog/__main__.py')
shutil.copyfile('state.py', 'llog/state.py')
shutil.copyfile('control.py', 'llog/control.py')
setup(
name='llog',
version='2.0.2',
description='Log with line number. Bedstead plots. State.',
author='Artemciy',
author_email='artemciy@gmail.com',
packages=['llog'], # Same as name
# NB: “state.py” requires “lmdb” and “cbor2”, but one doesn't have to use it,
# they are *optional* so to speak.
# ⌥ use square bracket dependencies hence
install_requires=[], # External packages as dependencies
)
#!/usr/bin/env python
from datetime import timezone
import cbor2
import lmdb
# Python module: https://lmdb.readthedocs.io/en/release/
# Binary serialization for the fields: https://docs.python.org/3/library/struct.html
# https://docs.python.org/3/reference/compound_stmts.html#with
# https://docs.python.org/3/reference/datamodel.html#context-managers
# https://docs.python.org/3/reference/datamodel.html?emulating-container-types#emulating-container-types
# A short example at https://stackoverflow.com/a/23976949/257568
# https://docs.python.org/3/library/stdtypes.html#bytes-and-bytearray-operations
def str2byt(sob):
if isinstance(sob, str):
return sob.encode()
else:
return sob
class TranIter:
def __init__(self, cur):
self.cur = cur
def __next__(self):
if self.cur:
if self.cur.next():
return str(self.cur.key(), 'utf-8')
else:
self.cur.close()
self.cur = None
raise StopIteration
else:
raise StopIteration
def __del__(self):
if self.cur:
self.cur.close()
class Tran:
def __init__(self, env, dbn=b'default', write=True):
self.env = env
self.dbn = str2byt(dbn)
self.write = write
def __enter__(self):
self.txn = self.env.begin(write=self.write, buffers=True)
self.db = self.env.open_db(self.dbn, txn=self.txn)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type == None:
self.txn.commit()
else:
self.txn.abort()
def clear(self, drop=False):
self.txn.drop(self.db, delete=drop)
def __delitem__(self, key):
if not self.txn.delete(str2byt(key), db=self.db):
raise KeyError(f"No such key: “{key}”")
def __setitem__(self, key, value):
cv = cbor2.dumps(value, timezone=timezone.utc, value_sharing=True, string_referencing=True)
self.txn.put(str2byt(key), cv, db=self.db)
async def amem(self, key, vf):
'''
memoize the key, invoking `vf` to generate a missing value
```
with state.begin() as st:
foo = await st.amem('foo', lambda: return_coroutine())
```
'''
cur = self.txn.cursor(self.db)
key = str2byt(key)
found = cur.set_key(key)
if found:
pv = cbor2.loads(cur.value())
else:
pv = await vf()
cv = cbor2.dumps(pv, timezone=timezone.utc, value_sharing=True, string_referencing=True)
cur.put(key, cv)
cur.close()
return pv
def mem(self, key, vf):
'''memoize the key, invoking `vf` to generate a missing value'''
cur = self.txn.cursor(self.db)
key = str2byt(key)
found = cur.set_key(key)
if found:
pv = cbor2.loads(cur.value())
else:
pv = vf()
cv = cbor2.dumps(pv, timezone=timezone.utc, value_sharing=True, string_referencing=True)
cur.put(key, cv)
cur.close()
return pv
def get(self, key, default=None):
mv = self.txn.get(str2byt(key), db=self.db)
if mv == None:
return default
# https://docs.python.org/3/library/stdtypes.html#memoryview
pv = cbor2.loads(mv)
mv.release()
return pv
def __getitem__(self, key):
return self.get(key)
def __contains__(self, key):
cur = self.txn.cursor(self.db)
found = cur.set_key(str2byt(key))
cur.close()
return found
def __iter__(self):
'''
```
with state.begin('lemmings') as st:
lemmings = {name: st[name] for name in st}
```
'''
return TranIter(self.txn.cursor(self.db))
def items(self):
'''Note that value is a copy, save it back with assignment to persist.'''
cur = self.txn.cursor(self.db)
while cur.next():
key = str(cur.key(), 'utf-8')
val = cbor2.loads(cur.value())
yield (key, val)
cur.close()
def save(self, key: str, empty_collection):
'''Context Manager persisting the referenced collection on a clean `__exit__`'''
return Saviour(self, key, empty_collection)
class Saviour:
'''Context Manager for a state-hosted collection (like `dict`), saving it back on `__exit__`'''
def __init__(self, st: Tran, key: str, empty_collection):
self.st = st
self.key = key
self.empty_collection = empty_collection
def __enter__(self):
self.collection = self.st.get(self.key, self.empty_collection)
return self.collection
def __exit__(self, exc_type, exc_val, exc_tb):
self.st[self.key] = self.collection
class StateSaviour:
'''Context Manager for a state-hosted collection (like `dict`), saving it back on `__exit__`'''
def __init__(self, state, key: str, empty_collection):
self.state = state
self.key = key
self.empty_collection = empty_collection
def __enter__(self):
self.collection = self.state.get(self.key, self.empty_collection)
return self.collection
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type == None:
self.state[self.key] = self.collection
class State:
def __init__(self, path, max_dbs=64):
self.env = lmdb.Environment(path,
metasync=False,
sync=False,
map_async=True,
writemap=True,
max_dbs=max_dbs)
def begin(self, dbn=b'default', write=True):
return Tran(self.env, dbn, write)
def cur2(self, callback, dbn=b'default', write=True):
'''run `callback` on a cursor'''
with self.begin(dbn, write) as st:
cur = st.txn.cursor(st.db)
callback(st, cur)
cur.close()
def __setitem__(self, key, value):
with self.begin() as st:
st[key] = value
async def amem(self, key, vf):
'''memoize the key, invoking coroutine-returning `vf` to generate a missing value'''
with self.begin() as st:
return await st.amem(key, vf)
def mem(self, key, vf):
'''memoize the key, invoking `vf` to generate a missing value'''
with self.begin() as st:
return st.mem(key, vf)
def get(self, key, default=None):
with self.begin() as st:
return st.get(key, default)
def __getitem__(self, key):
with self.begin() as st:
return st[key]
def __contains__(self, key):
with self.begin() as st:
return key in st
def save(self, key: str, empty_collection):
'''Context Manager persisting the referenced collection on a clean `__exit__`'''
return StateSaviour(self, key, empty_collection)
def dump(self, size=False, cbor=False):
'''
python -c "import json; from llog.state import *; print(json.dumps(State('state.mdb').dump(), indent=2))"
python -c "import yaml; from llog.state import *; print(yaml.dump(State('state.mdb').dump()))"
python -mllog --dump=state.mdb
atexit.register(lambda: open('state.yaml', 'w', encoding='utf-8').write(
yaml.dump(State('perpetual.mdb').dump(), allow_unicode=True, width=99)))
'''
dick = {}
with self.env.begin(write=False) as txn:
for dbn, _ in txn.cursor().iternext():
dick[dbn.decode()] = {}
for dbn in dick:
db = self.env.open_db(dbn.encode(), txn=txn)
dick_cheney = {}
for key, value in txn.cursor(db=db).iternext():
if size:
dick_cheney[key.decode()] = len(value)
elif cbor:
dick_cheney[key.decode()] = value
else:
pv = cbor2.loads(value)
dick_cheney[key.decode()] = pv
dick[dbn] = dick_cheney
return dick
if __name__ == '__main__':
import os
import shutil
import tempfile
import time
from llog import log
# clean the cache for library users
shutil.rmtree('__pycache__', ignore_errors=True)
log(f"Going to “{tempfile.gettempdir()}”…")
os.chdir(tempfile.gettempdir())
# to start with a clean slate
#shutil.rmtree('test-state.mdb')
instance = int(time.time() / 600)
state = State('test-state.mdb')
# Example of extending the database size
# which helps with “MDB_MAP_FULL: Environment mapsize limit reached”
# https://lmdb.readthedocs.io/en/release/#lmdb.Environment.set_mapsize
state.env.set_mapsize(11 * 1024 * 1024)
with state.begin() as st:
if st['instance'] != instance:
log(f"Instance {instance} is new, scrapping the state…")
st.clear()
st['instance'] = instance
tim = st.mem('tim', lambda: int(time.time()))
state['foo'] = 'bar'
bar = state.mem('bar', lambda: 40 + 2)
with state.begin('count') as ct:
ct['cnt'] = ct['cnt'] + 1 if ct['cnt'] != None else 0
assert 'cnt' in ct
with state.begin() as st:
with st.save('dick', {}) as dicst:
dicst['tim'] = int(time.time())
assert 'dick' in st
with state.save('dick', {}) as dicst:
dicst['sec'] = dicst['tim'] % 60
log(state.dump())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment