#!/usr/bin/env python3
+"""This is a module dealing with trying to guess a person's location
+based on the location of certain devices (e.g. phones, laptops)
+belonging to that person. It works with networks I run that log
+device MAC addresses active."""
+
import datetime
import logging
import re
class PresenceDetection(object):
+ """See above. This is a base class for determining a person's
+ location on networks I administer."""
+
def __init__(self) -> None:
# Note: list most important devices first.
self.devices_by_person: Dict[Person, List[str]] = {
if mac not in self.names_by_mac:
continue
mac_name = self.names_by_mac[mac]
- logger.debug(
- 'Looking for %s... check for mac %s (%s)',
- name, mac, mac_name
- )
+ logger.debug('Looking for %s... check for mac %s (%s)', name, mac, mac_name)
for location in self.location_ts_by_mac:
if mac in self.location_ts_by_mac[location]:
ts = (self.location_ts_by_mac[location])[mac]
- logger.debug(
- 'Seen %s (%s) at %s since %s',
- mac, mac_name, location, ts
- )
+ logger.debug('Seen %s (%s) at %s since %s', mac, mac_name, location, ts)
tiebreaks[location] = ts
(
for person in Person:
print(f'{person} => {p.where_is_person_now(person)}')
print()
-
-
-# for location in Location:
-# print(f'{location} => {p.is_anyone_in_location_now(location)}')
+ for location in Location:
+ print(f'{location} => {p.is_anyone_in_location_now(location)}')
if __name__ == '__main__':
#!/usr/bin/env python3
+"""This is a module for wrapping around python programs and doing some
+minor setup and tear down work for them. With it, you can break into
+pdb on unhandled top level exceptions, profile your code by passing a
+commandline argument in, audit module import events, examine where
+memory is being used in your program, and so on."""
+
import functools
import importlib
import logging
class ImportInterceptor(importlib.abc.MetaPathFinder):
+ """An interceptor that always allows module load events but dumps a
+ record into the log and onto stdout when modules are loaded and
+ produces an audit of who imported what at the end of the run. It
+ can't see any load events that happen before it, though, so move
+ bootstrap up in your __main__'s import list just temporarily to
+ get a good view.
+
+ """
+
def __init__(self):
import collect.trie
def repopulate_modules_by_filename(self):
self.module_by_filename_cache.clear()
- for mod in sys.modules:
- if hasattr(sys.modules[mod], '__file__'):
- fname = getattr(sys.modules[mod], '__file__')
+ for _, mod in sys.modules.copy().items(): # copy here because modules is volatile
+ if hasattr(mod, '__file__'):
+ fname = getattr(mod, '__file__')
else:
fname = 'unknown'
self.module_by_filename_cache[fname] = mod
# Maybe log some info about the python interpreter itself.
logger.debug(
- 'Platform: %s, maxint=0x%x, byteorder=%s',
- sys.platform, sys.maxsize, sys.byteorder
+ 'Platform: %s, maxint=0x%x, byteorder=%s', sys.platform, sys.maxsize, sys.byteorder
)
logger.debug('Python interpreter version: %s', sys.version)
logger.debug('Python implementation: %s', sys.implementation)
'child system: %.4fs\n'
'machine uptime: %.4fs\n'
'walltime: %.4fs',
- utime, stime, cutime, cstime, elapsed_time, walltime
+ utime,
+ stime,
+ cutime,
+ cstime,
+ elapsed_time,
+ walltime,
)
# If it doesn't return cleanly, call attention to the return value.
# taking any unnecessary dependencies back on them.
# Defer logging messages until later when logging has been initialized.
-saved_messages: List[str] = []
+SAVED_MESSAGES: List[str] = []
# Make a copy of the original program arguments.
-program_name: str = os.path.basename(sys.argv[0])
-original_argv: List[str] = [arg for arg in sys.argv]
+PROGRAM_NAME: str = os.path.basename(sys.argv[0])
+ORIG_ARGV: List[str] = sys.argv.copy()
class OptionalRawFormatter(argparse.HelpFormatter):
# A global parser that we will collect arguments into.
-args = argparse.ArgumentParser(
+ARGS = argparse.ArgumentParser(
description=None,
formatter_class=OptionalRawFormatter,
fromfile_prefix_chars="@",
- epilog=f'{program_name} uses config.py ({__file__}) for global, cross-module configuration setup and parsing.',
+ epilog=f'{PROGRAM_NAME} uses config.py ({__file__}) for global, cross-module configuration setup and parsing.',
)
# Keep track of if we've been called and prevent being called more
# than once.
-config_parse_called = False
+CONFIG_PARSE_CALLED = False
# A global configuration dictionary that will contain parsed arguments.
# This is the data that is most interesting to our callers; it will hold
# the configuration result.
config: Dict[str, Any] = {}
+
# 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."""
- return args.add_argument_group(title, description)
+ return ARGS.add_argument_group(title, description)
group = add_commandline_args(
def overwrite_argparse_epilog(msg: str) -> None:
- args.epilog = msg
+ ARGS.epilog = msg
def is_flag_already_in_argv(var: str):
def reorder_arg_action_groups(entry_module: Optional[str]):
- global program_name, args
reordered_action_groups = []
- for group in args._action_groups:
- if entry_module is not None and entry_module in group.title: # type: ignore
- reordered_action_groups.append(group)
- elif program_name in group.title: # type: ignore
- reordered_action_groups.append(group)
+ for grp in ARGS._action_groups:
+ if entry_module is not None and entry_module in grp.title: # type: ignore
+ reordered_action_groups.append(grp)
+ elif PROGRAM_NAME in group.title: # type: ignore
+ reordered_action_groups.append(grp)
else:
- reordered_action_groups.insert(0, group)
+ reordered_action_groups.insert(0, grp)
return reordered_action_groups
def augment_sys_argv_from_environment_variables():
- global saved_messages
- usage_message = args.format_usage()
+ usage_message = ARGS.format_usage()
optional = False
var = ''
for x in usage_message.split():
if env in os.environ:
if not is_flag_already_in_argv(var):
value = os.environ[env]
- saved_messages.append(f'Initialized from environment: {var} = {value}')
+ SAVED_MESSAGES.append(f'Initialized from environment: {var} = {value}')
from string_utils import to_bool
if len(chunks) == 1 and to_bool(value):
def augment_sys_argv_from_loadfile():
- global saved_messages
loadfile = None
saw_other_args = False
grab_next_arg = False
else:
msg = f'Reading commandline arguments from {loadfile}.'
print(msg, file=sys.stderr)
- saved_messages.append(msg)
+ SAVED_MESSAGES.append(msg)
with open(loadfile, 'r') as rf:
newargs = rf.readlines()
bootstrap.initialize wrapper takes care of this automatically.
"""
- global config_parse_called
- if config_parse_called:
+ global CONFIG_PARSE_CALLED
+ if CONFIG_PARSE_CALLED:
return config
- global saved_messages
# If we're about to do the usage message dump, put the main
# module's argument group last in the list (if possible) so that
# when the user passes -h or --help, it will be visible on the
# screen w/o scrolling.
for arg in sys.argv:
- if arg == '--help' or arg == '-h':
- args._action_groups = reorder_arg_action_groups(entry_module)
+ if arg in ('--help', '-h'):
+ ARGS._action_groups = reorder_arg_action_groups(entry_module)
# Examine the environment for variables that match known flags.
# For a flag called --example_flag the corresponding environment
# Parse (possibly augmented, possibly completely overwritten)
# commandline args with argparse normally and populate config.
- known, unknown = args.parse_known_args()
+ known, unknown = ARGS.parse_known_args()
config.update(vars(known))
# Reconstruct the argv with unrecognized flags for the benefit of
raise Exception(
f'Encountered unrecognized config argument(s) {unknown} with --config_rejects_unrecognized_arguments enabled; halting.'
)
- saved_messages.append(f'Config encountered unrecognized commandline arguments: {unknown}')
+ SAVED_MESSAGES.append(f'Config encountered unrecognized commandline arguments: {unknown}')
sys.argv = sys.argv[:1] + unknown
# Check for savefile and populate it if requested.
savefile = config['config_savefile']
if savefile and len(savefile) > 0:
with open(savefile, 'w') as wf:
- wf.write("\n".join(original_argv[1:]))
+ wf.write("\n".join(ORIG_ARGV[1:]))
# Also dump the config on stderr if requested.
if config['config_dump']:
dump_config()
- config_parse_called = True
+ CONFIG_PARSE_CALLED = True
return config
def has_been_parsed() -> bool:
"""Has the global config been parsed yet?"""
- global config_parse_called
- return config_parse_called
+ return CONFIG_PARSE_CALLED
def dump_config():
def late_logging():
"""Log messages saved earlier now that logging has been initialized."""
logger = logging.getLogger(__name__)
- global saved_messages
- for _ in saved_messages:
+ for _ in SAVED_MESSAGES:
logger.debug(_)
projects / python_utils.git / commitdiff