efro package

Subpackages

• efro.dataclassio package
  • Submodules
  • efro.dataclassio.extras module
    • DataclassDiff
    • dataclass_diff()
  • Module contents
    • Codec
      • Codec.FIRESTORE
      • Codec.JSON
    • DataclassFieldLookup
      • DataclassFieldLookup.path()
      • DataclassFieldLookup.paths()
    • IOAttrs
      • IOAttrs.MISSING
      • IOAttrs.soft_default
      • IOAttrs.soft_default_factory
      • IOAttrs.storagename
      • IOAttrs.store_default
      • IOAttrs.validate_datetime()
      • IOAttrs.validate_for_field()
      • IOAttrs.whole_days
      • IOAttrs.whole_hours
      • IOAttrs.whole_minutes
    • IOExtendedData
      • IOExtendedData.did_input()
      • IOExtendedData.handle_input_error()
      • IOExtendedData.will_input()
      • IOExtendedData.will_output()
    • IOMultiType
      • IOMultiType.get_type()
      • IOMultiType.get_type_id()
      • IOMultiType.get_type_id_storage_name()
      • IOMultiType.get_type_id_type()
    • JsonStyle
      • JsonStyle.FAST
      • JsonStyle.PRETTY
      • JsonStyle.SORTED
    • dataclass_from_dict()
    • dataclass_from_json()
    • dataclass_hash()
    • dataclass_to_dict()
    • dataclass_to_json()
    • dataclass_validate()
    • ioprep()
    • ioprepped()
    • is_ioprepped_dataclass()
    • will_ioprep()
• efro.message package
  • Module contents
    • BoolResponse
      • BoolResponse.value
    • BoundMessageReceiver
      • BoundMessageReceiver.encode_error_response()
      • BoundMessageReceiver.protocol
    • BoundMessageSender
      • BoundMessageSender.fetch_raw_response_async_untyped()
      • BoundMessageSender.protocol
      • BoundMessageSender.send_async_untyped()
      • BoundMessageSender.send_untyped()
      • BoundMessageSender.unpack_raw_response_untyped()
    • EmptySysResponse
    • ErrorSysResponse
      • ErrorSysResponse.ErrorType
      • ErrorSysResponse.error_message
      • ErrorSysResponse.error_type
    • Message
      • Message.get_response_types()
    • MessageProtocol
      • MessageProtocol.decode_dict()
      • MessageProtocol.do_create_receiver_module()
      • MessageProtocol.do_create_sender_module()
      • MessageProtocol.encode_dict()
      • MessageProtocol.error_to_response()
      • MessageProtocol.message_from_dict()
      • MessageProtocol.message_to_dict()
      • MessageProtocol.response_from_dict()
      • MessageProtocol.response_to_dict()
    • MessageReceiver
      • MessageReceiver.decode_filter_method()
      • MessageReceiver.encode_error_response()
      • MessageReceiver.encode_filter_method()
      • MessageReceiver.encode_user_response()
      • MessageReceiver.handle_raw_message()
      • MessageReceiver.handle_raw_message_async()
      • MessageReceiver.is_async
      • MessageReceiver.register_handler()
      • MessageReceiver.validate()
    • MessageSender
      • MessageSender.decode_filter_method()
      • MessageSender.encode_filter_method()
      • MessageSender.fetch_raw_response()
      • MessageSender.fetch_raw_response_async()
      • MessageSender.peer_desc_method()
      • MessageSender.send()
      • MessageSender.send_async()
      • MessageSender.send_async_ex_method()
      • MessageSender.send_async_method()
      • MessageSender.send_ex_method()
      • MessageSender.send_method()
      • MessageSender.unpack_raw_response()
    • Response
    • StringResponse
      • StringResponse.value
    • SysResponse
      • SysResponse.get_local_exception()
      • SysResponse.set_local_exception()
    • UnregisteredMessageIDError
    • create_receiver_module()
    • create_sender_module()

Submodules

efro.call module

Call related functionality shared between all efro components.

class efro.call.SimpleCallbackSet

Bases: Generic[T]

A simple way to manage a set of callbacks.

add(call: T) → None

Add a callback.

getcalls() → list[T]

Return the current set of registered calls.

class efro.call.SimpleCallbackSetEntry(call: T)

Bases: Generic[T]

An entry for a callback set.

efro.cloudshell module

My nifty ssh/mosh/rsync mishmash.

