class StringWildcardBasedACL(PredicateListBasedACL):
- """An ACL that allows or denies based on string glob (*, ?) patterns."""
+ """An ACL that allows or denies based on string glob :code:`(*, ?)`
+ patterns.
+ """
def __init__(
self,
class ActionNoYes(argparse.Action):
- """An argparse Action that allows for commandline arguments like this:
+ """An argparse Action that allows for commandline arguments like this::
cfg.add_argument(
'--enable_the_thing',
help='Should we enable the thing?'
)
- This creates cmdline arguments:
+ This creates the following cmdline arguments::
--enable_the_thing
--no_enable_the_thing
@persistent.persistent_autoloaded_singleton() # type: ignore
class CachedWeatherData(persistent.Persistent):
def __init__(self, weather_data: Dict[datetime.date, WeatherData] = None):
+ """C'tor. Do not pass a dict except for testing purposes.
+
+ The @persistent_autoloaded_singleton decorator handles
+ invoking our load and save methods at construction time for
+ you.
+ """
+
if weather_data is not None:
self.weather_data = weather_data
return
@classmethod
@overrides
def load(cls) -> Any:
+
+ """Depending on whether we have fresh data persisted either uses that
+ data to instantiate the class or makes an HTTP request to fetch the
+ necessary data.
+
+ Note that because this is a subclass of Persistent this is taken
+ care of automatically.
+ """
+
if persistent.was_file_written_within_n_seconds(
config.config['weather_data_cachefile'],
config.config['weather_data_stalest_acceptable'].total_seconds(),
@overrides
def save(self) -> bool:
+ """
+ Saves the current data to disk if required. Again, because this is
+ a subclass of Persistent this is taken care of for you.
+ """
+
import pickle
with open(config.config['weather_data_cachefile'], 'wb') as wf:
Usage:
- module.py:
- ----------
- import config
-
- parser = config.add_commandline_args(
- "Module",
- "Args related to module doing the thing.",
- )
- parser.add_argument(
- "--module_do_the_thing",
- type=bool,
- default=True,
- help="Should the module do the thing?"
- )
-
- main.py:
- --------
- import config
-
- def main() -> None:
+ In your file.py::
+
+ import config
+
parser = config.add_commandline_args(
- "Main",
- "A program that does the thing.",
+ "Module",
+ "Args related to module doing the thing.",
)
parser.add_argument(
- "--dry_run",
+ "--module_do_the_thing",
type=bool,
- default=False,
- help="Should we really do the thing?"
+ default=True,
+ help="Should the module do the thing?"
)
- config.parse() # Very important, this must be invoked!
+
+ In your main.py::
+
+ import config
+
+ def main() -> None:
+ parser = config.add_commandline_args(
+ "Main",
+ "A program that does the thing.",
+ )
+ parser.add_argument(
+ "--dry_run",
+ type=bool,
+ default=False,
+ help="Should we really do the thing?"
+ )
+ config.parse() # Very important, this must be invoked!
If you set this up and remember to invoke config.parse(), all commandline
arguments will play nicely together. This is done automatically for you
- if you're using the bootstrap module's initialize wrapper.
+ if you're using the bootstrap module's initialize wrapper.::
- % main.py -h
- usage: main.py [-h]
- [--module_do_the_thing MODULE_DO_THE_THING]
- [--dry_run DRY_RUN]
+ % main.py -h
+ usage: main.py [-h]
+ [--module_do_the_thing MODULE_DO_THE_THING]
+ [--dry_run DRY_RUN]
- Module:
- Args related to module doing the thing.
+ Module:
+ Args related to module doing the thing.
- --module_do_the_thing MODULE_DO_THE_THING
- Should the module do the thing?
+ --module_do_the_thing MODULE_DO_THE_THING
+ Should the module do the thing?
- Main:
- A program that does the thing
+ Main:
+ A program that does the thing
- --dry_run
- Should we really do the thing?
+ --dry_run
+ Should we really do the thing?
Arguments themselves should be accessed via
- config.config['arg_name']. e.g.
+ :code:`config.config['arg_name']`. e.g.::
- if not config.config['dry_run']:
- module.do_the_thing()
+ if not config.config['dry_run']:
+ module.do_the_thing()
"""
"RAW|". In that case, the line breaks are preserved and the text
is not wrapped.
+ Use this, for example, when you need the helptext of an argument
+ to have its spacing preserved exactly, e.g.::
+
+ args.add_argument(
+ '--mode',
+ type=str,
+ default='PLAY',
+ choices=['CHEAT', 'AUTOPLAY', 'SELFTEST', 'PRECOMPUTE', 'PLAY'],
+ metavar='MODE',
+ help='''RAW|Our mode of operation. One of:
+
+ PLAY = play wordle with me! Pick a random solution or
+ specify a solution with --template.
+
+ CHEAT = given a --template and, optionally, --letters_in_word
+ and/or --letters_to_avoid, return the best guess word;
+
+ AUTOPLAY = given a complete word in --template, guess it step
+ by step showing work;
+
+ SELFTEST = autoplay every possible solution keeping track of
+ wins/losses and average number of guesses;
+
+ PRECOMPUTE = populate hash table with optimal guesses.
+ ''',
+ )
+
"""
def _split_lines(self, text, width):
# It would be really nice if this shit worked from interactive python
-def add_commandline_args(title: str, description: str = ""):
- """Create a new context for arguments and return a handle."""
+def add_commandline_args(title: str, description: str = "") -> argparse._ArgumentGroup:
+ """Create a new context for arguments and return a handle.
+
+ Args:
+ title: A title for your module's commandline arguments group.
+ description: A helpful description of your module.
+
+ Returns:
+ An argparse._ArgumentGroup to be populated by the caller.
+ """
return ARGS.add_argument_group(title, description)
def overwrite_argparse_epilog(msg: str) -> None:
+ """Allows your code to override the default epilog created by
+ argparse.
+
+ Args:
+ msg: The epilog message to substitute for the default.
+ """
ARGS.epilog = msg
-def is_flag_already_in_argv(var: str):
- """Is a particular flag passed on the commandline?"""
+def is_flag_already_in_argv(var: str) -> bool:
+ """Returns true if a particular flag is passed on the commandline?
+
+ Args:
+ var: The flag to search for.
+ """
for _ in sys.argv:
if var in _:
return True
return False
-def reorder_arg_action_groups_before_help(entry_module: Optional[str]):
+def _reorder_arg_action_groups_before_help(entry_module: Optional[str]):
+ """Internal. Used to reorder the arguments before dumping out a
+ generated help string such that the main program's arguments come
+ last.
+
+ """
reordered_action_groups = []
for grp in ARGS._action_groups:
if entry_module is not None and entry_module in grp.title: # type: ignore
return reordered_action_groups
-def augment_sys_argv_from_environment_variables():
+def _augment_sys_argv_from_environment_variables():
+ """Internal. Look at the system environment for variables that match
+ arg names. This is done via some munging such that:
+
+ :code:`--argument_to_match`
+
+ ...is matched by:
+
+ :code:`ARGUMENT_TO_MATCH`
+
+ This allows programmers to set args via shell environment variables
+ in lieu of passing them on the cmdline.
+
+ """
+
usage_message = ARGS.format_usage()
optional = False
var = ''
env = ''
-def augment_sys_argv_from_loadfile():
+def _augment_sys_argv_from_loadfile():
+ """Internal. Augment with arguments persisted in a saved file."""
+
loadfile = None
saw_other_args = False
grab_next_arg = False
def parse(entry_module: Optional[str]) -> Dict[str, Any]:
"""Main program should call this early in main(). Note that the
- bootstrap.initialize wrapper takes care of this automatically.
+ :code:`bootstrap.initialize` wrapper takes care of this automatically.
+ This should only be called once per program invocation.
"""
global CONFIG_PARSE_CALLED
if arg in ('--help', '-h'):
if entry_module is not None:
entry_module = os.path.basename(entry_module)
- ARGS._action_groups = reorder_arg_action_groups_before_help(entry_module)
+ ARGS._action_groups = _reorder_arg_action_groups_before_help(entry_module)
# Examine the environment for variables that match known flags.
# For a flag called --example_flag the corresponding environment
# variable would be called EXAMPLE_FLAG. If found, hackily add
# these into sys.argv to be parsed.
- augment_sys_argv_from_environment_variables()
+ _augment_sys_argv_from_environment_variables()
# Look for loadfile and read/parse it if present. This also
# works by jamming these values onto sys.argv.
- augment_sys_argv_from_loadfile()
+ _augment_sys_argv_from_loadfile()
# Parse (possibly augmented, possibly completely overwritten)
# commandline args with argparse normally and populate config.
def has_been_parsed() -> bool:
- """Has the global config been parsed yet?"""
+ """Returns True iff the global config has already been parsed"""
return CONFIG_PARSE_CALLED
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
- 'sphinx.ext.doctest',
'sphinx.ext.autodoc',
+ 'sphinx.ext.doctest',
+ 'sphinx.ext.napoleon',
'sphinx.ext.viewcode',
]
+autodoc_typehints = "both"
+
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
================================================
.. toctree::
- :maxdepth: 2
+ :maxdepth: 3
:caption: Contents:
modules
various logging levels, different files, different file handles,
the house log, etc...). See also OutputMultiplexerContext for an
easy usage pattern.
-
"""
class Destination(enum.IntEnum):
class OutputMultiplexerContext(OutputMultiplexer, contextlib.ContextDecorator):
"""
- A context that uses an OutputMultiplexer. e.g.
+ A context that uses an OutputMultiplexer. e.g.::
with OutputMultiplexerContext(
OutputMultiplexer.LOG_INFO |
handles = [ f, g ]
) as mplex:
mplex.print("This is a log message!")
-
"""
def __init__(
def parallelize(
_funct: typing.Optional[typing.Callable] = None, *, method: Method = Method.THREAD
) -> typing.Callable:
- """Usage:
+ """Usage::
@parallelize # defaults to thread-mode
def my_function(a, b, c) -> int:
def my_other_other_function(g, h) -> int:
...this work will be distributed to a remote machine pool
- This decorator will invoke the wrapped function on:
+ This decorator will invoke the wrapped function on::
Method.THREAD (default): a background thread
Method.PROCESS: a background process
"""
Save this thing somewhere that you'll remember when someone calls
load() later on in a way that makes sense to your code.
-
"""
pass
below) be save()d at program exit time.
Oh, in case this is handy, here's how to write a factory
- method that doesn't call the c'tor in python:
+ method that doesn't call the c'tor in python::
@classmethod
def load_from_somewhere(cls, somewhere):
# Load the piece(s) of obj that you want to from somewhere.
obj._state = load_from_somewhere(somewhere)
return obj
-
"""
pass
"""
An enum to describe the conditions under which state is persisted
to disk. See details below.
-
"""
NEVER = (0,)
The implementations of save() and load() and where the class
persists its state are details left to the Persistent
implementation.
-
"""
def __init__(
polling. StateTracker expects to be invoked periodically to maintain
state whereas the others automatically update themselves and,
optionally, expose an event for client code to wait on state changes.
-
"""
import datetime
invoked via the heartbeat() method. This method, in turn, invokes
update() with update_ids according to a schedule / periodicity
provided to the c'tor.
-
"""
def __init__(self, update_ids_to_update_secs: Dict[str, float]) -> None:
Note that, when more than one update is overdue, they will be
invoked in order by their update_ids so care in choosing these
identifiers may be in order.
-
"""
self.update_ids_to_update_secs = update_ids_to_update_secs
self.last_reminder_ts: Dict[str, Optional[datetime.datetime]] = {}
The now param is the approximate current timestamp and the
last_invocation param is the last time you were invoked (or
None on the first invocation)
-
"""
pass
Setting force_all_updates_to_run will invoke all updates
(ordered by update_id) immediately ignoring whether or not
they are due.
-
"""
+
self.now = datetime.datetime.now(tz=pytz.timezone("US/Pacific"))
for update_id in sorted(self.last_reminder_ts.keys()):
if force_all_updates_to_run:
"""Just like HeartbeatCurrentState but you don't need to pump the
heartbeat; it runs on a background thread. Call .shutdown() to
terminate the updates.
-
"""
@background_thread
"""Entry point for a background thread to own calling heartbeat()
at regular intervals so that the main thread doesn't need to do
so.
-
"""
while True:
if should_terminate.is_set():
def shutdown(self):
"""Terminates the background thread and waits for it to tear down.
This may block for as long as self.sleep_delay.
-
"""
+
logger.debug('Setting shutdown event and waiting for background thread.')
self.should_terminate.set()
self.updater_thread.join()
simply timed out. If the return value is true, the instance
should be reset() before wait is called again.
- Example usage:
+ Example usage::
detector = waitable_presence.WaitableAutomaticStateSubclass()
while True:
else:
# Just a timeout; no need to reset. Maybe do something
# else before looping up into wait again.
-
"""
def __init__(
class SprintfStdout(contextlib.AbstractContextManager):
"""
- A context manager that captures outputs to stdout.
+ A context manager that captures outputs to stdout to a buffer
+ without printing them. e.g.::
- with SprintfStdout() as buf:
- print("test")
- print(buf())
+ with SprintfStdout() as buf:
+ print("test")
+ print("1, 2, 3")
+ print(buf())
+
+ This yields::
+
+ 'test\\n1, 2, 3\\n'
- 'test\n'
"""
def __init__(self) -> None:
class Indenter(contextlib.AbstractContextManager):
"""
- with Indenter(pad_count = 8) as i:
- i.print('test')
- with i:
- i.print('-ing')
+ Context manager that indents stuff (even recursively). e.g.::
+
+ with Indenter(pad_count = 8) as i:
+ i.print('test')
with i:
- i.print('1, 2, 3')
+ i.print('-ing')
+ with i:
+ i.print('1, 2, 3')
+
+ Yields::
+
+ test
+ -ing
+ 1, 2, 3
"""
*** event as an input parameter and should periodically check ***
*** it and stop if the event is set. ***
- Usage:
+ Usage::
@background_thread
def random(a: int, b: str, stop_event: threading.Event) -> None:
if stop_event.is_set():
return
-
def main() -> None:
(thread, event) = random(22, "dude")
print("back!")
Returns a Thread object and an Event that, when signaled, will stop
the invocations. Note that it is possible to be invoked one time
after the Event is set. This event can be used to stop infinite
- invocation style or finite invocation style decorations.
+ invocation style or finite invocation style decorations.::
@periodically_invoke(period_sec=0.5, stop_after=None)
def there(name: str, age: int) -> None:
print(f" ...there {name}, {age}")
-
@periodically_invoke(period_sec=1.0, stop_after=3)
def hello(name: str) -> None:
print(f"Hello, {name}")
"""
This is a waitable class that keeps a PresenceDetector internally
and periodically polls it to detect changes in presence in a
- particular location. Example suggested usage pattern:
+ particular location. Example suggested usage pattern::
detector = waitable_presence.WaitablePresenceDetectorWithMemory(60.0)
while True:
projects / python_utils.git / commitdiff