+ Added sphinx style pydocs.
3. The timeout decortator in decorator_utils.py is based on original
- work published in ActiveState code recipes and covered by the PSF
- license. It is from here:
+ work by Stephen "Zero" Chappell published in ActiveState code
+ recipes and covered by the PSF license. It is from here:
https://code.activestate.com/recipes/307871-timing-out-function/
+ The original PSF license text is included in the relevant section
+ of decorator_utils.py.
+
Scott's modifications include:
+ Adding docs + comments including a doctest unittest,
+ Minor cleanup and style tweaks,
# pyutils
----
+When I was writing little tools in Python and found myself implementing
+a generally useful pattern I stuffed it into a local library. That
+library grew into pyutils: a set of collections, helpers and utilities
+that I find useful and hope you will too.
-This is a collection of Python utilities that I wrote and find useful.
-From collections that try to emulate Pythonic patterns
-(pyutils.collectionz) to a "smart" natural language date parser
-(pyutils.datetimez.dateparse_utils), to filesystem helpers
-(pyutils.files.file_utils) to a "simple" parallelization framework
-(pyutils.parallelize.parallelize). I hope you find them useful, too.
+Code is under [src/pyutils/](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=src/pyutils;h=e716e14b7a895e5c6206af90f4628bf756f040fe;hb=HEAD)
+Most code includes inlint documentation and doctests. I've tried to
+organize it into logical packages based on the code's functionality.
+Note that when words would collide with a Python library or reserved
+word I've used a 'z' at the end, e.g. 'collectionz' instead of
+'collections', 'typez' instead of 'type', etc...
-Code is under `src/pyutils/*`. Most code includes doctests inline.
+There's some example code that uses various features of this project checked
+in under [examples/](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=examples;h=d9744bf2b171ba7a9ff21ae1d3862b673647fff4;hb=HEAD) that you can check out.
-Tests are under tests/*. To run all tests:
+Unit and integration tests are under [tests/](
+https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=tests;h=8c303f23cd89b6d2e4fbf214a5c7dcc0941151b4;hb=HEAD). To run all tests:
cd tests/
./run_tests.py --all [--coverage]
-See the README under tests/ and the code of run_tests.py for more
-options / information.
+See the [README](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=tests/README;hb=HEAD)) under `tests/` and the code of `run_tests.py` for more options / information about running tests.
-This package generates Sphinx docs which are available at:
-
- [https://wannabe.guru.org/pydocs/pyutils/pyutils.html](https://wannabe.guru.org/pydocs/pyutils/pyutils.html)
+This package generates Sphinx docs which are available at [https://wannabe.guru.org/pydocs/pyutils/pyutils.html](https://wannabe.guru.org/pydocs/pyutils/pyutils.html)
Package code is checked into a local git server and available to clone
-from https://wannabe.guru.org/git/pyutils.git or under:
-
- [https://wannabe.guru.org/gitweb/?p=pyutils.git;a=summary](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=summary)
+from git at https://wannabe.guru.org/git/pyutils.git or to view in a
+web browser at [https://wannabe.guru.org/gitweb/?p=pyutils.git;a=summary](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=summary)
For a long time this was just a local library on my machine that my
-tools imported but I've now decided to release it on PyPi. Early
-development happened in a different git repo:
-
- [https://wannabe.guru.org/gitweb/?p=python_utils.git;a=summary](https://wannabe.guru.org/gitweb/?p=python_utils.git;a=summary)
-
-I hope you find this useful. LICENSE and NOTICE describe reusing it
-and where everything came from. Drop me a line if you are using this,
-find a bug, or have a question.
+tools imported but I've now decided to release it on PyPi. Earlier
+development happened in a different git repo [https://wannabe.guru.org/gitweb/?p=python_utils.git;a=summary](https://wannabe.guru.org/gitweb/?p=python_utils.git;a=summary)
+The [LICENSE](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=LICENSE;hb=HEAD)
+and [NOTICE](https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD)
+files at the root of the project describe reusing this code and where
+everything came from. Drop me a line if you are using this, find a
+bug, have a question, or have a suggestion:
# Don't skip __init__()!
def skip(app, what, name, obj, would_skip, options):
- if name == "__init__":
- return False
if name == "__repr__":
return False
return would_skip
def setup(app):
app.connect("autodoc-skip-member", skip)
+
+
+autoclass_content = 'both'
library grew into pyutils: a set of collections, helpers and utilities
that I find useful and hope you will too.
-Code is under `src/pyutils/`. Most code includes documentation and inline
-doctests. I've tried to organize it into logical packages based on the
-code's functionality.
+Code is under `src/pyutils/
+<https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=src/pyutils;h=e716e14b7a895e5c6206af90f4628bf756f040fe;hb=HEAD>`_.
+Most code includes inlint documentation and doctests. I've tried to
+organize it into logical packages based on the code's functionality.
+Note that when words would collide with a Python library or reserved
+word I've used a 'z' at the end, e.g. 'collectionz' instead of
+'collections', 'typez' instead of 'type', etc...
-Unit and integration tests are under `tests/*`. To run all tests::
+There's some example code that uses various features of this project checked
+in under `examples/ <https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=examples;h=d9744bf2b171ba7a9ff21ae1d3862b673647fff4;hb=HEAD>`_ that you can check out.
+
+Unit and integration tests are under `tests/
+<https://wannabe.guru.org/gitweb/?p=pyutils.git;a=tree;f=tests;h=8c303f23cd89b6d2e4fbf214a5c7dcc0941151b4;hb=HEAD>`_.
+To run all tests::
cd tests/
./run_tests.py --all [--coverage]
-See the README under `tests/` and the code of `run_tests.py` for more
+See the `README <https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=tests/README;hb=HEAD>`_ under `tests/` and the code of `run_tests.py` for more
options / information about running tests.
This package generates Sphinx docs which are available at:
https://wannabe.guru.org/gitweb/?p=python_utils.git;a=summary
-The LICENSE and NOTICE files at the root of the project describe
-reusing this code and where everything came from. Drop me a line
-if you are using this, find a bug, have a question, or have a
-suggestion:
+The `LICENSE
+<https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=LICENSE;hb=HEAD>`_
+and `NOTICE
+<[https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD>`_
+files at the root of the project describe reusing this code and where
+everything came from. Drop me a line if you are using this, find a
+bug, have a question, or have a suggestion:
pyutils.dict\_utils module
--------------------------
-A bunch of helpers for dealing with Python dicts.
-
----
-
.. automodule:: pyutils.dict_utils
:members:
:undoc-members:
pyutils.exec\_utils module
--------------------------
-Helper code for dealing with subprocesses.
-
----
-
.. automodule:: pyutils.exec_utils
:members:
:undoc-members:
pyutils.function\_utils module
------------------------------
-Helper util for dealing with functions.
-
----
-
.. automodule:: pyutils.function_utils
:members:
:undoc-members:
pyutils.id\_generator module
----------------------------
-Generate unique identifiers.
-
----
-
.. automodule:: pyutils.id_generator
:members:
:undoc-members:
pyutils.iter\_utils module
--------------------------
-Iterator utilities including a :py:class:`PeekingIterator`, :py:class:`PushbackIterator`,
-and :py:class:`SamplingIterator`.
-
----
-
.. automodule:: pyutils.iter_utils
:members:
:undoc-members:
pyutils.list\_utils module
--------------------------
-Utilities for dealing with Python lists.
-
----
-
.. automodule:: pyutils.list_utils
:members:
:undoc-members:
pyutils.logging\_utils module
-----------------------------
-This is a module that offers an opinionated take on how whole program
-logging should be initialized and controlled. It uses standard Python
-:py:mod:`logging` but gives you control, via commandline config, to:
-
- - Set the logging level of the program including overriding the
- logging level for individual modules,
- - Define the logging message format including easily adding a
- PID/TID marker on all messages to help with multithreaded debugging,
- - Control the destination (file, `sys.stderr`, syslog) of messages,
- - Control the facility and logging level used with syslog,
- - Squelch repeated messages,
- - Log probalistically,
- - Clear rogue logging handlers added by other imports.
-
----
-
.. automodule:: pyutils.logging_utils
:members:
:undoc-members:
pyutils.math\_utils module
--------------------------
-Helper utilities that are "mathy" such as a :py:class:`NumericPopulation` that
-makes population summary statistics available to your code quickly, GCD
-computation, literate float truncation, percentage <-> multiplier, prime
-number determination, etc...
-
----
-
.. automodule:: pyutils.math_utils
:members:
:undoc-members:
pyutils.misc\_utils module
--------------------------
-Miscellaneous utilities: are we running as root, and is a debugger attached?
-
----
-
.. automodule:: pyutils.misc_utils
:members:
:undoc-members:
pyutils.persistent module
-------------------------
-Persistent defines a class hierarchy and decorator for creating
-singleton classes that (optionally / conditionally) load their state
-from some external location and (optionally / conditionally) save their
-state to an external location at shutdown.
-
----
-
.. automodule:: pyutils.persistent
:members:
:undoc-members:
pyutils.remote\_worker module
-----------------------------
-This module defines a helper that is invoked by the remote executor to
-run pickled code on a remote machine. It is used by code marked with
-`@parallelize(method=Method.REMOTE)` in the parallelize framework.
-
----
-
.. automodule:: pyutils.remote_worker
:members:
:undoc-members:
pyutils.state\_tracker module
-----------------------------
-This module defines several classes (:py:class:`StateTracker`,
-:py:class:`AutomaticStateTracker`, and
-:py:class:`WaitableAutomaticStateTracker`) that can be used as base
-classes. These class patterns are meant to encapsulate and represent
-state that dynamically changes. These classes update their state
-(either automatically or when invoked to poll) and allow their callers
-to wait on state changes.
-
----
-
.. automodule:: pyutils.state_tracker
:members:
:undoc-members:
pyutils.string\_utils module
----------------------------
-A bunch of utilities for dealing with strings. Based on a really great
-starting library from Davide Zanotti, I've added a pile of other string
-functions so hopefully it will handle all of your string-needs.
+A bunch of utilities for dealing with strings. Based on a really
+great starting library from Davide Zanotti (forked from
+https://github.com/daveoncode/python-string-utils/tree/master/string_utils),
+I've added a pile of other string functions (see `NOTICE
+<[https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD>`_
+file in the root of this project for a detailed account of what was
+added and changed) so hopefully it will handle all of your
+string-needs.
---
pyutils.text\_utils module
--------------------------
-Utilities for dealing with and creating text chunks. For example:
-
- - Make a bar graph,
- - make a spark line,
- - left, right, center, justify text,
- - word wrap text,
- - indent text,
- - create a header line,
- - draw a box around some text.
-
----
-
.. automodule:: pyutils.text_utils
:members:
:undoc-members:
pyutils.unittest\_utils module
------------------------------
-Utilities to support smarter unit tests.
-
----
-
.. automodule:: pyutils.unittest_utils
:members:
:undoc-members:
pyutils.unscrambler module
--------------------------
-Unscramble scrambled English words quickly.
-
----
-
.. automodule:: pyutils.unscrambler
:members:
:undoc-members:
pyutils.zookeeper module
------------------------
-A helper module for dealing with Zookeeper that adds some functionality.
-
----
-
.. automodule:: pyutils.zookeeper
:members:
:undoc-members:
Module contents
---------------
----
-
.. automodule:: pyutils
:members:
:undoc-members:
* automatic support for :py:mod:`pyutils.config` (argument parsing, see
that module for details),
* The ability to break into pdb on unhandled exceptions (which is
- enabled/disabled via the commandline flag `--debug_unhandled_exceptions`),
- * automatic logging support from :py:mod:`pyutils.logging` controllable
+ enabled/disabled via the commandline flag :code:`--debug_unhandled_exceptions`),
+ * automatic logging support from :py:mod:`pyutils.logging_utils` controllable
via several commandline flags,
* the ability to optionally enable whole-program code profiling and reporting
- when you run your code using commandline flag `--run_profiler`,
+ when you run your code using commandline flag :code:`--run_profiler`,
* the ability to optionally enable import auditing via the commandline flag
- `--audit_import_events`. This logs a message whenever a module is imported
+ :code:`--audit_import_events`. This logs a message whenever a module is imported
*after* the bootstrap module itself is loaded. Note that other modules may
already be loaded when bootstrap is loaded and these imports will not be
logged. If you're trying to debug import events or dependency problems,
I suggest putting bootstrap very early in your import list and using this
flag.
* optional memory profiling for your program set via the commandline flag
- `--trace_memory`. This provides a report of python memory utilization
+ :code:`--trace_memory`. This provides a report of python memory utilization
at program termination time.
* the ability to set the global random seed via commandline flag for
reproducable runs (as long as subsequent code doesn't reset the seed)
- using the `--set_random_seed` flag,
+ using the :code:`--set_random_seed` flag,
* automatic program timing and reporting logged to the INFO log,
* more verbose error handling and reporting.
#!/usr/bin/env python3
-"""
-The MIT License (MIT)
+"""The MIT License (MIT)
Copyright (c) 2020 LuizaLabs
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
-This class is based on https://github.com/luizalabs/shared-memory-dict.
-For details about what is preserved from the original and what was changed
-by Scott, see NOTICE at the root of this module.
+This class is based on
+https://github.com/luizalabs/shared-memory-dict. For details about
+what is preserved from the original and what was changed by Scott, see
+`NOTICE
+<https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD>`_
+at the root of this module.
+
"""
import pickle
Subsequent processes may safely omit name and size args.
+ Args:
+ name: the name of the shared dict, only required for initial caller
+ size_bytes: the maximum size of data storable in the shared dict,
+ only required for the first caller.
+
"""
assert size_bytes is None or size_bytes > 0
self._serializer = PickleSerializer()
self.name = self.shared_memory.name
def get_name(self):
- """Returns the name of the shared memory buffer backing the dict."""
+ """
+ Returns:
+ The name of the shared memory buffer backing the dict.
+ """
return self.name
def _get_or_create_memory_block(
name: Optional[str] = None,
size_bytes: Optional[int] = None,
) -> shared_memory.SharedMemory:
+ """Internal helper."""
try:
return shared_memory.SharedMemory(name=name)
except FileNotFoundError:
return shared_memory.SharedMemory(name=name, create=True, size=size_bytes)
def _ensure_memory_initialization(self):
+ """Internal helper."""
with SharedDict.LOCK:
memory_is_empty = (
bytes(self.shared_memory.buf).split(SharedDict.NULL_BYTE, 1)[0] == b''
self.clear()
def _write_memory(self, db: Dict[Hashable, Any]) -> None:
+ """Internal helper."""
data = self._serializer.dumps(db)
with SharedDict.LOCK:
try:
raise ValueError("exceeds available storage") from e
def _read_memory(self) -> Dict[Hashable, Any]:
+ """Internal helper."""
with SharedDict.LOCK:
return self._serializer.loads(self.shared_memory.buf.tobytes())
@contextmanager
def _modify_dict(self):
+ """Internal helper."""
with SharedDict.LOCK:
db = self._read_memory()
yield db
def close(self) -> None:
"""Unmap the shared dict and memory behind it from this
- process. Called by automatically __del__"""
+ process. Called by automatically :meth:`__del__`.
+ """
if not hasattr(self, 'shared_memory'):
return
self.shared_memory.close()
def cleanup(self) -> None:
- """Unlink the shared dict and memory behind it. Only the last process should
- invoke this. Not called automatically."""
+ """Unlink the shared dict and memory behind it. Only the last process
+ should invoke this. Not called automatically."""
if not hasattr(self, 'shared_memory'):
return
with SharedDict.LOCK:
self.shared_memory.unlink()
def clear(self) -> None:
- """Clear the dict."""
+ """Clears the shared dict."""
self._write_memory({})
def copy(self) -> Dict[Hashable, Any]:
- """Returns a shallow copy of the dict."""
+ """
+ Returns:
+ A shallow copy of the shared dict.
+ """
return self._read_memory()
def __getitem__(self, key: Hashable) -> Any:
return repr(self._read_memory())
def get(self, key: str, default: Optional[Any] = None) -> Any:
- """Gets the value associated with key or a default."""
+ """
+ Args:
+ key: the key to lookup
+ default: the value returned if key is not present
+
+ Returns:
+ The value associated with key or a default.
+ """
return self._read_memory().get(key, default)
def keys(self) -> KeysView[Hashable]:
#!/usr/bin/env python3
# © Copyright 2021-2022, Scott Gasch
-# Portions (marked) below retain the original author's copyright.
+# A portion (marked) below retain the original author's copyright.
"""This is a grab bag of, hopefully, useful decorators."""
# © Copyright 2021-2022, Scott Gasch
-"""Helper functions for dealing with dictionaries."""
+"""This module contains helper functions for dealing with Python dictionaries."""
from itertools import islice
from typing import Any, Callable, Dict, Iterator, List, Tuple
) -> bool:
"""
Initialize a dict value (if it doesn't exist) or increments it (using the
- inc_function, which is customizable) if it already does exist. Returns
- True if the key already existed or False otherwise.
+ inc_function, which is customizable) if it already does exist.
+
+ Args:
+ d: the dict to increment or initialize a value in
+ key: the key to increment or initialize
+ init_value: default initial value
+ inc_function: Callable use to increment a value
+
+ Returns:
+ True if the key already existed or False otherwise
+
+ See also: :py:class:`collections.defaultdict` and
+ :py:class:`collections.Counter`.
>>> d = {}
>>> init_or_inc(d, "test")
False
>>> d
{'test': 2, 'ing': 1}
-
"""
if key in d.keys():
d[key] = inc_function(d[key])
def shard(d: Dict[Any, Any], size: int) -> Iterator[Dict[Any, Any]]:
"""
- Shards a dict into N subdicts which, together, contain all keys/values
- from the original unsharded dict.
+ Shards (i.e. splits) a dict into N subdicts which, together,
+ contain all keys/values from the original unsharded dict.
+
+ Args:
+ d: the input dict to be sharded (split)
+ size: the ideal shard size (number of elements per shard)
+
+ Returns:
+ A generator that yields subsequent shards.
+
+ .. note::
+
+ If `len(d)` is not an even multiple of `size` then the last
+ shard will not have `size` items in it. It will have
+ `len(d) % size` items instead.
+
+ >>> d = {
+ ... 'a': 1, 'b': 2, 'c': 3, 'd': 4, 'e': 5, 'f': 6,
+ ... 'g': 7, 'h': 8, 'i': 9, 'j': 10, 'k': 11, 'l': 12,
+ ... }
+ >>> for r in shard(d, 5):
+ ... r
+ {'a': 1, 'b': 2, 'c': 3, 'd': 4, 'e': 5}
+ {'f': 6, 'g': 7, 'h': 8, 'i': 9, 'j': 10}
+ {'k': 11, 'l': 12}
"""
items = d.items()
for x in range(0, len(d), size):
*,
aggregation_function: Callable[[Any, Any, Any], Any] = coalesce_by_creating_list,
) -> Dict[Any, Any]:
- """Merge N dicts into one dict containing the union of all keys /
- values in the input dicts. When keys collide, apply the
- aggregation_function which, by default, creates a list of values.
- See also several other alternative functions for coalescing values:
-
- * :meth:`coalesce_by_creating_set`
- * :meth:`coalesce_first_write_wins`
- * :meth:`coalesce_last_write_wins`
- * :meth:`raise_on_duplicated_keys`
- * or provive your own collision resolution code.
+ """Coalesce (i.e. combine) N input dicts into one output dict
+ ontaining the union of all keys / values in every input dict.
+ When keys collide, apply the aggregation_function which, by
+ default, creates a list of values with the same key in the output
+ dict.
+
+ Args:
+ inputs: an iterable set of dicts to coalesce
+ aggregation_function: a Callable to deal with key collisions; one of
+ the below functions already defined or your own strategy:
+
+ * :meth:`coalesce_by_creating_list` creates a list of values
+ with the same key in the output dict.
+ * :meth:`coalesce_by_creating_set` creates a set of values with
+ the same key in the output dict.
+ * :meth:`coalesce_first_write_wins` only preserves the first
+ value with a duplicated key. Others are dropped silently.
+ * :meth:`coalesce_last_write_wins` only preserves the last
+ value with a duplicated key. Others are dropped silently.
+ * :meth:`raise_on_duplicated_keys` raises an Exception on
+ duplicated keys; use when keys should never collide.
+ * Your own strategy; Callables will be passed the key and
+ two values and can return whatever they want which will
+ be stored in the output dict.
+
+ Returns:
+ The coalesced output dict.
>>> a = {'a': 1, 'b': 2}
>>> b = {'b': 1, 'c': 2, 'd': 3}
Traceback (most recent call last):
...
Exception: Key b is duplicated in more than one input dict.
-
"""
out: Dict[Any, Any] = {}
for d in inputs:
def item_with_max_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
- """Returns the key and value of the item with the max value in a dict.
+ """
+ Args:
+ d: a dict with comparable values
+
+ Returns:
+ The key and value of the item with the highest value in a
+ dict as a `Tuple[key, value]`.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> item_with_max_value(d)
def item_with_min_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
- """Returns the key and value of the item with the min value in a dict.
+ """
+ Args:
+ d: a dict with comparable values
+
+ Returns:
+ The key and value of the item with the lowest value in a
+ dict as a `Tuple[key, value]`.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> item_with_min_value(d)
def key_with_max_value(d: Dict[Any, Any]) -> Any:
- """Returns the key with the max value in the dict.
+ """
+ Args:
+ d: a dict with comparable keys
+
+ Returns:
+ The maximum key in the dict when comparing the keys with
+ each other.
+
+ .. note:: This code totally ignores values; it is comparing key
+ against key to find the maximum key in the keyspace.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> key_with_max_value(d)
def key_with_min_value(d: Dict[Any, Any]) -> Any:
- """Returns the key with the min value in the dict.
+ """
+ Args:
+ d: a dict with comparable keys
+
+ Returns:
+ The minimum key in the dict when comparing the keys with
+ each other.
+
+ .. note:: This code totally ignores values; it is comparing key
+ against key to find the minimum key in the keyspace.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> key_with_min_value(d)
def max_value(d: Dict[Any, Any]) -> Any:
- """Returns the maximum value in the dict.
+ """
+ Args:
+ d: a dict with compatable values
+
+ Returns:
+ The maximum value in the dict *without its key*.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> max_value(d)
3
-
"""
return item_with_max_value(d)[1]
def min_value(d: Dict[Any, Any]) -> Any:
- """Returns the minimum value in the dict.
+ """
+ Args:
+ d: a dict with comparable values
+
+ Returns:
+ The minimum value in the dict *without its key*.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> min_value(d)
1
-
"""
return item_with_min_value(d)[1]
def max_key(d: Dict[Any, Any]) -> Any:
- """Returns the maximum key in dict (ignoring values totally)
+ """
+ Args:
+ d: a dict with comparable keys
+
+ Returns:
+ The maximum key in dict (ignoring values totally)
+
+ .. note:: This code totally ignores values; it is comparing key
+ against key to find the maximum key in the keyspace.
>>> d = {'a': 3, 'b': 2, 'c': 1}
>>> max_key(d)
'c'
-
"""
return max(d.keys())
def min_key(d: Dict[Any, Any]) -> Any:
- """Returns the minimum key in dict (ignoring values totally)
+ """
+ Args:
+ d: a dict with comparable keys
+
+ Returns:
+ The minimum key in dict (ignoring values totally)
+
+ .. note:: This code totally ignores values; it is comparing key
+ against key to find the minimum key in the keyspace.
>>> d = {'a': 3, 'b': 2, 'c': 1}
>>> min_key(d)
'a'
-
"""
return min(d.keys())
"""Given two parallel lists (keys and values), create and return
a dict.
+ Args:
+ keys: list containing keys and no duplicated keys
+ values: a parallel list (to keys) containing values
+
+ Returns:
+ A dict composed of zipping the keys list and values list together.
+
>>> k = ['name', 'phone', 'address', 'zip']
>>> v = ['scott', '555-1212', '123 main st.', '12345']
>>> parallel_lists_to_dict(k, v)
{'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
-
"""
if len(keys) != len(values):
raise Exception("Parallel keys and values lists must have the same length")
def dict_to_key_value_lists(d: Dict[Any, Any]) -> Tuple[List[Any], List[Any]]:
- """
+ """Given a dict, decompose it into a list of keys and values.
+
+ Args:
+ d: a dict
+
+ Returns:
+ A tuple of two elements: the first is the keys list and the second
+ is the values list.
+
>>> d = {'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
>>> (k, v) = dict_to_key_value_lists(d)
>>> k
['name', 'phone', 'address', 'zip']
>>> v
['scott', '555-1212', '123 main st.', '12345']
-
"""
r: Tuple[List[Any], List[Any]] = ([], [])
for (k, v) in d.items():
timeout_seconds: Optional[float] = None,
) -> int:
"""Kick off a child process. Capture and emit all output that it
- produces on stdout and stderr in a character by character manner
- so that we don't have to wait on newlines. This was done to
- capture the output of a subprocess that created dots to show
- incremental progress on a task and render it correctly.
+ produces on stdout and stderr in a raw, character by character,
+ manner so that we don't have to wait on newlines. This was done
+ to capture, for example, the output of a subprocess that creates
+ dots to show incremental progress on a task and render it
+ correctly.
Args:
command: the command to execute
Returns:
the exit status of the subprocess once the subprocess has
- exited. Raises TimeoutExpired after killing the subprocess
+ exited. Raises `TimeoutExpired` after killing the subprocess
if the timeout expires.
Side effects:
def cmd_exitcode(command: str, timeout_seconds: Optional[float] = None) -> int:
- """Run a command silently and return its exit code once it has
- finished. If timeout_seconds is provided and the command runs too
- long it will raise a TimeoutExpired exception.
+ """Run a command silently in the background and return its exit
+ code once it has finished. If timeout_seconds is provided and the
+ command runs longer than timeout_seconds, raise a `TimeoutExpired`
+ exception.
Args:
command: the command to run
- timeout_seconds: the max number of seconds to allow the subprocess
- to execute or None to indicate no timeout
+ timeout_seconds: optional the max number of seconds to allow
+ the subprocess to execute or None to indicate no timeout
Returns:
the exit status of the subprocess once the subprocess has
def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
- """Run a command silently but raise subprocess.CalledProcessError if
- it fails and raise a TimeoutExpired if it runs too long.
+ """Run a command silently but raise
+ `subprocess.CalledProcessError` if it fails (i.e. returns a
+ non-zero return value) and raise a `TimeoutExpired` if it runs too
+ long.
Args:
- command: the command to run
- timeout_seconds: the max number of seconds to allow the subprocess
- to execute or None to indicate no timeout
+ command: the command to run timeout_seconds: the optional
+ max number of seconds to allow the subprocess to execute or
+ None to indicate no timeout
Returns:
No return value; error conditions (including non-zero child process
from dataclasses import dataclass
from typing import Literal, Optional
-from pyutils import config, decorator_utils
+from pyutils import argparse_utils, config, decorator_utils
from pyutils.datetimez import datetime_utils
cfg = config.add_commandline_args(f'Lockfile ({__file__})', 'Args related to lockfiles')
cfg.add_argument(
- '--lockfile_held_duration_warning_threshold_sec',
- type=float,
- default=60.0,
- metavar='SECONDS',
+ '--lockfile_held_duration_warning_threshold',
+ type=argparse_utils.valid_duration,
+ default=datetime.timedelta(60.0),
+ metavar='DURATION',
help='If a lock is held for longer than this threshold we log a warning',
)
logger = logging.getLogger(__name__)
duration = ts - self.locktime
if (
duration
- >= config.config['lockfile_held_duration_warning_threshold_sec']
+ >= config.config[
+ 'lockfile_held_duration_warning_threshold'
+ ].total_seconds()
):
# Note: describe duration briefly only does 1s granularity...
str_duration = datetime_utils.describe_duration_briefly(int(duration))
def function_identifier(f: Callable) -> str:
"""
- Given a callable function, return a string that identifies it.
- Usually that string is just __module__:__name__ but there's a
+ Given a named `Callable`, return a string that identifies it.
+ Usually that string is just "__module__:__name__" but there's a
corner case: when __module__ is __main__ (i.e. the callable is
defined in the same module as __main__). In this case,
f.__module__ returns "__main__" instead of the file that it is
- defined in. Work around this using pathlib.Path (see below).
+ defined in. Work around this using `pathlib.Path`.
+
+ Args:
+ f: a Callable
+
+ Returns:
+ A unique identifier for that callable in the format
+ module:function that avoids the pseudo-module '__main__'
>>> function_identifier(function_identifier)
'function_utils:function_identifier'
+
"""
if f.__module__ == '__main__':
# © Copyright 2021-2022, Scott Gasch
-"""A helper class for generating thread safe monotonically increasing
+"""
+A helper class for generating thread safe monotonically increasing
id numbers.
-
"""
import itertools
Returns a thread-safe, monotonically increasing id suitable for use
as a globally unique identifier.
+ Args:
+ name: the sequence identifier name.
+ start: the starting id (i.e. the first id that should be returned)
+
+ Returns:
+ An integer id such that within one sequence identifier name the
+ id returned is unique and is the maximum id ever returned.
+
>>> import id_generator
>>> id_generator.get('student_id')
0
if name not in generators:
generators[name] = itertools.count(start, 1)
x = next(generators[name])
- logger.debug("Generated next id %d", x)
+ logger.debug("Generated next id %d in sequence %s", x, name)
return x
# © Copyright 2021-2022, Scott Gasch
-"""A collection if :class:`Iterator` subclasses that can be composed
-with another iterator and provide extra functionality. e.g.
+"""A collection of :class:`Iterator` subclasses that can be composed
+with another iterator and provide extra functionality:
+ :class:`PeekingIterator`
+ :class:`PushbackIterator`
class PeekingIterator(Iterator):
"""An iterator that lets you :meth:`peek` at the next item on deck.
Returns None when there is no next item (i.e. when
- :meth:`__next__` will produce a StopIteration exception).
+ :meth:`__next__` will produce a `StopIteration` exception).
>>> p = PeekingIterator(iter(range(3)))
>>> p.__next__()
Traceback (most recent call last):
...
StopIteration
-
"""
def __init__(self, source_iter: Iterator):
+ """
+ Args:
+ source_iter: the iterator we want to peek at
+ """
self.source_iter = source_iter
self.on_deck: List[Any] = []
return item
def peek(self) -> Optional[Any]:
+ """Peek at the upcoming value on the top of our contained
+ :py:class:`Iterator` non-destructively (i.e. calling :meth:`__next__` will
+ still produce the peeked value).
+
+ Returns:
+ The value that will be produced by the contained iterator next
+ or None if the contained Iterator is exhausted and will raise
+ `StopIteration` when read.
+
+ """
if len(self.on_deck) > 0:
return self.on_deck[0]
try:
class PushbackIterator(Iterator):
- """An iterator that allows you to push items back
- onto the front of the sequence. e.g.
+ """An iterator that allows you to push items back onto the front
+ of the sequence so that they are produced before the items at the
+ front/top of the contained py:class:`Iterator`. e.g.
>>> i = PushbackIterator(iter(range(3)))
>>> i.__next__()
Traceback (most recent call last):
...
StopIteration
+
"""
def __init__(self, source_iter: Iterator):
return self.pushed_back.pop()
return self.source_iter.__next__()
- def push_back(self, item: Any):
+ def push_back(self, item: Any) -> None:
+ """Push an item onto the top of the contained iterator such that
+ the next time :meth:`__next__` is invoked we produce that item.
+
+ Args:
+ item: the item to produce from :meth:`__next__` next.
+ """
self.pushed_back.append(item)
class SamplingIterator(Iterator):
- """An iterator that simply echoes what source_iter produces but also
- collects a random sample (of size sample_size) of the stream that can
- be queried at any time.
+ """An :py:class:`Iterator` that simply echoes what its
+ `source_iter` produces but also collects a random sample (of size
+ `sample_size`) from the stream that can be queried at any time.
.. note::
- Until sample_size elements have been seen the sample will be
- less than sample_size elements in length.
+ Until `sample_size` elements have been produced by the
+ `source_iter`, the sample return will be less than `sample_size`
+ elements in length.
.. note::
- If sample_size is > len(source_iter) then it will produce a
- copy of source_iter.
+ If `sample_size` is >= `len(source_iter)` then this will produce
+ a copy of `source_iter`.
>>> import collections
>>> import random
return item
def get_sample(self) -> List[Any]:
+ """
+ Returns:
+ The current sample set populated randomly from the items
+ returned by the contained :class:`Iterator` so far.
+
+ .. note::
+ Until `sample_size` elements have been produced by the
+ `source_iter`, the sample return will be less than `sample_size`
+ elements in length.
+
+ .. note::
+ If `sample_size` is >= `len(source_iter)` then this will produce
+ a copy of `source_iter`.
+ """
return self.resovoir
# © Copyright 2021-2022, Scott Gasch
-"""Some useful(?) utilities for dealing with Lists."""
+"""This module contains helper functions for dealing with Python lists."""
import random
from collections import Counter
def shard(lst: List[Any], size: int) -> Iterator[Any]:
"""
- Yield successive size-sized shards from lst.
+ Shards (i.e. splits) a list into sublists of size `size` whcih,
+ together, contain all items in the original unsharded list.
+
+ Args:
+ lst: the original input list to shard
+ size: the ideal shard size (number of elements per shard)
+
+ Returns:
+ A generator that yields successive shards.
+
+ .. note::
+
+ If `len(lst)` is not an even multiple of `size` then the last
+ shard will not have `size` items in it. It will have
+ `len(lst) % size` items instead.
>>> for sublist in shard([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12], 3):
... [_ for _ in sublist]
[4, 5, 6]
[7, 8, 9]
[10, 11, 12]
-
"""
for x in range(0, len(lst), size):
yield islice(lst, x, x + size)
def flatten(lst: List[Any]) -> List[Any]:
"""
- Flatten out a list:
+ Flatten out a list. That is, for each item in list that contains
+ a list, remove the nested list and replace it with its items.
+
+ Args:
+ lst: the list to flatten
+
+ Returns:
+ The flattened list. See example.
>>> flatten([ 1, [2, 3, 4, [5], 6], 7, [8, [9]]])
[1, 2, 3, 4, 5, 6, 7, 8, 9]
-
"""
if len(lst) == 0:
return lst
def prepend(item: Any, lst: List[Any]) -> List[Any]:
"""
- Prepend an item to a list.
+ Prepend an item to a list. An alias for `list.insert(0, item)`.
+ The opposite of `list.append()`.
+
+ Args:
+ item: the item to be prepended
+ lst: the list on which to prepend
+
+ Returns:
+ The list with item prepended.
>>> prepend('foo', ['bar', 'baz'])
['foo', 'bar', 'baz']
-
"""
lst.insert(0, item)
return lst
"""
Remove the list and return the 0th element iff its length is one.
+ Args:
+ lst: the List to check
+
+ Returns:
+ Either `lst` (if `len(lst) > 1`) or `lst[0]` (if `len(lst) == 1`).
+
>>> remove_list_if_one_element([1234])
1234
>>> remove_list_if_one_element([1, 2, 3, 4])
[1, 2, 3, 4]
-
"""
if len(lst) == 1:
return lst[0]
"""
Return a population count mapping for the list (i.e. the keys are
list items and the values are the number of occurrances of that
- list item in the original list.
+ list item in the original list). Note: this is used internally
+ to implement :meth:`most_common` and :meth:`least_common`.
+
+ Args:
+ lst: the list whose population should be counted
+
+ Returns:
+ a `Counter` containing the population count of `lst` items.
>>> population_counts([1, 1, 1, 2, 2, 3, 3, 3, 4])
Counter({1: 3, 3: 3, 2: 2, 4: 1})
-
"""
return Counter(lst)
def most_common(lst: List[Any], *, count=1) -> Any:
-
"""
- Return the most common item in the list. In the case of ties,
- which most common item is returned will be random.
+ Return the N most common item in the list.
+
+ Args:
+ lst: the list to find the most common item in
+ count: the number of most common items to return
+
+ Returns:
+ The most common item in `lst`.
+
+ .. warning::
+
+ In the case of ties for most common item, which most common
+ item is returned is undefined.
>>> most_common([1, 1, 1, 2, 2, 3, 3, 3, 3, 4, 4])
3
def least_common(lst: List[Any], *, count=1) -> Any:
"""
- Return the least common item in the list. In the case of
- ties, which least common item is returned will be random.
+ Return the N least common item in the list.
+
+ Args:
+ lst: the list to find the least common item in
+ count: the number of least common items to return
+
+ Returns:
+ The least common item in `lst`
+
+ .. warning::
+
+ In the case of ties, which least common item is returned
+ is undefined.
>>> least_common([1, 1, 1, 2, 2, 3, 3, 3, 4])
4
>>> least_common([1, 1, 1, 2, 2, 3, 3, 3, 4], count=2)
[4, 2]
-
"""
p = population_counts(lst)
mc = p.most_common()[-count:]
def dedup_list(lst: List[Any]) -> List[Any]:
"""
- Remove duplicates from the list performantly.
+ Remove duplicates from the list.
+
+ Args:
+ lst: the list to de-duplicate
+
+ Returns:
+ The de-duplicated input list. That is, the same list with
+ all extra duplicate items removed. The list composed of
+ the set of unique items from the input `lst`
>>> dedup_list([1, 2, 1, 3, 3, 4, 2, 3, 4, 5, 1])
[1, 2, 3, 4, 5]
-
"""
return list(set(lst))
def uniq(lst: List[Any]) -> List[Any]:
"""
- Alias for dedup_list.
+ Alias for :meth:`dedup_list`.
"""
return dedup_list(lst)
def contains_duplicates(lst: List[Any]) -> bool:
"""
- Does the list contian duplicate elements or not?
+ Does the list contain duplicate elements or not?
+
+ Args:
+ lst: the list to check for duplicates
+
+ Returns:
+ True if the input `lst` contains duplicated items and
+ False otherwise.
>>> lst = [1, 2, 1, 3, 3, 4, 4, 5, 6, 1, 3, 4]
>>> contains_duplicates(lst)
>>> contains_duplicates(dedup_list(lst))
False
-
"""
seen = set()
for _ in lst:
def all_unique(lst: List[Any]) -> bool:
"""
- Inverted alias for contains_duplicates.
+ Inverted alias for :meth:`contains_duplicates`.
"""
return not contains_duplicates(lst)
"""
Transpose a list of lists.
+ Args:
+ lst: the list of lists to be transposed.
+
+ Returns:
+ The transposed result. See example.
+
>>> lst = [[1, 2], [3, 4], [5, 6]]
>>> transpose(lst)
[[1, 3, 5], [2, 4, 6]]
return [list(_) for _ in transposed]
-def ngrams(lst: Sequence[Any], n):
+def ngrams(lst: Sequence[Any], n: int):
"""
Return the ngrams in the sequence.
+ Args:
+ lst: the list in which to find ngrams
+ n: the size of each ngram to return
+
+ Returns:
+ A generator that yields all ngrams of size `n` in `lst`.
+
>>> seq = 'encyclopedia'
>>> for _ in ngrams(seq, 3):
... _
def permute(seq: str):
"""
- Returns all permutations of a sequence; takes O(N!) time.
+ Returns all permutations of a sequence.
+
+ Args:
+ seq: the sequence to permute
+
+ Returns:
+ All permutations creatable by shuffling items in `seq`.
+
+ .. warning::
+
+ Takes O(N!) time, beware of large inputs.
>>> for x in permute('cat'):
... print(x)
atc
tca
tac
-
"""
yield from _permute(seq, "")
def _permute(seq: str, path: str):
+ """Internal helper to permute items recursively."""
seq_len = len(seq)
if seq_len == 0:
yield path
def shuffle(seq: MutableSequence[Any]) -> MutableSequence[Any]:
"""Shuffles a sequence into a random order.
+ Args:
+ seq: a sequence to shuffle
+
+ Returns:
+ The shuffled sequence.
+
>>> random.seed(22)
>>> shuffle([1, 2, 3, 4, 5])
[3, 4, 1, 5, 2]
>>> shuffle('example')
'empaelx'
-
"""
if isinstance(seq, str):
from pyutils import string_utils
def scramble(seq: MutableSequence[Any]) -> MutableSequence[Any]:
+ """An alias for :meth:`shuffle`."""
return shuffle(seq)
def binary_search(lst: Sequence[Any], target: Any) -> Tuple[bool, int]:
"""Performs a binary search on lst (which must already be sorted).
- Returns a Tuple composed of a bool which indicates whether the
- target was found and an int which indicates the index closest to
- target whether it was found or not.
+
+ Args:
+ lst: the (already sorted!) list in which to search
+ target: the item value to be found
+
+ Returns:
+ A Tuple composed of a bool which indicates whether the
+ target was found and an int which indicates the index closest to
+ target whether it was found or not.
>>> a = [1, 4, 5, 6, 7, 9, 10, 11]
>>> binary_search(a, 4)
def _binary_search(
lst: Sequence[Any], target: Any, low: int, high: int
) -> Tuple[bool, int]:
+ """Internal helper to perform a binary search recursively."""
if high >= low:
mid = (high + low) // 2
if lst[mid] == target:
return (False, low)
-def powerset(lst: Sequence[Any]) -> Iterator[Sequence[Any]]:
- """Returns the powerset of the items in the input sequence.
+def powerset(seq: Sequence[Any]) -> Iterator[Sequence[Any]]:
+ """Returns the powerset of the items in the input sequence. That is,
+ return the set containing every set constructable using items from
+ seq (including the empty set and the "full" set: `seq` itself).
+
+ Args:
+ seq: the sequence whose items will be used to construct the powerset.
+
+ Returns:
+ The powerset composed of all sets possible to create with items from `seq`.
+ See: https://en.wikipedia.org/wiki/Power_set.
>>> for x in powerset([1, 2, 3]):
... print(x)
(2, 3)
(1, 2, 3)
"""
- return chain.from_iterable(combinations(lst, r) for r in range(len(lst) + 1))
+ return chain.from_iterable(combinations(seq, r) for r in range(len(seq) + 1))
if __name__ == '__main__':
# © Copyright 2021-2022, Scott Gasch
-"""Utilities related to logging. To use it you must invoke
-:meth:`initialize_logging`. If you use the
-:meth:`bootstrap.initialize` decorator on your program's entry point,
-it will call this for you. See :meth:`python_modules.bootstrap.initialize`
-for more details. If you use this you get:
-
-* Ability to set logging level,
-* ability to define the logging format,
-* ability to tee all logging on stderr,
-* ability to tee all logging into a file,
-* ability to rotate said file as it grows,
-* ability to tee all logging into the system log (syslog) and
- define the facility and level used to do so,
-* easy automatic pid/tid stamp on logging for debugging threads,
-* ability to squelch repeated log messages,
-* ability to log probabilistically in code,
-* ability to only see log messages from a particular module or
- function,
-* ability to clear logging handlers added by earlier loaded modules.
-
-All of these are controlled via commandline arguments to your program,
-see the code below for details.
+"""
+This is a module that offers an opinionated take on how whole program
+logging should be initialized and controlled. It uses the standard
+Python :mod:`logging` but gives you control, via commandline config,
+to do things such as:
+
+ * Set the logging default level (debug, info, warning, error, critical)
+ of the whole program (see: :code:`--logging_level`)... and to override
+ the logging level for individual modules/functions based on their names
+ (see :code:`--lmodule`),
+ * define the logging message format (see :code:`--logging_format` and
+ :code:`--logging_date_format`) including easily adding a PID/TID
+ marker on all messages to help with multithreaded debugging
+ (:code:`--logging_debug_threads`) and force module names of code
+ that emits log messages to be included in the format
+ (:code:`--logging_debug_modules`),
+ * control the destination of logged messages:
+
+ - log to the console/stderr (:code:`--logging_console`) and/or
+ - log to a rotated file (:code:`--logging_filename`,
+ :code:`--logging_filename_maxsize` and :code:`--logging_filename_count`)
+ and/or
+ - log to the UNIX syslog (:code:`--logging_syslog` and
+ :code:`--logging_syslog_facility`)
+
+ * optionally squelch repeated messages (:code:`--logging_squelch_repeats`),
+ * optionally log probalistically (:code:`--logging_probabilistically`),
+ * capture printed messages into the info log (:code:`--logging_captures_prints`),
+ * and optionally clear unwanted logging handlers added by other imports
+ before this one (:code:`--logging_clear_preexisting_handlers`).
+
+To use this functionality, call :meth:`initialize_logging` early
+in your program entry point. If you use the
+:meth:`pyutils.bootstrap.initialize` decorator on your program's entry
+point, it will call this for you automatically.
"""
import collections
of code produces different messages (because of, e.g., a format
string), the messages are considered to be different.
+ An example of this from the pyutils code itself can be found in
+ :meth:`pyutils.ansi.fg` and :meth:`pyutils.ansi.bg` methods::
+
+ @logging_utils.squelch_repeated_log_messages(1)
+ def fg(
+ name: Optional[str] = "",
+ red: Optional[int] = None,
+ green: Optional[int] = None,
+ blue: Optional[int] = None,
+ *,
+ force_16color: bool = False,
+ force_216color: bool = False,
+ ) -> str:
+ ...
+
+ These methods log stuff like "Using 24-bit color strategy" which
+ gets old really fast and fills up the logs. By decorating the methods
+ with :code:`@logging_utils.squelch_repeated_log_messages(1)` the code
+ is requesting that its logged messages be dropped silently after the
+ first one is produced (note the argument 1).
+
+ Users can insist that all logged messages always be reflected in the
+ logs using the :code:`--no_logging_squelch_repeats` flag but the default
+ behavior is to allow code to request it be squelched.
+
+ :code:`--logging_squelch_repeats` only affects code with this decorator
+ on it; it ignores all other code.
+
+ Args:
+ squelch_after_n_repeats: the number of repeated messages allowed to
+ log before subsequent messages are silently dropped.
"""
def squelch_logging_wrapper(f: Callable):
This filter only affects logging messages that repeat more than a
threshold number of times from functions that are tagged with the
- @logging_utils.squelched_logging_ok decorator (see above); others
- are ignored.
+ :code:`@logging_utils.squelched_logging_ok` decorator (see above);
+ all others are ignored.
This functionality is enabled by default but can be disabled via
the :code:`--no_logging_squelch_repeats` commandline flag.
@overrides
def filter(self, record: logging.LogRecord) -> bool:
+ """Should we drop this log message?"""
id1 = f'{record.module}:{record.funcName}'
if id1 not in squelched_logging_counts:
return True
class DynamicPerScopeLoggingLevelFilter(logging.Filter):
"""This filter only allows logging messages from an allow list of
- module names or module:function names. Blocks all others.
+ module names or `module:function` names. Blocks all others. This
+ filter is used to implement the :code:`--lmodule` commandline option.
+
+ .. note::
+
+ You probably don't need to use this directly, just use
+ :code:`--lmodule`. For example, to set logging level to INFO
+ everywhere except "module:function" where it should be DEBUG::
+
+ # myprogram.py --logging_level=INFO --lmodule=module:function=DEBUG
+
"""
@staticmethod
def level_name_to_level(name: str) -> int:
+ """Given a level name, return its numberic value."""
numeric_level = getattr(logging, name, None)
if not isinstance(numeric_level, int):
raise ValueError(f'Invalid level: {name}')
def __init__(
self,
default_logging_level: int,
- per_scope_logging_levels: str,
+ per_scope_logging_levels: Optional[str],
) -> None:
+ """Construct the Filter.
+
+ Args:
+ default_logging_level: the logging level of the whole program
+ per_scope_logging_levels: optional, comma separated overrides of
+ logging level per scope of the format scope=level where
+ scope is of the form "module:function" or ":function" and
+ level is one of NOTSET, DEBUG, INFO, WARNING, ERROR or
+ CRITICAL.
+ """
super().__init__()
self.valid_levels = set(
['NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
def logging_is_probabilistic(probability_of_logging: float) -> Callable:
"""A decorator that indicates that all logging statements within the
- scope of a particular (marked) function are not deterministic
- (i.e. they do not always unconditionally log) but rather are
- probabilistic (i.e. they log N% of the time, randomly).
+ scope of a particular (marked via decorator) function are not
+ deterministic (i.e. they do not always unconditionally log) but rather
+ are probabilistic (i.e. they log N% of the time, randomly) when the
+ user passes the :code:`--logging_probabilistically` commandline flag
+ (which is enabled by default).
.. note::
+
This affects *ALL* logging statements within the marked function.
+ If you want it to only affect a subset of logging statements,
+ log those statements in a separate function that you invoke
+ from within the "too large" scope and mark that separate function
+ with the :code:`logging_is_probabilistic` decorator instead.
That this functionality can be disabled (forcing all logged
messages to produce output) via the
class ProbabilisticFilter(logging.Filter):
"""
A filter that logs messages probabilistically (i.e. randomly at some
- percent chance).
+ percent chance). This filter is used with a decorator (see
+ :meth:`logging_is_probabilistic`) to implement the
+ :code:`--logging_probabilistically` commandline flag.
This filter only affects logging messages from functions that have
- been tagged with the @logging_utils.probabilistic_logging decorator.
+ been tagged with the `@logging_utils.probabilistic_logging` decorator.
"""
@overrides
def filter(self, record: logging.LogRecord) -> bool:
+ """Should the message be logged?"""
id1 = f'{record.module}:{record.funcName}'
if id1 not in probabilistic_logging_levels:
return True
class OnlyInfoFilter(logging.Filter):
"""A filter that only logs messages produced at the INFO logging
- level. This is used by the ::code`--logging_info_is_print`
+ level. This is used by the :code:`--logging_info_is_print`
commandline option to select a subset of the logging stream to
send to a stdout handler.
"""
class MillisecondAwareFormatter(logging.Formatter):
"""
A formatter for adding milliseconds to log messages which, for
- whatever reason, the default python logger doesn't do.
+ whatever reason, the default Python logger doesn't do.
+
+ .. note::
+
+ You probably don't need to use this directly but it is
+ wired in under :meth:initialize_logging so that the
+ timestamps in log messages have millisecond level
+ precision.
"""
converter = datetime.datetime.fromtimestamp # type: ignore
return s
-def log_about_logging(
+def _log_about_logging(
logger,
default_logging_level,
preexisting_handlers_count,
fmt,
facility_name,
):
- """Some of the initial messages in the debug log are about how we
- have set up logging itself."""
-
+ """This is invoked automatically after logging is initialized such
+ that the first messages in the log are about how logging itself
+ was configured.
+ """
level_name = logging._levelToName.get(
default_logging_level, str(default_logging_level)
)
def initialize_logging(logger=None) -> logging.Logger:
- """Initialize logging for the program. This must be called if you want
- to use any of the functionality provided by this module such as:
-
- * Ability to set logging level,
- * ability to define the logging format,
- * ability to tee all logging on stderr,
- * ability to tee all logging into a file,
- * ability to rotate said file as it grows,
- * ability to tee all logging into the system log (syslog) and
- define the facility and level used to do so,
- * easy automatic pid/tid stamp on logging for debugging threads,
- * ability to squelch repeated log messages,
- * ability to log probabilistically in code,
- * ability to only see log messages from a particular module or
- function,
- * ability to clear logging handlers added by earlier loaded modules.
-
- All of these are controlled via commandline arguments to your program,
- see the code below for details.
+ """Initialize logging for the program. See module level comments
+ for information about what functionality this provides and how to
+ enable or disable functionality via the commandline.
If you use the
:meth:`bootstrap.initialize` decorator on your program's entry point,
- it will call this for you. See :meth:`python_modules.bootstrap.initialize`
+ it will call this for you. See :meth:`pyutils.bootstrap.initialize`
for more details.
"""
global LOGGING_INITIALIZED
# At this point the logger is ready, handlers are set up,
# etc... so log about the logging configuration.
- log_about_logging(
+ _log_about_logging(
logger,
default_logging_level,
preexisting_handlers_count,
pass
-def dprint(*args, **kwargs) -> None:
- """Legacy function used to print to stderr still needed by some code.
- Please just use normal logging with --logging_console which
- accomplishes the same thing in new code.
- """
- print(*args, file=sys.stderr, **kwargs)
-
-
class OutputMultiplexer(object):
"""A class that broadcasts printed messages to several sinks
(including various logging levels, different files, different file
def print(self, *args, **kwargs):
"""Produce some output to all sinks."""
- from pyutils.string_utils import sprintf, strip_escape_sequences
+ from pyutils.string_utils import _sprintf, strip_escape_sequences
end = kwargs.pop("end", None)
if end is not None:
raise TypeError("sep must be None or a string")
if kwargs:
raise TypeError("invalid keyword arguments to print()")
- buf = sprintf(*args, end="", sep=sep)
+ buf = _sprintf(*args, end="", sep=sep)
if sep is None:
sep = " "
if end is None:
filenames=None,
handles=None,
):
+ """
+ Args:
+ destination_bitv: a bitvector that indicates where we should
+ send output. See :class:`OutputMultiplexer` for options.
+ logger: optional logger to use for log destination messages.
+ filenames: optional filenames to write for filename destination
+ messages.
+ handles: optional open filehandles to write for filehandle
+ destination messages.
+ """
super().__init__(
destination_bitv,
logger=logger,
def hlog(message: str) -> None:
"""Write a message to the house log (syslog facility local7 priority
- info) by calling /usr/bin/logger. This is pretty hacky but used
- by a bunch of code. Another way to do this would be to use
+ info) by calling `/usr/bin/logger`. This is pretty hacky but used
+ by a bunch of (my) code. Another way to do this would be to use
:code:`--logging_syslog` and :code:`--logging_syslog_facility` but
I can't actually say that's easier.
+
+ TODO: this needs to move.
"""
message = message.replace("'", "'\"'\"'")
os.system(f"/usr/bin/logger -p local7.info -- '{message}'")
# © Copyright 2021-2022, Scott Gasch
-"""Mathematical helpers."""
+"""Helper utilities with a mathematical / statictical focus."""
import collections
import functools
class NumericPopulation(object):
- """A numeric population with some statistics such as median, mean, pN,
- stdev, etc...
+ """This object *store* a numerical population in a way that enables relatively
+ fast addition of new numbers (:math:`O(2log_2 n)`) and instant access to the
+ median value in the population (:math:`O(1)`). It also provides other population
+ summary statistics such as the :meth:`mode`, :meth:`get_percentile` and
+ :meth:`stdev`.
+
+ .. note::
+
+ Because this class stores a copy of all numbers added to it, it shouldn't
+ be used for very large populations. Consider sampling.
>>> pop = NumericPopulation()
>>> pop.add_number(1)
def add_number(self, number: float):
"""Adds a number to the population. Runtime complexity of this
- operation is :math:`O(2 log_2 n)`"""
+ operation is :math:`O(2 log_2 n)`
+
+ Args:
+ number: the number to add_number to the population
+ """
if not self.highers or number > self.highers[0]:
heappush(self.highers, number)
self.minimum = number
def __len__(self):
- """Return the population size."""
+ """
+ Returns:
+ the population's current size.
+ """
n = 0
if self.highers:
n += len(self.highers)
return n
def _rebalance(self):
+ """Internal helper for rebalancing the `lowers` and `highers` heaps"""
if len(self.lowers) - len(self.highers) > 1:
heappush(self.highers, -heappop(self.lowers))
elif len(self.highers) - len(self.lowers) > 1:
heappush(self.lowers, -heappop(self.highers))
def get_median(self) -> float:
- """Returns the approximate median (p50) so far in :math:`O(1)` time."""
-
+ """
+ Returns:
+ The median (p50) of the current population in :math:`O(1)` time.
+ """
if len(self.lowers) == len(self.highers):
return -self.lowers[0]
elif len(self.lowers) > len(self.highers):
return self.highers[0]
def get_mean(self) -> float:
- """Returns the mean (arithmetic mean) so far in :math:`O(1)` time."""
-
+ """
+ Returns:
+ The mean (arithmetic mean) so far in :math:`O(1)` time.
+ """
count = len(self)
return self.aggregate / count
def get_mode(self) -> Tuple[float, int]:
- """Returns the mode (most common member in the population)
- in :math:`O(n)` time."""
-
+ """
+ Returns:
+ The population mode (most common member in the population)
+ in :math:`O(n)` time.
+ """
count: Dict[float, int] = collections.defaultdict(int)
for n in self.lowers:
count[-n] += 1
return dict_utils.item_with_max_value(count)
def get_stdev(self) -> float:
- """Returns the stdev so far in :math:`O(n)` time."""
-
+ """
+ Returns:
+ The stdev of the current population in :math:`O(n)` time.
+ """
mean = self.get_mean()
variance = 0.0
for n in self.lowers:
return math.sqrt(variance) / count
def _create_sorted_copy_if_needed(self, count: int):
+ """Internal helper."""
if not self.sorted_copy or count != len(self.sorted_copy):
self.sorted_copy = []
for x in self.lowers:
self.sorted_copy = sorted(self.sorted_copy)
def get_percentile(self, n: float) -> float:
- """Returns the number at approximately pn% (i.e. the nth percentile)
- of the distribution in :math:`O(n log_2 n)` time. Not thread-safe;
- does caching across multiple calls without an invocation to
- add_number for perf reasons.
+ """
+ Returns: the number at approximately pn% in the population
+ (i.e. the nth percentile) in :math:`O(n log_2 n)` time (it
+ performs a full sort). This is not the most efficient
+ algorithm.
+
+ Not thread-safe; does caching across multiple calls without
+ an invocation to :meth:`add_number` for perf reasons.
+
+ Args:
+ n: the percentile to compute
"""
if n == 50:
return self.get_median()
def gcd_floats(a: float, b: float) -> float:
- """Returns the greatest common divisor of a and b."""
+ """
+ Returns:
+ The greatest common divisor of a and b.
+
+ Args:
+ a: first operand
+ b: second operatnd
+ """
if a < b:
return gcd_floats(b, a)
def gcd_float_sequence(lst: List[float]) -> float:
- """Returns the greatest common divisor of a list of floats."""
+ """
+ Returns:
+ The greatest common divisor of a list of floats.
+
+ Args:
+ lst: a list of operands
+ """
if len(lst) <= 0:
raise ValueError("Need at least one number")
elif len(lst) == 1:
def truncate_float(n: float, decimals: int = 2):
- """Truncate a float to a particular number of decimals.
+ """
+ Returns:
+ A truncated float to a particular number of decimals.
+
+ Args:
+ n: the float to truncate
>>> truncate_float(3.1415927, 3)
3.141
-
"""
assert 0 < decimals < 10
multiplier = 10**decimals
def percentage_to_multiplier(percent: float) -> float:
- """Given a percentage (e.g. 155%), return a factor needed to scale a
- number by that percentage.
+ """Given a percentage that represents a return or percent change
+ (e.g. 155%), determine the factor (i.e. multiplier) needed to
+ scale a number by that percentage (e.g. 2.55x)
+
+ Args:
+ percent: the return percent to scale by
>>> percentage_to_multiplier(155)
2.55
1.45
>>> percentage_to_multiplier(-25)
0.75
+
"""
multiplier = percent / 100
multiplier += 1.0
def multiplier_to_percent(multiplier: float) -> float:
- """Convert a multiplicative factor into a percent change.
+ """Convert a multiplicative factor into a percent change or return
+ percentage.
+
+ Args:
+ multiplier: the multiplier for which to compute the percent change
>>> multiplier_to_percent(0.75)
-25.0
@functools.lru_cache(maxsize=1024, typed=True)
def is_prime(n: int) -> bool:
"""
- Returns True if n is prime and False otherwise. Obviously(?) very slow for
- very large input numbers.
+ Args:
+ n: the number for which primeness is to be determined.
+
+ Returns:
+ True if n is prime and False otherwise.
+
+ .. note::
+
+ Obviously(?) very slow for very large input numbers until
+ we get quantum computers.
>>> is_prime(13)
True
def is_running_as_root() -> bool:
- """Returns True if running as root.
+ """
+ Returns:
+ True if running as root, False otherwise.
>>> is_running_as_root()
False
def debugger_is_attached() -> bool:
- """Return if the debugger is attached"""
-
+ """
+ Returns:
+ True if a debugger is attached, False otherwise.
+ """
gettrace = getattr(sys, 'gettrace', lambda: None)
return gettrace() is not None
# © Copyright 2021-2022, Scott Gasch
-"""A :class:`Persistent` is just a class with a load and save method. This
-module defines the :class:`Persistent` base and a decorator that can be used to
-create a persistent singleton that autoloads and autosaves."""
+"""
+This module defines a class hierarchy (base class :class:`Persistent`) and
+a decorator (`@persistent_autoloaded_singleton`) that can be used to create
+objects that load and save their state from some external storage location
+automatically, optionally and conditionally.
+
+A :class:`Persistent` is just a class with a :meth:`Persistent.load` and
+:meth:`Persistent.save` method. Various subclasses such as
+:class:`JsonFileBasedPersistent` and :class:`PicklingFileBasedPersistent`
+define these methods to, save data in a particular format. The details
+of where and whether to save are left to your code to decide by implementing
+interface methods like :meth:`Persistent.get_filename` and
+:meth:`Persistent.should_we_load_data`.
+
+This module inculdes some helpers to make deciding whether to load persisted
+state easier such as :meth:`was_file_written_today` and
+:meth:`was_file_written_within_n_seconds`.
+
+:class:`Persistent` classes are good for things backed by persisted
+state that is loaded all or most of the time. For example, the high
+score list of a game, the configuration settings of a tool,
+etc... Really anything that wants to save/load state from storage and
+not bother with the plumbing to do so.
+"""
import atexit
import datetime
class FileBasedPersistent(Persistent):
- """A Persistent that uses a file to save/load data and knows the conditions
- under which the state should be saved/loaded."""
+ """A :class:`Persistent` subclass that uses a file to save/load
+ data and knows the conditions under which the state should be
+ saved/loaded.
+ """
@staticmethod
@abstractmethod
def get_filename() -> str:
- """Since this class saves/loads to/from a file, what's its full path?"""
+ """
+ Returns:
+ The full path of the file in which we are saving/loading data.
+ """
pass
@staticmethod
@abstractmethod
def should_we_save_data(filename: str) -> bool:
+ """
+ Returns:
+ True if we should save our state now or False otherwise.
+ """
pass
@staticmethod
@abstractmethod
def should_we_load_data(filename: str) -> bool:
+ """
+ Returns:
+ True if we should load persisted state now or False otherwise.
+ """
pass
@abstractmethod
def get_persistent_data(self) -> Any:
+ """
+ Returns:
+ The raw state data read from the filesystem. Can be any format.
+ """
pass
class PicklingFileBasedPersistent(FileBasedPersistent):
"""
- A class that stores its state in a file as pickled python objects.
+ A class that stores its state in a file as pickled Python objects.
Example usage::
@persistent.persistent_autoloaded_singleton()
class MyClass(persistent.PicklingFileBasedPersistent):
- def __init__(self, data: Whatever):
- #initialize youself from data
+ def __init__(self, data: Optional[Whatever]):
+ if data:
+ # initialize state from data
+ else:
+ # if desired, initialize an "empty" object with new state.
@staticmethod
@overrides
return persistent.was_file_written_within_n_seconds(whatever)
# Persistent will handle the plumbing to instantiate your class from its
- # persisted state iff the should_we_load_data says it's ok to. It will
- # also persist the current in memory state to disk at program exit iff
- # the should_we_save_data methods says to.
+ # persisted state iff the :meth:`should_we_load_data` says it's ok to. It
+ # will also persist the current in-memory state to disk at program exit iff
+ # the :meth:`should_we_save_data` methods says to.
c = MyClass()
"""
class JsonFileBasedPersistent(FileBasedPersistent):
- """
- A class that stores its state in a JSON format file.
+ """A class that stores its state in a JSON format file.
Example usage::
@persistent.persistent_autoloaded_singleton()
class MyClass(persistent.JsonFileBasedPersistent):
- def __init__(self, data: Whatever):
- #initialize youself from data
+ def __init__(self, data: Optional[dict[str, Any]]):
+ # load already deserialized the JSON data for you; it's
+ # a "cooked" JSON dict of string -> values, lists, dicts,
+ # etc...
+ if data:
+ #initialize youself from data...
+ else:
+ # if desired, initialize an empty state object
+ # when json_data isn't provided.
@staticmethod
@overrides
def should_we_load_data(filename: str) -> bool:
return persistent.was_file_written_within_n_seconds(whatever)
- # Persistent will handle the plumbing to instantiate your class from its
- # persisted state iff the should_we_load_data says it's ok to. It will
- # also persist the current in memory state to disk at program exit iff
- # the should_we_save_data methods says to.
+ # Persistent will handle the plumbing to instantiate your
+ # class from its persisted state iff the
+ # :meth:`should_we_load_data` says it's ok to. It will also
+ # persist the current in memory state to disk at program exit
+ # iff the :meth:`should_we_save_data methods` says to.
c = MyClass()
-
"""
@classmethod
for line in lines:
line = re.sub(r'#.*$', '', line)
buf += line
-
json_dict = json.loads(buf)
return cls(json_dict)
"""Convenience wrapper around :meth:`was_file_written_within_n_seconds`.
Args:
- filename: filename to check
+ filename: path / filename to check
Returns:
True if filename was written today.
>>> was_file_written_today(filename)
False
"""
-
if not file_utils.does_file_exist(filename):
return False
chance to read state from somewhere persistent (disk, db,
whatever). Subsequent calls to construt instances of the wrapped
class will return a single, global instance (i.e. the wrapped
- class is a singleton).
+ class is must be a singleton).
If :meth:`load` fails (returns None), the c'tor is invoked with the
original args as a fallback.
# © Copyright 2021-2022, Scott Gasch
-"""A simple utility to unpickle some code, run it, and pickle the
-results. Please don't unpickle (or run!) code you do not know.
-
-This script is used by code in parallelize, namely the
-:class:`RemoteExecutor`, to schedule work on a remote machine.
-The code in :file:`parallelize.py` uses a user-defined configuration
-to schedule work this way. See that file for setup instructions.
+"""A simple utility to unpickle some code from the filesystem, run it,
+pickle the results, and save them back on the filesystem. This file
+helps :mod:`pyutils.parallelize.parallelize` and
+:mod:`pyutils.parallelize.executors` implement the
+:class:`pyutils.parallelize.executors.RemoteExecutor` that distributes
+work to different machines when code is marked with the
+`@parallelize(method=Method.REMOTE)` decorator.
+
+.. warning::
+ Please don't unpickle (or run!) code you do not know! This
+ helper is designed to be run with your own code.
+
+See details in :mod:`pyutils.parallelize.parallelize` for instructions
+about how to set this up.
"""
import logging
@background_thread
-def watch_for_cancel(terminate_event: threading.Event) -> None:
+def _watch_for_cancel(terminate_event: threading.Event) -> None:
logger.debug('Starting up background thread...')
p = psutil.Process(os.getpid())
while True:
time.sleep(1.0)
-def cleanup_and_exit(
+def _cleanup_and_exit(
thread: Optional[threading.Thread],
- stop_thread: Optional[threading.Event],
+ stop_event: Optional[threading.Event],
exit_code: int,
) -> None:
- if stop_thread is not None:
- stop_thread.set()
+ if stop_event is not None:
+ stop_event.set()
assert thread is not None
thread.join()
sys.exit(exit_code)
@bootstrap.initialize
def main() -> None:
+ """Remote worker entry point."""
+
in_file = config.config['code_file']
assert in_file and type(in_file) == str
out_file = config.config['result_file']
assert out_file and type(out_file) == str
thread = None
- stop_thread = None
+ stop_event = None
if config.config['watch_for_cancel']:
- (thread, stop_thread) = watch_for_cancel()
+ thread, stop_event = _watch_for_cancel()
logger.debug('Reading %s.', in_file)
try:
except Exception as e:
logger.exception(e)
logger.critical('Problem reading %s. Aborting.', in_file)
- cleanup_and_exit(thread, stop_thread, 1)
+ _cleanup_and_exit(thread, stop_event, 1)
logger.debug('Deserializing %s', in_file)
try:
except Exception as e:
logger.exception(e)
logger.critical('Problem deserializing %s. Aborting.', in_file)
- cleanup_and_exit(thread, stop_thread, 2)
+ _cleanup_and_exit(thread, stop_event, 2)
- logger.debug('Invoking user code...')
+ logger.debug('Invoking user-defined code...')
with Timer() as t:
ret = fun(*args, **kwargs)
logger.debug('User code took %.1fs', t())
except Exception as e:
logger.exception(e)
logger.critical('Could not serialize result (%s). Aborting.', type(ret))
- cleanup_and_exit(thread, stop_thread, 3)
+ _cleanup_and_exit(thread, stop_event, 3)
logger.debug('Writing %s', out_file)
try:
except Exception as e:
logger.exception(e)
logger.critical('Error writing %s. Aborting.', out_file)
- cleanup_and_exit(thread, stop_thread, 4)
- cleanup_and_exit(thread, stop_thread, 0)
+ _cleanup_and_exit(thread, stop_event, 4)
+ _cleanup_and_exit(thread, stop_event, 0)
if __name__ == '__main__':
# © Copyright 2021-2022, Scott Gasch
-"""Several helpers to keep track of internal state via periodic
-polling. :class:`StateTracker` expects to be invoked periodically to
-maintain state whereas the others (:class:`AutomaticStateTracker` and
-:class:`WaitableAutomaticStateTracker`) automatically update themselves
-and, optionally, expose an event for client code to wait on state
+"""This module defines several classes (:py:class:`StateTracker`,
+:py:class:`AutomaticStateTracker`, and
+:py:class:`WaitableAutomaticStateTracker`) that can be used as base
+classes by your code. These class patterns are meant to encapsulate
+and represent some state that dynamically changes and must be updated
+periodically. These classes update their state (either automatically
+or when invoked to poll) and allow their callers to wait on state
changes.
"""
class StateTracker(ABC):
- """A base class that maintains and updates a global state via an
- update routine. Instances of this class should be periodically
- invoked via the heartbeat() method. This method, in turn, invokes
- update() with update_ids according to a schedule / periodicity
- provided to the c'tor.
+ """A base class that maintains and updates its state via an update
+ routine called :meth:`heartbeat`. This method is not automatic:
+ instances of this class should be periodically invoked via their
+ :meth:`heartbeat` method by some other thread.
+
+ See also :class:`AutomaticStateTracker` if you'd rather not have
+ to invoke your code regularly.
"""
def __init__(self, update_ids_to_update_secs: Dict[str, float]) -> None:
- """The update_ids_to_update_secs dict parameter describes one or more
- update types (unique update_ids) and the periodicity(ies), in
- seconds, at which it/they should be invoked.
+ """The update_ids_to_update_secs dict parameter describes one
+ or more update types (unique update_ids) and the
+ periodicity(ies), in seconds, at which it/they should be
+ invoked.
.. note::
When more than one update is overdue, they will be
This would indicate that every 10s we would like to
refresh local state whereas every 60s we'd like to
refresh remote state.
+
"""
self.update_ids_to_update_secs = update_ids_to_update_secs
self.last_reminder_ts: Dict[str, Optional[datetime.datetime]] = {}
Args:
update_id: the string you passed to the c'tor as a key in
the update_ids_to_update_secs dict. :meth:`update` will
- only be invoked on the shoulder, at most, every update_secs
+ only be invoked, at most, every update_secs
seconds.
now: the approximate current timestamp at invocation time.
pass
def heartbeat(self, *, force_all_updates_to_run: bool = False) -> None:
- """Invoke this method to cause the StateTracker instance to identify
- and invoke any overdue updates based on the schedule passed to
- the c'tor. In the base :class:`StateTracker` class, this method must
- be invoked manually by a thread from external code. Other subclasses
- are available that create their own updater threads (see below).
-
- If more than one type of update (update_id) are overdue,
- they will be invoked in order based on their update_ids.
-
- Setting force_all_updates_to_run will invoke all updates
- (ordered by update_id) immediately ignoring whether or not
+ """Invoke this method periodically to cause the :class:`StateTracker`
+ instance to identify and invoke any overdue updates based on the
+ schedule passed to the c'tor. In the base :class:`StateTracker` class,
+ this method must be invoked manually by a thread from external code.
+ Other subclasses (e.g. :class:`AutomaticStateTracker`) are available
+ that create their own updater threads (see below).
+
+ If more than one type of update (`update_id`) is overdue,
+ overdue updates will be invoked in order based on their `update_id`.
+
+ Setting `force_all_updates_to_run` will invoke all updates
+ (ordered by `update_id`) immediately ignoring whether or not
they are due.
"""
"""
@background_thread
- def pace_maker(self, should_terminate: threading.Event) -> None:
+ def _pace_maker(self, should_terminate: threading.Event) -> None:
"""Entry point for a background thread to own calling :meth:`heartbeat`
at regular intervals so that the main thread doesn't need to
do so.
+
+ Args:
+ should_terminate: an event which, when set, indicates we should terminate.
"""
while True:
if should_terminate.is_set():
- logger.debug('pace_maker noticed event; shutting down')
+ logger.debug('_pace_maker noticed event; shutting down')
return
self.heartbeat()
- logger.debug('pace_maker is sleeping for %.1fs', self.sleep_delay)
+ logger.debug('_pace_maker is sleeping for %.1fs', self.sleep_delay)
time.sleep(self.sleep_delay)
def __init__(
periods_list = list(update_ids_to_update_secs.values())
self.sleep_delay = math_utils.gcd_float_sequence(periods_list)
logger.info('Computed sleep_delay=%.1f', self.sleep_delay)
- (thread, stop_event) = self.pace_maker()
+ (thread, stop_event) = self._pace_maker()
self.should_terminate = stop_event
self.updater_thread = thread
def shutdown(self):
"""Terminates the background thread and waits for it to tear down.
- This may block for as long as self.sleep_delay.
+ This may block for as long as `self.sleep_delay`.
"""
logger.debug('Setting shutdown event and waiting for background thread.')
self.should_terminate.set()
detector = waitable_presence.WaitableAutomaticStateSubclass()
while True:
- changed = detector.wait(timeout=60 * 5)
+ changed = detector.wait(timeout=60)
if changed:
detector.reset()
- # Figure out what changed and react
+ # Figure out what changed and react somehow
else:
# Just a timeout; no need to reset. Maybe do something
# else before looping up into wait again.
self._something_changed.clear()
def wait(self, *, timeout=None):
- """Wait for something to change or a timeout to lapse.
+ """Blocking wait for something to change or a timeout to lapse.
Args:
timeout: maximum amount of time to wait. If None, wait
- forever (until something changes).
+ forever (until something changes or shutdown).
"""
return self._something_changed.wait(timeout=timeout)
"""
A stopwatch to time how long something takes (walltime).
- e.g.
+ Example usage::
with stopwatch.Timer() as t:
do_the_thing()
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
-This class is based on: https://github.com/daveoncode/python-string-utils.
-See NOTICE in the root of this module for a detailed enumeration of what
-work is Davide's and what work was added by Scott.
+This class is based on:
+https://github.com/daveoncode/python-string-utils. See `NOTICE
+<[https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD>`_
+in the root of this module for a detailed enumeration of what work is
+Davide's and what work was added by Scott.
+
"""
import base64
"ninety",
]
-scales = ["hundred", "thousand", "million", "billion", "trillion", "quadrillion"]
+MAGNITUDE_SCALES = [
+ "hundred",
+ "thousand",
+ "million",
+ "billion",
+ "trillion",
+ "quadrillion",
+]
NUM_WORDS = {}
NUM_WORDS["and"] = (1, 0)
NUM_WORDS[word] = (1, i)
for i, word in enumerate(TENS_WORDS):
NUM_WORDS[word] = (1, i * 10)
-for i, word in enumerate(scales):
+for i, word in enumerate(MAGNITUDE_SCALES):
if i == 0:
NUM_WORDS[word] = (100, 0)
else:
True if the input string is either None or an empty string,
False otherwise.
+ See also :meth:`is_string` and :meth:`is_empty_string`.
+
>>> is_none_or_empty("")
True
>>> is_none_or_empty(None)
Returns:
True if the object is a string and False otherwise.
+ See also :meth:`is_empty_string`, :meth:`is_none_or_empty`.
+
>>> is_string('test')
True
>>> is_string(123)
Returns:
True if the string is empty and False otherwise.
+
+ See also :meth:`is_none_or_empty`, :meth:`is_full_string`.
"""
return is_empty(in_str)
Returns:
True if the string is empty and false otherwise.
+ See also :meth:`is_none_or_empty`, :meth:`is_full_string`.
+
>>> is_empty('')
True
>>> is_empty(' \t\t ')
True if the object is a string and is not empty ('') and
is not only composed of whitespace.
+ See also :meth:`is_string`, :meth:`is_empty_string`, :meth:`is_none_or_empty`.
+
>>> is_full_string('test!')
True
>>> is_full_string('')
True if the string contains a valid numberic value and
False otherwise.
+ See also :meth:`is_integer_number`, :meth:`is_decimal_number`,
+ :meth:`is_hexidecimal_integer_number`, :meth:`is_octal_integer_number`,
+ etc...
+
>>> is_number(100.5)
Traceback (most recent call last):
...
decimal, hex, or octal, regular or scientific) integral
expression and False otherwise.
+ See also :meth:`is_number`, :meth:`is_decimal_number`,
+ :meth:`is_hexidecimal_integer_number`, :meth:`is_octal_integer_number`,
+ etc...
+
>>> is_integer_number('42')
True
>>> is_integer_number('42.0')
Returns:
True if the string is a hex integer number and False otherwise.
+ See also :meth:`is_integer_number`, :meth:`is_decimal_number`,
+ :meth:`is_octal_integer_number`, :meth:`is_binary_integer_number`, etc...
+
>>> is_hexidecimal_integer_number('0x12345')
True
>>> is_hexidecimal_integer_number('0x1A3E')
Returns:
True if the string is a valid octal integral number and False otherwise.
+ See also :meth:`is_integer_number`, :meth:`is_decimal_number`,
+ :meth:`is_hexidecimal_integer_number`, :meth:`is_binary_integer_number`,
+ etc...
+
>>> is_octal_integer_number('0o777')
True
>>> is_octal_integer_number('-0O115')
Returns:
True if the string contains a binary integral number and False otherwise.
+ See also :meth:`is_integer_number`, :meth:`is_decimal_number`,
+ :meth:`is_hexidecimal_integer_number`, :meth:`is_octal_integer_number`,
+ etc...
+
>>> is_binary_integer_number('0b10111')
True
>>> is_binary_integer_number('-0b111')
Returns:
The integral value of the string or raises on error.
+ See also :meth:`is_integer_number`, :meth:`is_decimal_number`,
+ :meth:`is_hexidecimal_integer_number`, :meth:`is_octal_integer_number`,
+ :meth:`is_binary_integer_number`, etc...
+
>>> to_int('1234')
1234
+ >>> to_int('0x1234')
+ 4660
+ >>> to_int('0b01101')
+ 13
+ >>> to_int('0o777')
+ 511
>>> to_int('test')
Traceback (most recent call last):
...
def number_string_to_integer(in_str: str) -> int:
"""Convert a string containing a written-out number into an int.
+ Args:
+ in_str: the string containing the long-hand written out integer number
+ in English. See examples below.
+
+ Returns:
+ The integer whose value was parsed from in_str.
+
+ See also :meth:`integer_to_number_string`.
+
+ .. warning::
+ This code only handles integers; it will not work with decimals / floats.
+
>>> number_string_to_integer("one hundred fifty two")
152
def integer_to_number_string(num: int) -> str:
"""
- Opposite of number_string_to_integer; convert a number to a written out
- longhand format.
+ Opposite of :meth:`number_string_to_integer`; converts a number to a written out
+ longhand format in English.
+
+ Args:
+ num: the integer number to convert
+
+ Returns:
+ The long-hand written out English form of the number. See examples below.
+
+ See also :meth:`number_string_to_integer`.
+
+ .. warning::
+ This method does not handle decimals or floats, only ints.
>>> integer_to_number_string(9)
'nine'
>>> integer_to_number_string(123219982)
'one hundred twenty three million two hundred nineteen thousand nine hundred eighty two'
-
"""
if num < 20:
otherwise. A decimal may be signed or unsigned or use
a "scientific notation".
+ See also :meth:`is_integer_number`.
+
.. note::
We do not consider integers without a decimal point
to be decimals; they return False (see example).
Returns:
in_str with escape sequences removed.
+ See also: :mod:`pyutils.ansi`.
+
.. note::
What is considered to be an "escape sequence" is defined
by a regular expression. While this gets common ones,
def _add_thousands_separator(in_str: str, *, separator_char=',', places=3) -> str:
+ """Internal helper"""
decimal_part = ""
if '.' in in_str:
(in_str, decimal_part) = in_str.split('.')
Returns:
An integer number of bytes or None to indicate an error.
+ See also :meth:`number_to_suffix_string`.
+
>>> suffix_string_to_number('1Mb')
1048576
>>> suffix_string_to_number('13.1Gb')
A string with a suffix representing num bytes concisely or
None to indicate an error.
+ See also: :meth:`suffix_string_to_number`.
+
>>> number_to_suffix_string(14066017894)
'13.1Gb'
>>> number_to_suffix_string(1024 * 1024)
Returns:
True if in_str is a valid credit card number.
+
+ .. warning::
+ This code is not verifying the authenticity of the credit card (i.e.
+ not checking whether it's a real card that can be charged); rather
+ it's only checking that the number follows the "rules" for numbering
+ established by credit card issuers.
+
"""
if not is_full_string(in_str):
return False
* it's composed only by letters ([a-zA-Z]) and optionally numbers ([0-9])
* it contains both lowercase and uppercase letters
* it does not start with a number
+
+ See also :meth:`is_snake_case`, :meth:`is_slug`, and :meth:`camel_case_to_snake_case`.
"""
return is_full_string(in_str) and CAMEL_CASE_TEST_RE.match(in_str) is not None
* it contains at least one underscore (or provided separator)
* it does not start with a number
+ See also :meth:`is_camel_case`, :meth:`is_slug`, and :meth:`snake_case_to_camel_case`.
+
>>> is_snake_case('this_is_a_test')
True
>>> is_snake_case('___This_Is_A_Test_1_2_3___')
Returns:
True if the in_str contains a valid UUID and False otherwise.
+ See also :meth:`generate_uuid`.
+
>>> is_uuid('6f8aa2f9-686c-4ac3-8766-5712354a04cf')
True
>>> is_uuid('6f8aa2f9686c4ac387665712354a04cf')
Returns:
True if in_str contains a valid IPv4 address and False otherwise.
+ See also :meth:`extract_ip_v4`, :meth:`is_ip_v6`, :meth:`extract_ip_v6`,
+ and :meth:`is_ip`.
+
>>> is_ip_v4('255.200.100.75')
True
>>> is_ip_v4('nope')
The first extracted IPv4 address from in_str or None if
none were found or an error occurred.
+ See also :meth:`is_ip_v4`, :meth:`is_ip_v6`, :meth:`extract_ip_v6`,
+ and :meth:`is_ip`.
+
>>> extract_ip_v4(' The secret IP address: 127.0.0.1 (use it wisely) ')
'127.0.0.1'
>>> extract_ip_v4('Your mom dresses you funny.')
Returns:
True if in_str contains a valid IPv6 address and False otherwise.
+ See also :meth:`is_ip_v4`, :meth:`extract_ip_v4`, :meth:`extract_ip_v6`,
+ and :meth:`is_ip`.
+
>>> is_ip_v6('2001:db8:85a3:0000:0000:8a2e:370:7334')
True
>>> is_ip_v6('2001:db8:85a3:0000:0000:8a2e:370:?') # invalid "?"
The first IPv6 address found in in_str or None if no address
was found or an error occurred.
+ See also :meth:`is_ip_v4`, :meth:`is_ip_v6`, :meth:`extract_ip_v4`,
+ and :meth:`is_ip`.
+
>>> extract_ip_v6('IP: 2001:db8:85a3:0000:0000:8a2e:370:7334')
'2001:db8:85a3:0000:0000:8a2e:370:7334'
>>> extract_ip_v6("(and she's ugly too, btw)")
True if in_str contains a valid IP address (either IPv4 or
IPv6).
+ See also :meth:`is_ip_v4`, :meth:`is_ip_v6`, :meth:`extract_ip_v6`,
+ and :meth:`extract_ip_v4`.
+
>>> is_ip('255.200.100.75')
True
>>> is_ip('2001:db8:85a3:0000:0000:8a2e:370:7334')
The first IP address (IPv4 or IPv6) found in in_str or
None to indicate none found or an error condition.
+ See also :meth:`is_ip_v4`, :meth:`is_ip_v6`, :meth:`extract_ip_v6`,
+ and :meth:`extract_ip_v4`.
+
>>> extract_ip('Attacker: 255.200.100.75')
'255.200.100.75'
>>> extract_ip('Remote host: 2001:db8:85a3:0000:0000:8a2e:370:7334')
Returns:
True if in_str is a valid MAC address False otherwise.
+ See also :meth:`extract_mac_address`, :meth:`is_ip`, etc...
+
>>> is_mac_address("34:29:8F:12:0D:2F")
True
>>> is_mac_address('34:29:8f:12:0d:2f')
The first MAC address found in in_str or None to indicate no
match or an error.
+ See also :meth:`is_mac_address`, :meth:`is_ip`, and :meth:`extract_ip`.
+
>>> extract_mac_address(' MAC Address: 34:29:8F:12:0D:2F')
'34:29:8F:12:0D:2F'
Returns:
True if in_str is a slug string and False otherwise.
+ See also :meth:`is_camel_case`, :meth:`is_snake_case`, and :meth:`slugify`.
+
>>> is_slug('my-blog-post-title')
True
>>> is_slug('My blog post title')
True if the given string contains HTML/XML tags and False
otherwise.
+ See also :meth:`strip_html`.
+
.. warning::
By design, this function matches ANY type of tag, so don't expect
to use it as an HTML validator. It's a quick sanity check at
The number of words contained in the given string.
.. note::
-
This method is "smart" in that it does consider only sequences
of one or more letter and/or numbers to be "words". Thus a
string like this: "! @ # % ... []" will return zero. Moreover
The number of words contained in the given string.
.. note::
-
This method is "smart" in that it does consider only sequences
of one or more letter and/or numbers to be "words". Thus a
string like this: "! @ # % ... []" will return zero. Moreover
A generated UUID string (using `uuid.uuid4()`) with or without
dashes per the omit_dashes arg.
+ See also :meth:`is_uuid`, :meth:`generate_random_alphanumeric_string`.
+
generate_uuid() # possible output: '97e3a716-6b33-4ab9-9bb1-8128cb24d76b'
generate_uuid(omit_dashes=True) # possible output: '97e3a7166b334ab99bb18128cb24d76b'
"""
A string of the specified size containing random characters
(uppercase/lowercase ascii letters and digits).
+ See also :meth:`asciify`, :meth:`generate_uuid`.
+
>>> random.seed(22)
>>> generate_random_alphanumeric_string(9)
'96ipbNClS'
original string if it is not a valid camel case string or some
other error occurs.
+ See also :meth:`is_camel_case`, :meth:`is_snake_case`, and :meth:`is_slug`.
+
>>> camel_case_to_snake_case('MacAddressExtractorFactory')
'mac_address_extractor_factory'
>>> camel_case_to_snake_case('Luke Skywalker')
provided or the original string back again if it is not valid
snake case or another error occurs.
+ See also :meth:`is_camel_case`, :meth:`is_snake_case`, and :meth:`is_slug`.
+
>>> snake_case_to_camel_case('this_is_a_test')
'ThisIsATest'
>>> snake_case_to_camel_case('Han Solo')
Returns:
A list of strings of length one each.
+ See also :meth:`from_char_list`.
+
>>> to_char_list('test')
['t', 'e', 's', 't']
"""
The string resulting from gluing the characters in in_list
together.
+ See also :meth:`to_char_list`.
+
>>> from_char_list(['t', 'e', 's', 't'])
'test'
"""
in the same original string as no check is done. Returns
None to indicate error conditions.
+ See also :mod:`pyutils.unscrambler`.
+
>>> random.seed(22)
>>> scramble('awesome')
'meosaew'
A string with all HTML tags removed (optionally with tag contents
preserved).
+ See also :meth:`contains_html`.
+
.. note::
This method uses simple regular expressions to strip tags and is
not a full fledged HTML parser by any means. Consider using
by translating all non-ascii chars into their closest possible
ASCII representation (eg: ó -> o, Ë -> E, ç -> c...).
+ See also :meth:`to_ascii`, :meth:`generate_random_alphanumeric_string`.
+
.. warning::
Some chars may be lost if impossible to translate.
* all chars are encoded as ascii (by using :meth:`asciify`)
* is safe for URL
+ See also :meth:`is_slug` and :meth:`asciify`.
+
>>> slugify('Top 10 Reasons To Love Dogs!!!')
'top-10-reasons-to-love-dogs'
>>> slugify('Mönstér Mägnët')
Otherwise False is returned.
+ See also :mod:`pyutils.argparse_utils`.
+
>>> to_bool('True')
True
an error. This parser is relatively clever; see
:class:`datetimez.dateparse_utils` docs for details.
+ See also: :mod:`pyutils.datetimez.dateparse_utils`, :meth:`extract_date`,
+ :meth:`is_valid_date`, :meth:`to_datetime`, :meth:`valid_datetime`.
+
>>> to_date('9/11/2001')
datetime.date(2001, 9, 11)
>>> to_date('xyzzy')
Returns:
a datetime if date was found, otherwise None
+ See also: :mod:`pyutils.datetimez.dateparse_utils`, :meth:`to_date`,
+ :meth:`is_valid_date`, :meth:`to_datetime`, :meth:`valid_datetime`.
+
>>> extract_date("filename.txt dec 13, 2022")
datetime.datetime(2022, 12, 13, 0, 0)
and False otherwise. This parser is relatively clever; see
:class:`datetimez.dateparse_utils` docs for details.
+ See also: :mod:`pyutils.datetimez.dateparse_utils`, :meth:`to_date`,
+ :meth:`extract_date`, :meth:`to_datetime`, :meth:`valid_datetime`.
+
>>> is_valid_date('1/2/2022')
True
>>> is_valid_date('christmas')
an error. This parser is relatively clever; see
:class:`datetimez.dateparse_utils` docs for details.
+ See also: :mod:`pyutils.datetimez.dateparse_utils`, :meth:`to_date`,
+ :meth:`extract_date`, :meth:`valid_datetime`.
+
>>> to_datetime('7/20/1969 02:56 GMT')
datetime.datetime(1969, 7, 20, 2, 56, tzinfo=<StaticTzInfo 'GMT'>)
"""
Returns:
A string with tab indentation removed or None on error.
- .. note::
-
- Inspired by analogous Scala function.
+ See also :meth:`indent`.
>>> dedent('\t\ttest\\n\t\ting')
'test\\ning'
Returns:
An indented string created by prepending amount spaces.
+ See also :meth:`dedent`.
+
>>> indent('This is a test', 4)
' This is a test'
"""
return line_separator.join(lines)
-def sprintf(*args, **kwargs) -> str:
- """
- Args:
- This function uses the same syntax as the builtin print
- function.
-
- Returns:
- An interpolated string capturing print output, like man(3)
- `sprintf`.
- """
+def _sprintf(*args, **kwargs) -> str:
+ """Internal helper."""
ret = ""
sep = kwargs.pop("sep", None)
Returns:
in_str with recognized ANSI escape sequences removed.
+ See also :mod:`pyutils.ansi`.
+
.. warning::
This method works by using a regular expression.
It works for all ANSI escape sequences I've tested with but
>>> print(buf(), end='')
test
1, 2, 3
-
"""
def __init__(self) -> None:
'Test'
>>> capitalize_first_letter("ALREADY!")
'ALREADY!'
-
"""
return in_str[0].upper() + in_str[1:]
Returns:
'it' if n is one or 'they' otherwize.
+ See also :meth:`is_are`, :meth:`pluralize`, :meth:`make_contractions`,
+ :meth:`thify`.
+
Suggested usage::
n = num_files_saved_to_tmp()
Returns:
'is' if n is one or 'are' otherwize.
+ See also :meth:`it_they`, :meth:`pluralize`, :meth:`make_contractions`,
+ :meth:`thify`.
+
Suggested usage::
n = num_files_saved_to_tmp()
Returns:
's' if n is greater than one otherwize ''.
+ See also :meth:`it_they`, :meth:`is_are`, :meth:`make_contractions`,
+ :meth:`thify`.
+
Suggested usage::
n = num_files_saved_to_tmp()
Output text identical to original input except for any
recognized contractions are formed.
+ See also :meth:`it_they`, :meth:`is_are`, :meth:`make_contractions`.
+
.. note::
The order in which we create contractions is defined by the
implementation and what I thought made more sense when writing
Returns:
The proper cardinal suffix for a number.
+ See also :meth:`it_they`, :meth:`is_are`, :meth:`make_contractions`.
+
Suggested usage::
attempt_count = 0
Returns:
Generates the ngrams from the input string.
+ See also :meth:`ngrams_presplit`, :meth:`bigrams`, :meth:`trigrams`.
+
>>> [x for x in ngrams('This is a test', 2)]
['This is', 'is a', 'a test']
"""
def ngrams_presplit(words: Sequence[str], n: int):
"""
Same as :meth:`ngrams` but with the string pre-split.
+
+ See also :meth:`ngrams`, :meth:`bigrams`, :meth:`trigrams`.
"""
return list_utils.ngrams(words, n)
def bigrams(txt: str):
"""Generates the bigrams (n=2) of the given string.
+ See also :meth:`ngrams`, :meth:`trigrams`.
+
>>> [x for x in bigrams('this is a test')]
['this is', 'is a', 'a test']
"""
def trigrams(txt: str):
- """Generates the trigrams (n=3) of the given string."""
+ """Generates the trigrams (n=3) of the given string.
+
+ See also :meth:`ngrams`, :meth:`bigrams`.
+ """
return ngrams(txt, 3)
A list of string created by following the instructions set forth
in column_specs.
+ See also :meth:`shuffle_columns_into_dict`.
+
>>> cols = '-rwxr-xr-x 1 scott wheel 3.1K Jul 9 11:34 acl_test.py'.split()
>>> shuffle_columns_into_list(
... cols,
Returns:
A dict formed by applying the column_specs instructions.
+ See also :meth:`shuffle_columns_into_list`, :meth:`interpolate_using_dict`.
+
>>> cols = '-rwxr-xr-x 1 scott wheel 3.1K Jul 9 11:34 acl_test.py'.split()
>>> shuffle_columns_into_dict(
... cols,
txt: the mad libs template
values: what you and your kids chose for each category.
+ See also :meth:`shuffle_columns_into_list`, :meth:`shuffle_columns_into_dict`.
+
>>> interpolate_using_dict('This is a {adjective} {noun}.',
... {'adjective': 'good', 'noun': 'example'})
'This is a good example.'
"""
- return sprintf(txt.format(**values), end='')
+ return _sprintf(txt.format(**values), end='')
def to_ascii(txt: str):
Returns:
txt encoded as an ASCII byte string.
+ See also :meth:`to_base64`, :meth:`to_bitstring`, :meth:`to_bytes`,
+ :meth:`generate_random_alphanumeric_string`, :meth:`asciify`.
+
>>> to_ascii('test')
b'test'
txt encoded with a 64-chracter alphabet. Similar to and compatible
with uuencode/uudecode.
+ See also :meth:`is_base64`, :meth:`to_ascii`, :meth:`to_bitstring`,
+ :meth:`from_base64`.
+
>>> to_base64('hello?')
b'aGVsbG8/\\n'
"""
txt was encoded with Python's standard base64 alphabet which
is the same as what uuencode/uudecode uses).
+ See also :meth:`to_base64`, :meth:`from_base64`.
+
>>> is_base64('test') # all letters in the b64 alphabet
True
The decoded form of b64 as a normal python string. Similar to
and compatible with uuencode / uudecode.
+ See also :meth:`to_base64`, :meth:`is_base64`.
+
>>> from_base64(b'aGVsbG8/\\n')
'hello?'
"""
Returns:
txt converted to ascii/binary and then chopped into bytes.
+ See also :meth:`to_base64`, :meth:`from_bitstring`, :meth:`is_bitstring`,
+ :meth:`chunk`.
+
>>> to_bitstring('hello?')
'011010000110010101101100011011000110111100111111'
Note that if delimiter is non empty this code will not
recognize the bitstring.
+ See also :meth:`to_base64`, :meth:`from_bitstring`, :meth:`to_bitstring`,
+ :meth:`chunk`.
+
>>> is_bitstring('011010000110010101101100011011000110111100111111')
True
The regular python string represented by bits. Note that this
code does not work with to_bitstring when delimiter is non-empty.
+ See also :meth:`to_base64`, :meth:`to_bitstring`, :meth:`is_bitstring`,
+ :meth:`chunk`.
+
>>> from_bitstring('011010000110010101101100011011000110111100111111')
'hello?'
"""
IP addresses using a normal comparator will do something sane
and desireable.
+ See also :meth:`is_ip_v4`.
+
>>> ip_v4_sort_key('10.0.0.18')
(10, 0, 0, 18)
volumes using a normal comparator will do something sane
and desireable.
+ See also :mod:`pyutils.files.file_utils`.
+
>>> path_ancestors_before_descendants_sort_key('/usr/local/bin')
('usr', 'local', 'bin')
replacement: the character to replace any member of replace_set
with
+ See also :meth:`replace_nth`.
+
Returns:
The string with replacements executed.
target: the replacement text
nth: which occurrance of source to replace?
+ See also :meth:`replace_all`.
+
>>> replace_nth('this is a test', ' ', '-', 3)
'this is a-test'
"""
# © Copyright 2021-2022, Scott Gasch
-"""Utilities for dealing with "text"."""
+"""
+Utilities for dealing with and creating text chunks. For example:
+
+ - Make a bar graph / progress graph,
+ - make a spark line,
+ - left, right, center, justify text,
+ - word wrap text,
+ - indent / dedent text,
+ - create a header line,
+ - draw a box around some text.
+
+"""
import contextlib
import enum
redraw: if True, omit a line feed after the carriage return
so that subsequent calls to this method redraw the graph
iteratively.
+
+ See also :meth:`bar_graph_string`, :meth:`sparkline`.
+
+ Example::
+
+ '[███████████████████████████████████ ] 0.5'
+
"""
ret = "\r" if redraw else "\n"
bar = bar_graph_string(
left_end: the character at the left side of the graph
right_end: the character at the right side of the graph
+ See also :meth:`bar_graph`, :meth:`sparkline`.
+
>>> bar_graph_string(5, 10, fgcolor='', reset_seq='')
'[███████████████████████████████████ ] 0.5'
* the maximum number in the population
* a string representation of the population in a concise format
+ See also :meth:`bar_graph`, :meth:`bar_graph_string`.
+
>>> sparkline([1, 2, 3, 5, 10, 3, 5, 7])
(1, 10, '▁▁▂▄█▂▄▆')
Returns:
The distributed, justified string.
+ See also :meth:`justify_string`, :meth:`justify_text`.
+
>>> distribute_strings(['this', 'is', 'a', 'test'], width=40)
' this is a test '
"""
Returns:
The justified text.
+ See also :meth:`justify_text`.
+
>>> justify_text('This is a test of the emergency broadcast system. This is only a test.',
... width=40, alignment='j') #doctest: +NORMALIZE_WHITESPACE
'This is a test of the emergency\\nbroadcast system. This is only a test.'
Returns:
the box as a string
+ See also :meth:`print_box`, :meth:`preformatted_box`.
+
>>> print(box('title', 'this is some text', width=20).strip())
╭──────────────────╮
│ title │
Returns:
the box as a string
+ See also :meth:`print_box`, :meth:`box`.
+
>>> print(preformatted_box('title', 'this\\nis\\nsome\\ntext', width=20).strip())
╭──────────────────╮
│ title │
) -> None:
"""Draws a box with nice rounded corners.
+ Args:
+ title: the title of the box
+ text: the text inside the box
+ width: the width of the box
+ color: the box's color
+
+ Returns:
+ None
+
+ Side-effects:
+ Prints a box with your text on the console to sys.stdout.
+
+ See also :meth:`preformatted_box`, :meth:`box`.
+
>>> print_box('Title', 'This is text', width=30)
╭────────────────────────────╮
│ Title │
"""Helpers for unittests.
-.. note::
-
- When you import this we automatically wrap unittest.main()
- with a call to bootstrap.initialize so that we getLogger
- config, commandline args, logging control, etc... this works
- fine but it's a little hacky so caveat emptor.
+.. warning::
+ When you import this we automatically wrap the standard Python
+ `unittest.main` with a call to :meth:`pyutils.bootstrap.initialize`
+ so that we get logger config, commandline args, logging control,
+ etc... this works fine but may be unexpected behavior.
"""
import contextlib
help='Db connection spec for perf data (iff --unittest_persistance_strategy is DATABASE)',
)
-# >>> This is the hacky business, FYI. <<<
unittest.main = bootstrap.initialize(unittest.main)
class PerfRegressionDataPersister(ABC):
- """A base class for a signature dealing with persisting perf
- regression data."""
-
- def __init__(self):
- pass
+ """A base class that defines an interface for dealing with
+ persisting perf regression data.
+ """
@abstractmethod
def load_performance_data(self, method_id: str) -> Dict[str, List[float]]:
+ """Load the historical performance data for the supplied method.
+
+ Args:
+ method_id: the method for which we want historical perf data.
+ """
pass
@abstractmethod
def save_performance_data(self, method_id: str, data: Dict[str, List[float]]):
+ """Save the historical performance data of the supplied method.
+
+ Args:
+ method_id: the method whose historical perf data we're saving.
+ data: the historical performance data being persisted.
+ """
pass
@abstractmethod
def delete_performance_data(self, method_id: str):
+ """Delete the historical performance data of the supplied method.
+
+ Args:
+ method_id: the method whose data should be erased.
+ """
pass
"""A perf regression data persister that uses files."""
def __init__(self, filename: str):
+ """
+ Args:
+ filename: the filename to save/load historical performance data
+ """
super().__init__()
self.filename = filename
self.traces_to_delete: List[str] = []
def check_method_for_perf_regressions(func: Callable) -> Callable:
- """
- This is meant to be used on a method in a class that subclasses
- unittest.TestCase. When thus decorated it will time the execution
- of the code in the method, compare it with a database of
- historical perfmance, and fail the test with a perf-related
- message if it has become too slow.
+ """This decorator is meant to be used on a method in a class that
+ subclasses :class:`unittest.TestCase`. When decorated, method
+ execution timing (i.e. performance) will be measured and compared
+ with a database of historical performance for the same method.
+ The wrapper will then fail the test with a perf-related message if
+ it has become too slow.
+
+ See also :meth:`check_all_methods_for_perf_regressions`.
+
+ Example usage::
+
+ class TestMyClass(unittest.TestCase):
+
+ @check_method_for_perf_regressions
+ def test_some_part_of_my_class(self):
+ ...
"""
def check_all_methods_for_perf_regressions(prefix='test_'):
- """Decorate unittests with this to pay attention to the perf of the
- testcode and flag perf regressions. e.g.
+ """This decorator is meant to apply to classes that subclass from
+ :class:`unittest.TestCase` and, when applied, has the affect of
+ decorating each method that matches the `prefix` given with the
+ :meth:`check_method_for_perf_regressions` wrapper (see above).
+ This wrapper causes us to measure perf and fail tests that regress
+ perf dramatically.
+
+ Args:
+ prefix: the prefix of method names to check for regressions
+
+ See also :meth:`check_method_for_perf_regressions` to check only
+ a single method.
+
+ Example usage. By decorating the class, all methods with names
+ that begin with `test_` will be perf monitored::
- import pyutils.unittest_utils as uu
+ import pyutils.unittest_utils as uu
- @uu.check_all_methods_for_perf_regressions()
- class TestMyClass(unittest.TestCase):
+ @uu.check_all_methods_for_perf_regressions()
+ class TestMyClass(unittest.TestCase):
- def test_some_part_of_my_class(self):
- ...
+ def test_some_part_of_my_class(self):
+ ...
+ def test_som_other_part_of_my_class(self):
+ ...
"""
def decorate_the_testcase(cls):
class RecordStdout(contextlib.AbstractContextManager):
"""
- Record what is emitted to stdout.
+ Records what is emitted to stdout into a buffer instead.
>>> with RecordStdout() as record:
... print("This is a test!")
class RecordMultipleStreams(contextlib.AbstractContextManager):
"""
Record the output to more than one stream.
+
+ Example usage::
+
+ with RecordMultipleStreams(sys.stdout, sys.stderr) as r:
+ print("This is a test!", file=sys.stderr)
+ print("This is too", file=sys.stdout)
+
+ print(r().readlines())
+ r().close()
+
"""
def __init__(self, *files) -> None:
# © Copyright 2021-2022, Scott Gasch
-"""A fast word unscrambler library."""
+"""A fast (English) word unscrambler."""
import logging
from typing import Dict, Mapping, Optional
population and then using a pregenerated index to look up known
words the same set of letters.
- Note that each instance of Unscrambler caches its index to speed
- up lookups number 2..N; careless reinstantiation will by slower.
-
Sigs are designed to cluster similar words near each other so both
lookup methods support a "fuzzy match" argument that can be set to
request similar words that do not match exactly in addition to any
exact matches.
+
+ .. note::
+
+ Each instance of Unscrambler caches its index to speed up
+ lookups number 2..N; careless deletion / reinstantiation will
+ suffer slower performance.
"""
def __init__(self, indexfile: Optional[str] = None):
Constructs an unscrambler.
Args:
- indexfile: overrides the default indexfile location if provided
+ indexfile: overrides the default indexfile location if provided.
+ To [re]generate the indexfile, see :meth:`repopulate`.
"""
# Cached index per instance.
@staticmethod
def get_indexfile(indexfile: Optional[str]) -> str:
- """Returns the current indexfile location."""
+ """
+ Returns:
+ The current indexfile location
+ """
if indexfile is None:
if 'unscrambler_default_indexfile' in config.config:
indexfile = config.config['unscrambler_default_indexfile']
Returns:
The word's signature.
- >>> train = Unscrambler.compute_word_sig('train')
- >>> train
- 23178969883741
+ >>> test = Unscrambler.compute_word_sig('test')
+ >>> test
+ 105560478284788
- >>> retain = Unscrambler.compute_word_sig('retrain')
- >>> retain
- 24282502197479
+ >>> teste = Unscrambler.compute_word_sig('teste')
+ >>> teste
+ 105562386542095
- >>> retain - train
- 1103532313738
+ >>> teste - test
+ 1908257307
"""
population = list_utils.population_counts(word)
"""
Repopulates the indexfile.
+ Args:
+ dictfile: a file that contains one word per line
+ indexfile: the file to populate with sig, word pairs for future use
+ by this class.
+
.. warning::
- Before calling this method, change letter_sigs from the
+ Before calling this method, change `letter_sigs` from the
default above unless you want to populate the same exact
files.
"""
return ret
-#
-# To repopulate, change letter_sigs and then call Unscrambler.repopulate.
-# See notes above. See also ~/bin/unscramble.py --populate_destructively.
-#
-
-
if __name__ == "__main__":
import doctest
# © Copyright 2022, Scott Gasch
-"""This is a module for making it easier to deal with Zookeeper / Kazoo."""
+"""
+This is a module for making it easier to deal with Zookeeper / Kazoo.
+Apache Zookeeper (https://zookeeper.apache.org/) is a consistent centralized
+datastore. :mod:`pyutils.config` optionally uses it to save/read program
+configuration. But it's also very useful for things like distributed
+master election, locking, etc...
+"""
import datetime
def get_started_zk_client() -> KazooClient:
+ """
+ Returns:
+ A zk client library reference that has been connected and started
+ using the commandline provided address, certificates and passphrase.
+ """
zk = KazooClient(
hosts=config.config['zookeeper_nodes'],
use_ssl=True,
class RenewableReleasableLease(NonBlockingLease):
"""This is a hacky subclass of kazoo.recipe.lease.NonBlockingLease
- that adds some behaviors:
+ (see https://kazoo.readthedocs.io/en/latest/api/recipe/lease.html#kazoo.recipe.lease.NonBlockingLease) that adds some behaviors:
+ Ability to renew the lease if it's already held without
going through the effort of reobtaining the same lease
identifier: str = None,
utcnow=datetime.datetime.utcnow,
):
+ """Construct the RenewableReleasableLease.
+
+ Args:
+ client: a KazooClient that is connected and started
+ path: the path to the lease in zookeeper
+ duration: duration during which the lease is reserved
+ identifier: unique name to use for this lease holder.
+ Reuse in order to renew the lease.
+ utcnow: clock function, by default returning
+ :meth:`datetime.datetime.utcnow`. Used for testing.
+
+ """
super().__init__(client, path, duration, identifier, utcnow)
self.client = client
self.path = path
Args:
duration: the amount of additional time to add to the
- current lease expiration.
+ current lease expiration.
Returns:
True if the lease was successfully renewed,
return False
def __bool__(self):
- """Note that this implementation differs from that of the base
- class in that it probes zookeeper to ensure that the lease is
- not yet expired and is therefore more expensive.
"""
+ .. note:
+ This implementation differs from that of the base class in
+ that it probes zookeeper to ensure that the lease is not yet
+ expired and is therefore more expensive.
+
+ """
if not self.obtained:
return False
lock = self.client.Lock(self.path, self.identifier)
projects / pyutils.git / commitdiff