class efro.cloudshell.HostConfig(address: str | None = None, user: str = 'ubuntu', port: int = 22, mosh_port: int | None = None, mosh_port_2: int | None = None, mosh_server_path: str | None = None, mosh_shell: str = 'sh', workspaces_root: str = '/home/${USER}/cloudshell_workspaces', sync_perms: bool = True, precommand_noninteractive: str | None = None, precommand_interactive: str | None = None, managed: bool = False, region: str | None = None, idle_minutes: int = 5, can_sudo_reboot: bool = False, max_sessions: int = 4, reboot_wait_seconds: int = 20, reboot_attempts: int = 1)

Bases: object

Config for a cloud machine to run commands on.

precommand, if set, will be run before the passed commands. Note that it is not run in interactive mode (when no command is given).

address: str | None = None

can_sudo_reboot: bool = False

idle_minutes: int = 5

managed: bool = False

max_sessions: int = 4

mosh_port: int | None = None

mosh_port_2: int | None = None

mosh_server_path: str | None = None

mosh_shell: str = 'sh'

port: int = 22

precommand_interactive: str | None = None

precommand_noninteractive: str | None = None

reboot_attempts: int = 1

reboot_wait_seconds: int = 20

region: str | None = None

resolved_workspaces_root() → str

Returns workspaces_root with standard substitutions.

sync_perms: bool = True

user: str = 'ubuntu'

workspaces_root: str = '/home/${USER}/cloudshell_workspaces'

class efro.cloudshell.LockType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Types of locks that can be acquired on a host.

CLION = 'clion'

HOST = 'host'

PYCHARM = 'pycharm'

WORKSPACE = 'workspace'

efro.debug module

Utilities for debugging memory leaks or other issues.

IMPORTANT - these functions use the gc module which looks ‘under the hood’ at Python and sometimes returns not-fully-initialized objects, which may cause crashes or errors due to suddenly having references to them that they didn’t expect, etc. See https://github.com/python/cpython/issues/59313. For this reason, these methods should NEVER be called in production code. Enable them only for debugging situations and be aware that their use may itself cause problems. The same is true for the gc module itself.

class efro.debug.DeadlockDumper(timeout: float)

Bases: object

Dumps thread states if still around after timeout seconds.

This uses low level Python functionality so should still fire even in the case of deadlock.

lock = <unlocked _thread.lock object>

watch_in_progress = False

class efro.debug.DeadlockWatcher(timeout: float = 10.0, logger: Logger | None = None, logextra: dict | None = None)

Bases: object

Individual watcher for deadlock conditions.

Use the enable_deadlock_watchers() to enable this system.

Next, use these wrapped in a with statement around some operation that may deadlock. If the with statement does not complete within the timeout period, a traceback of all threads will be dumped.

Note that the checker thread runs a cycle every ~5 seconds, so something stuck needs to remain stuck for 5 seconds or so to be caught for sure.

classmethod enable_deadlock_watchers() → None

Spins up deadlock-watcher functionality.

Must be explicitly called before any DeadlockWatchers are created.

watchers: list[weakref.ref[DeadlockWatcher]] | None = None

watchers_lock: threading.Lock | None = None

efro.debug.getobj(objid: int, expanded: bool = False) → Any

Return a garbage-collected object by its id.

Remember that this is VERY inefficient and should only ever be used for debugging.

efro.debug.getobjs(cls: type | str, contains: str | None = None, expanded: bool = False) → list[Any]

Return all garbage-collected objects matching criteria.

‘type’ can be an actual type or a string in which case objects whose types contain that string will be returned.

If ‘contains’ is provided, objects will be filtered to those containing that in their str() representations.

efro.debug.getrefs(obj: Any) → list[Any]

Given an object, return things referencing it.

efro.debug.printfiles(file: TextIO | None = None) → None

Print info about open files in the current app.

efro.debug.printrefs(obj: Any, max_level: int = 2, exclude_objs: list[Any] | None = None, expand_ids: list[int] | None = None, file: TextIO | None = None) → None

Print human readable list of objects referring to an object.

‘max_level’ specifies how many levels of recursion are printed. ‘exclude_objs’ can be a list of exact objects to skip if found in the

referrers list. This can be useful to avoid printing the local context where the object was passed in from (locals(), etc).

‘expand_ids’ can be a list of object ids; if that particular object is

found, it will always be expanded even if max_level has been reached.

efro.debug.printsizes(limit: int = 50, file: TextIO | None = None, expanded: bool = False) → None

Print total allocated sizes of different types.

efro.debug.printtypes(limit: int = 50, file: TextIO | None = None, expanded: bool = False) → None

Print a human readable list of which types have the most instances.

efro.error module

Common errors and related functionality.

exception efro.error.AuthenticationError

Bases: Exception

Authentication has failed for some operation.

This can be raised if server-side-verification does not match client-supplied credentials, if an invalid password is supplied for a sign-in attempt, etc.

exception efro.error.CleanError

Bases: Exception

An error that can be presented to the user as a simple message.

These errors should be completely self-explanatory, to the point where a traceback or other context would not be useful.

A CleanError with no message can be used to inform a script to fail without printing any message.

This should generally be limited to errors that will always be presented to the user (such as those in high level tool code). Exceptions that may be caught and handled by other code should use more descriptive exception types.

pretty_print(flush: bool = True, prefix: str = 'Error', file: Any = None, clr: type[ClrBase] | None = None) → None

Print the error to stdout, using red colored output if available.

If the error has an empty message, prints nothing (not even a newline).

exception efro.error.CommunicationError

Bases: Exception

A communication related error has occurred.

This covers anything network-related going wrong in the sending of data or receiving of a response. Basically anything that is out of our control should get lumped in here. This error does not imply that data was not received on the other end; only that a full acknowledgement round trip was not completed.

These errors should be gracefully handled whenever possible, as occasional network issues are unavoidable.

exception efro.error.IntegrityError

Bases: ValueError

Data has been tampered with or corrupted in some form.

exception efro.error.RemoteError(msg: str, peer_desc: str)

Bases: Exception

An error occurred on the other end of some connection.

This occurs when communication succeeds but another type of error occurs remotely. The error string can consist of a remote stack trace or a simple message depending on the context.

Communication systems should aim to communicate specific errors gracefully as standard message responses when specific details are needed; this is intended more as a catch-all.

efro.error.is_asyncio_streams_communication_error(exc: BaseException) → bool

Should this streams error be considered a communication error?

This should be passed an exception which resulted from creating and using asyncio streams. It should return True for any errors that could conceivably arise due to unavailable/poor network connections, firewall/connectivity issues, etc. These issues can often be safely ignored or presented to the user as general ‘connection-lost’ events.

efro.error.is_requests_communication_error(exc: BaseException) → bool

Is the provided exception a communication-related error from requests?

efro.error.is_udp_communication_error(exc: BaseException) → bool

Should this udp-related exception be considered a communication error?

This should be passed an exception which resulted from creating and using a socket.SOCK_DGRAM type socket. It should return True for any errors that could conceivably arise due to unavailable/poor network conditions, firewall/connectivity issues, etc. These issues can often be safely ignored or presented to the user as general ‘network-unavailable’ states.

efro.error.is_urllib_communication_error(exc: BaseException, url: str | None) → bool

Is the provided exception from urllib a communication-related error?

Url, if provided, can provide extra context for when to treat an error as such an error.

This should be passed an exception which resulted from opening or reading a urllib Request. It returns True for any errors that could conceivably arise due to unavailable/poor network connections, firewall/connectivity issues, or other issues out of our control. These errors can often be safely ignored or presented to the user as general ‘network-unavailable’ states.

efro.log module

Logging functionality.

class efro.log.FileLogEcho(original: TextIO, name: str, handler: LogHandler)

Bases: object

A file-like object for forwarding stdout/stderr to a LogHandler.

flush() → None

Flush the file.

isatty() → bool

Are we a terminal?

write(output: Any) → None

Override standard write call.

class efro.log.LogArchive(log_size: ~typing.Annotated[int, <efro.dataclassio.IOAttrs object at 0x7fc67d0bbe90>], start_index: ~typing.Annotated[int, <efro.dataclassio.IOAttrs object at 0x7fc67dd523f0>], entries: ~typing.Annotated[list[~efro.log.LogEntry], <efro.dataclassio.IOAttrs object at 0x7fc6848682f0>])

Bases: object

Info and data for a log.

entries: IOAttrs object at 0x7fc684868b30>]

log_size: IOAttrs object at 0x7fc67dcb5370>]

start_index: IOAttrs object at 0x7fc684869880>]

class efro.log.LogEntry(name: ~typing.Annotated[str, <efro.dataclassio.IOAttrs object at 0x7fc67ddd1040>], message: ~typing.Annotated[str, <efro.dataclassio.IOAttrs object at 0x7fc67ddd1130>], level: ~typing.Annotated[~efro.log.LogLevel, <efro.dataclassio.IOAttrs object at 0x7fc684849cd0>], time: ~typing.Annotated[~datetime.datetime, <efro.dataclassio.IOAttrs object at 0x7fc684b8e420>], labels: ~typing.Annotated[dict[str, str], <efro.dataclassio.IOAttrs object at 0x7fc68484a990>] = <factory>)

Bases: object

Single logged message.

labels: IOAttrs object at 0x7fc6846f5730>]

level: IOAttrs object at 0x7fc68484acc0>]

message: IOAttrs object at 0x7fc67dddb860>]

name: IOAttrs object at 0x7fc67ddd8050>]

time: IOAttrs object at 0x7fc67df882c0>]

class efro.log.LogHandler(*, path: str | Path | None, echofile: TextIO | None, suppress_non_root_debug: bool, cache_size_limit: int, cache_time_limit: datetime.timedelta | None)

Bases: Handler

Fancy-pants handler for logging output.

Writes logs to disk in structured json format and echoes them to stdout/stderr with pretty colors.

add_callback(call: Callable[[LogEntry], None], feed_existing_logs: bool = False) → None

Add a callback to be run for each LogEntry.

Note that this callback will always run in a background thread. Passing True for feed_existing_logs will cause all cached logs in the handler to be fed to the callback (still in the background thread though).

call_in_thread(call: Callable[[], Any]) → None

Submit a call to be run in the logging background thread.

emit(record: LogRecord) → None

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

file_flush(name: str) → None

Send raw stdout/stderr flush to the logger to be collated.

file_write(name: str, output: str) → None

Send raw stdout/stderr output to the logger to be collated.

get_cached(start_index: int = 0, max_entries: int | None = None) → LogArchive

Build and return an archive of cached log entries.

This will only include entries that have been processed by the background thread, so may not include just-submitted logs or entries for partially written stdout/stderr lines. Entries from the range [start_index:start_index+max_entries] which are still present in the cache will be returned.

shutdown() → None

Block until all pending logs/prints are done.

class efro.log.LogLevel(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Severity level for a log entry.

These enums have numeric values so they can be compared in severity. Note that these values are not currently interchangeable with the logging.ERROR, logging.DEBUG, etc. values.

CRITICAL = 4

DEBUG = 0

ERROR = 3

INFO = 1

WARNING = 2

classmethod from_python_logging_level(levelno: int) → LogLevel

Given a Python logging level, return a LogLevel.

property python_logging_level: int

Give the corresponding logging level.

efro.log.setup_logging(log_path: str | Path | None, level: LogLevel, *, suppress_non_root_debug: bool = False, log_stdout_stderr: bool = False, echo_to_stderr: bool = True, cache_size_limit: int = 0, cache_time_limit: datetime.timedelta | None = None) → LogHandler

Set up our logging environment.

Returns the custom handler which can be used to fetch information about logs that have passed through it. (worst log-levels, caches, etc.).

efro.rpc module

Remote procedure call related functionality.

class efro.rpc.RPCEndpoint(handle_raw_message_call: Callable[[bytes], Awaitable[bytes]], reader: asyncio.StreamReader, writer: asyncio.StreamWriter, label: str, *, debug_print: bool = False, debug_print_io: bool = False, debug_print_call: Callable[[str], None] | None = None, keepalive_interval: float = 10.73, keepalive_timeout: float = 30.0)

Bases: object

Facilitates asynchronous multiplexed remote procedure calls.

Be aware that, while multiple calls can be in flight in either direction simultaneously, packets are still sent serially in a single stream. So excessively long messages/responses will delay all other communication. If/when this becomes an issue we can look into breaking up long messages into multiple packets.

DEFAULT_KEEPALIVE_INTERVAL = 10.73

DEFAULT_KEEPALIVE_TIMEOUT = 30.0

DEFAULT_MESSAGE_TIMEOUT = 60.0

close() → None

I said seagulls; mmmm; stop it now.

is_closing() → bool

Have we begun the process of closing?

async run() → None

Run the endpoint until the connection is lost or closed.

Handles closing the provided reader/writer on close.

send_message(message: bytes, timeout: float | None = None, close_on_error: bool = True) → Awaitable[bytes]

Send a message to the peer and return a response.

If timeout is not provided, the default will be used. Raises a CommunicationError if the round trip is not completed for any reason.

By default, the entire endpoint will go down in the case of errors. This allows messages to be treated as ‘reliable’ with respect to a given endpoint. Pass close_on_error=False to override this for a particular message.

test_suppress_keepalives: bool = False

async wait_closed() → None

I said seagulls; mmmm; stop it now.

Wait for the endpoint to finish closing. This is called by run() so generally does not need to be explicitly called.

efro.rpc.ssl_stream_writer_force_close_check(writer: StreamWriter) → None

Ensure a writer is closed; hacky workaround for odd hang.

efro.rpc.ssl_stream_writer_underlying_transport_info(writer: StreamWriter) → str

For debugging SSL Stream connections; returns raw transport info.

efro.terminal module

Functionality related to terminal IO.

efro.terminal.Clr

alias of ClrNever

class efro.terminal.ClrAlways

Bases: ClrBase

Convenience class for color terminal output.

This version has colors always enabled. Generally you should use Clr which points to the correct enabled/disabled class depending on the environment.

BBLK: ClassVar[str] = '\x1b[40m'

BBLU: ClassVar[str] = '\x1b[44m'

BCYN: ClassVar[str] = '\x1b[46m'

BGRN: ClassVar[str] = '\x1b[42m'

BLD: ClassVar[str] = '\x1b[1m'

BLK: ClassVar[str] = '\x1b[30m'

BLU: ClassVar[str] = '\x1b[34m'

BMAG: ClassVar[str] = '\x1b[45m'

BRED: ClassVar[str] = '\x1b[41m'

BWHT: ClassVar[str] = '\x1b[47m'

BYLW: ClassVar[str] = '\x1b[43m'

CYN: ClassVar[str] = '\x1b[36m'

GRN: ClassVar[str] = '\x1b[32m'

INV: ClassVar[str] = '\x1b[7m'

MAG: ClassVar[str] = '\x1b[35m'

RED: ClassVar[str] = '\x1b[31m'

RST: ClassVar[str] = '\x1b[0m'

SBBLK: ClassVar[str] = '\x1b[100m'

SBBLU: ClassVar[str] = '\x1b[104m'

SBCYN: ClassVar[str] = '\x1b[106m'

SBGRN: ClassVar[str] = '\x1b[102m'

SBLK: ClassVar[str] = '\x1b[90m'

SBLU: ClassVar[str] = '\x1b[94m'

SBMAG: ClassVar[str] = '\x1b[105m'

SBRED: ClassVar[str] = '\x1b[101m'

SBWHT: ClassVar[str] = '\x1b[107m'

SBYLW: ClassVar[str] = '\x1b[103m'

SCYN: ClassVar[str] = '\x1b[96m'

SGRN: ClassVar[str] = '\x1b[92m'

SMAG: ClassVar[str] = '\x1b[95m'

SRED: ClassVar[str] = '\x1b[91m'

SWHT: ClassVar[str] = '\x1b[97m'

SYLW: ClassVar[str] = '\x1b[93m'

UND: ClassVar[str] = '\x1b[4m'

WHT: ClassVar[str] = '\x1b[37m'

YLW: ClassVar[str] = '\x1b[33m'

color_enabled = True

class efro.terminal.ClrBase

Bases: object

Base class for color convenience class.

BBLK: ClassVar[str]

BBLU: ClassVar[str]

BCYN: ClassVar[str]

BGRN: ClassVar[str]

BLD: ClassVar[str]

BLK: ClassVar[str]

BLU: ClassVar[str]

BMAG: ClassVar[str]

BRED: ClassVar[str]

BWHT: ClassVar[str]

BYLW: ClassVar[str]

CYN: ClassVar[str]

GRN: ClassVar[str]

INV: ClassVar[str]

MAG: ClassVar[str]

RED: ClassVar[str]

RST: ClassVar[str]

SBBLK: ClassVar[str]

SBBLU: ClassVar[str]

SBCYN: ClassVar[str]

SBGRN: ClassVar[str]

SBLK: ClassVar[str]

SBLU: ClassVar[str]

SBMAG: ClassVar[str]

SBRED: ClassVar[str]

SBWHT: ClassVar[str]

SBYLW: ClassVar[str]

SCYN: ClassVar[str]

SGRN: ClassVar[str]

SMAG: ClassVar[str]

SRED: ClassVar[str]

SWHT: ClassVar[str]

SYLW: ClassVar[str]

UND: ClassVar[str]

WHT: ClassVar[str]

YLW: ClassVar[str]

class efro.terminal.ClrNever

Bases: ClrBase

Convenience class for color terminal output.

This version has colors disabled. Generally you should use Clr which points to the correct enabled/disabled class depending on the environment.

BBLK: ClassVar[str] = ''

BBLU: ClassVar[str] = ''

BCYN: ClassVar[str] = ''

BGRN: ClassVar[str] = ''

BLD: ClassVar[str] = ''

BLK: ClassVar[str] = ''

BLU: ClassVar[str] = ''

BMAG: ClassVar[str] = ''

BRED: ClassVar[str] = ''

BWHT: ClassVar[str] = ''

BYLW: ClassVar[str] = ''

CYN: ClassVar[str] = ''

GRN: ClassVar[str] = ''

INV: ClassVar[str] = ''

MAG: ClassVar[str] = ''

RED: ClassVar[str] = ''

RST: ClassVar[str] = ''

SBBLK: ClassVar[str] = ''

SBBLU: ClassVar[str] = ''

SBCYN: ClassVar[str] = ''

SBGRN: ClassVar[str] = ''

SBLK: ClassVar[str] = ''

SBLU: ClassVar[str] = ''

SBMAG: ClassVar[str] = ''

SBRED: ClassVar[str] = ''

SBWHT: ClassVar[str] = ''

SBYLW: ClassVar[str] = ''

SCYN: ClassVar[str] = ''

SGRN: ClassVar[str] = ''

SMAG: ClassVar[str] = ''

SRED: ClassVar[str] = ''

SWHT: ClassVar[str] = ''

SYLW: ClassVar[str] = ''

UND: ClassVar[str] = ''

WHT: ClassVar[str] = ''

YLW: ClassVar[str] = ''

color_enabled = False

class efro.terminal.TerminalColor(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Color codes for printing to terminals.

Generally the Clr class should be used when incorporating color into terminal output, as it handles non-color-supporting terminals/etc.

BG_BLACK = '\x1b[40m'

BG_BLUE = '\x1b[44m'

BG_CYAN = '\x1b[46m'

BG_GREEN = '\x1b[42m'

BG_MAGENTA = '\x1b[45m'

BG_RED = '\x1b[41m'

BG_WHITE = '\x1b[47m'

BG_YELLOW = '\x1b[43m'

BLACK = '\x1b[30m'

BLUE = '\x1b[34m'

BOLD = '\x1b[1m'

CYAN = '\x1b[36m'

GREEN = '\x1b[32m'

INVERSE = '\x1b[7m'

MAGENTA = '\x1b[35m'

RED = '\x1b[31m'

RESET = '\x1b[0m'

STRONG_BG_BLACK = '\x1b[100m'

STRONG_BG_BLUE = '\x1b[104m'

STRONG_BG_CYAN = '\x1b[106m'

STRONG_BG_GREEN = '\x1b[102m'

STRONG_BG_MAGENTA = '\x1b[105m'

STRONG_BG_RED = '\x1b[101m'

STRONG_BG_WHITE = '\x1b[107m'

STRONG_BG_YELLOW = '\x1b[103m'

STRONG_BLACK = '\x1b[90m'

STRONG_BLUE = '\x1b[94m'

STRONG_CYAN = '\x1b[96m'

STRONG_GREEN = '\x1b[92m'

STRONG_MAGENTA = '\x1b[95m'

STRONG_RED = '\x1b[91m'

STRONG_WHITE = '\x1b[97m'

STRONG_YELLOW = '\x1b[93m'

UNDERLINE = '\x1b[4m'

WHITE = '\x1b[37m'

YELLOW = '\x1b[33m'

efro.util module

Small handy bits of functionality.

class efro.util.DirtyBit(dirty: bool = False, retry_interval: float = 5.0, *, use_lock: bool = False, auto_dirty_seconds: float | None = None, min_update_interval: float | None = None)

Bases: object

Manages whether a thing is dirty and regulates attempts to clean it.

To use, simply set the ‘dirty’ value on this object to True when some action is needed, and then check the ‘should_update’ value to regulate when attempts to clean it should be made. Set ‘dirty’ back to False after a successful update. If ‘use_lock’ is True, an asyncio Lock will be created and incorporated into update attempts to prevent simultaneous updates (should_update will only return True when the lock is unlocked). Note that It is up to the user to lock/unlock the lock during the actual update attempt. If a value is passed for ‘auto_dirty_seconds’, the dirtybit will flip itself back to dirty after being clean for the given amount of time. ‘min_update_interval’ can be used to enforce a minimum update interval even when updates are successful (retry_interval only applies when updates fail)

property dirty: bool

Whether the target is currently dirty.

This should be set to False once an update is successful.

property should_update: bool

Whether an attempt should be made to clean the target now.

Always returns False if the target is not dirty. Takes into account the amount of time passed since the target was marked dirty or since should_update last returned True.

class efro.util.DispatchMethodWrapper

Bases: Generic[ArgT, RetT]

Type-aware standin for the dispatch func returned by dispatchmethod.

static register(func: Callable[[Any, Any], RetT]) → Callable[[Any, Any], RetT]

Register a new dispatch handler for this dispatch-method.

registry: dict[Any, Callable]

class efro.util.ValueDispatcher(call: Callable[[ValT], RetT])

Bases: Generic[ValT, RetT]

Used by the valuedispatch decorator

register(value: ValT) → Callable[[Callable[[], RetT]], Callable[[], RetT]]

Add a handler to the dispatcher.

class efro.util.ValueDispatcher1Arg(call: Callable[[ValT, ArgT], RetT])

Bases: Generic[ValT, ArgT, RetT]

Used by the valuedispatch1arg decorator

register(value: ValT) → Callable[[Callable[[ArgT], RetT]], Callable[[ArgT], RetT]]

Add a handler to the dispatcher.

efro.util.ago_str(timeval: datetime, maxparts: int = 1, now: datetime | None = None, decimals: int = 0) → str

Given a datetime, return a clean human readable ‘ago’ str.

Note that this is hard-coded in English so should not be used for visible in-game elements; only tools/etc.

If now is not passed, efro.util.utc_now() is used.

efro.util.assert_non_optional(obj: T | None) → T

Return an object with Optional typing removed.

Assert is used to check its actual type, so only use this when failures are not expected. Use check_non_optional otherwise.

efro.util.asserttype(obj: Any, typ: type[T]) → T

Return an object typed as a given type.

Assert is used to check its actual type, so only use this when failures are not expected. Otherwise use checktype.

efro.util.asserttype_o(obj: Any, typ: type[T]) → T | None

Return an object typed as a given optional type.

Assert is used to check its actual type, so only use this when failures are not expected. Otherwise use checktype.

efro.util.caller_source_location() → str

Returns source file name and line of the code calling us.

Example: ‘mymodule.py:23’

efro.util.check_non_optional(obj: T | None) → T

Return an object with Optional typing removed.

Always checks the actual type and throws a TypeError on failure. Use assert_non_optional for a more efficient (but less safe) equivalent.

efro.util.check_utc(value: datetime) → None

Ensure a datetime value is timezone-aware utc.

efro.util.checktype(obj: Any, typ: type[T]) → T

Return an object typed as a given type.

Always checks the type at runtime with isinstance and throws a TypeError on failure. Use asserttype for more efficient (but less safe) equivalent.

efro.util.checktype_o(obj: Any, typ: type[T]) → T | None

Return an object typed as a given optional type.

Always checks the type at runtime with isinstance and throws a TypeError on failure. Use asserttype for more efficient (but less safe) equivalent.

efro.util.compact_id(num: int) → str

Given a positive int, return a compact string representation for it.

Handy for visualizing unique numeric ids using as few as possible chars. This version is more compact than human_readable_compact_id() but less friendly to humans due to using both capital and lowercase letters, both ‘O’ and ‘0’, etc.

Therefore for n chars this can store values of 62^n.

Sort order for these ids is the same as the original numbers.

efro.util.data_size_str(bytecount: int, compact: bool = False) → str

Given a size in bytes, returns a short human readable string.

In compact mode this should be 6 or fewer chars for most all sane file sizes.

efro.util.dispatchmethod(func: Callable[[Any, ArgT], RetT]) → DispatchMethodWrapper[ArgT, RetT]

A variation of functools.singledispatch for methods.

Note: as of Python 3.9 there is now functools.singledispatchmethod, but it currently (as of Jan 2021) is not type-aware (at least in mypy), which gives us a reason to keep this one around for now.

efro.util.empty_weakref(objtype: type[T]) → ReferenceType[T]

Return an invalidated weak-reference for the specified type.

efro.util.explicit_bool(val: bool) → bool

Return a non-inferable boolean value.

Useful to be able to disable blocks of code without type checkers complaining/etc.

efro.util.extract_arg(args: list[str], name: str, required: Literal[False] = False) → str | None

efro.util.extract_arg(args: list[str], name: str, required: Literal[True]) → str

Given a list of args and an arg name, returns a value.

The arg flag and value are removed from the arg list. raises CleanErrors on any problems.

efro.util.extract_flag(args: list[str], name: str) → bool

Given a list of args and a flag name, returns whether it is present.

The arg flag, if present, is removed from the arg list.

efro.util.float_hash_from_string(s: str) → float

Given a string value, returns a float between 0 and 1.

If consistent across processes. Can be useful for assigning db ids shard values for efficient parallel processing.

efro.util.human_readable_compact_id(num: int) → str

Given a positive int, return a compact string representation for it.

Handy for visualizing unique numeric ids using as few as possible chars. This representation uses only lowercase letters and numbers (minus the following letters for readability):

‘s’ is excluded due to similarity to ‘5’. ‘l’ is excluded due to similarity to ‘1’. ‘i’ is excluded due to similarity to ‘1’. ‘o’ is excluded due to similarity to ‘0’. ‘z’ is excluded due to similarity to ‘2’.

Therefore for n chars this can store values of 21^n.

When reading human input consisting of these IDs, it may be desirable to map the disallowed chars to their corresponding allowed ones (‘o’ -> ‘0’, etc).

Sort order for these ids is the same as the original numbers.

If more compactness is desired at the expense of readability, use compact_id() instead.

efro.util.linearstep(edge0: float, edge1: float, x: float) → float

A linear transition function.

Returns a value that linearly moves from 0 to 1 as we go between edges. Values outside of the range return 0 or 1.

efro.util.make_hash(obj: Any) → int

Makes a hash from a dictionary, list, tuple or set to any level, that contains only other hashable types (including any lists, tuples, sets, and dictionaries).

Note that this uses Python’s hash() function internally so collisions/etc. may be more common than with fancy cryptographic hashes.

Also be aware that Python’s hash() output varies across processes, so this should only be used for values that will remain in a single process.

efro.util.set_canonical_module_names(module_globals: dict[str, Any]) → None

Do the thing.

efro.util.smoothstep(edge0: float, edge1: float, x: float) → float

A smooth transition function.

Returns a value that smoothly moves from 0 to 1 as we go between edges. Values outside of the range return 0 or 1.

efro.util.snake_case_to_camel_case(val: str) → str

Given a snake-case string ‘foo_bar’, returns camel-case ‘FooBar’.

efro.util.snake_case_to_title(val: str) → str

Given a snake-case string ‘foo_bar’, returns ‘Foo Bar’.

efro.util.split_list(input_list: list[T], max_length: int) → list[list[T]]

Split a single list into smaller lists.

efro.util.timedelta_str(timeval: timedelta | float, maxparts: int = 2, decimals: int = 0) → str

Return a simple human readable time string for a length of time.

Time can be given as a timedelta or a float representing seconds. Example output:

“23d 1h 2m 32s” (with maxparts == 4) “23d 1h” (with maxparts == 2) “23d 1.08h” (with maxparts == 2 and decimals == 2)

Note that this is hard-coded in English and probably not especially performant.

efro.util.unchanging_hostname() → str

Return an unchanging name for the local device.

Similar to the hostname call (or os.uname().nodename in Python) except attempts to give a name that doesn’t change depending on network conditions. (A Mac will tend to go from Foo to Foo.local, Foo.lan etc. throughout its various adventures)

efro.util.utc_now() → datetime

Get timezone-aware current utc time.

Just a shortcut for datetime.datetime.now(datetime.UTC). Avoid datetime.datetime.utcnow() which is deprecated and gives naive times.

efro.util.utc_now_naive() → datetime

Get naive utc time.

This can be used to replace datetime.utcnow(), which is now deprecated. Most all code should migrate to use timezone-aware times instead of this.

efro.util.utc_this_hour() → datetime

Get offset-aware beginning of the current hour in the utc time zone.

efro.util.utc_this_minute() → datetime

Get offset-aware beginning of current minute in the utc time zone.

efro.util.utc_today() → datetime

Get offset-aware midnight in the utc time zone.

efro.util.valuedispatch(call: Callable[[ValT], RetT]) → ValueDispatcher[ValT, RetT]

Decorator for functions to allow dispatching based on a value.

This differs from functools.singledispatch in that it dispatches based on the value of an argument, not based on its type. The ‘register’ method of a value-dispatch function can be used to assign new functions to handle particular values. Unhandled values wind up in the original dispatch function.

efro.util.valuedispatch1arg(call: Callable[[ValT, ArgT], RetT]) → ValueDispatcher1Arg[ValT, ArgT, RetT]

Like valuedispatch but for functions taking an extra argument.

efro.util.valuedispatchmethod(call: Callable[[SelfT, ValT], RetT]) → ValueDispatcherMethod[ValT, RetT]

Like valuedispatch but works with methods instead of functions.

efro.util.warntype(obj: Any, typ: type[T]) → T

Return an object typed as a given type.

Always checks the type at runtime and simply logs a warning if it is not what is expected.

efro.util.warntype_o(obj: Any, typ: type[T]) → T | None

Return an object typed as a given type.

Always checks the type at runtime and simply logs a warning if it is not what is expected.

Module contents

Common bits of functionality shared between all efro projects.

Things in here should be hardened, highly type-safe, and well-covered by unit tests since they are widely used in live client and server code.