What is Eva?

If you have ever written a simple Bash script, you probably noticed how easy it looks at first. It feels almost like typing in a terminal. One command after another, with some control flow. The code is simple and direct.

Then you start to automate some tasks with scripts, so you do not have to type them every time. Little by little the code becomes more complex, and at some point you start to ask yourself: is Bash enough for what I need, or should I choose another language?

When a Bash script has hundreds of lines, full of defensive checks and calls to external commands, some of them installed only for that script, it is no longer a small automation script. That is usually the moment to move to another language.

Python, like Bash, is available on almost every Linux distribution, but it also has a very strong standard library. High portability, batteries included, clean syntax.

For this reason I decided to write this scripting toolbox for Linux systems: a collection of opinionated functions and classes, used as a layer of syntactic sugar.

Everything in a single file. Zero dependencies. The Python standard library is enough.

What is M?

Before explaining what M is, it is important to say why it is documented at all. In fact, it is not meant to be a public feature. But this variable is so special, and so widely used inside Eva’s source code, that it needs to be explained.

Its purpose is very simple: behind this variable there is a dynamic module loader, so modules can be accessed as a hierarchical namespace. Let us see an example. Suppose we want to use the time module. In Python you write import time at the start of the code and then, for example, time.sleep(1) anywhere. With M, you just write M.time.sleep(1), and that is all.

Many people will ask… why? The answer is also simple: because it feels more comfortable and cleaner to write code this way. The global namespace is not filled with names that may collide with others, and modules are loaded only when they are needed.

Do not use this in your own code. Normal module imports are good and reliable.

• constants
  • DAY, HOUR, MINUTE, SECOND, WEEK
  • LoggerTimestamp: COMPACT, DATE, FULL, HUMAN, ISO, SHORT, TIME
  • LoggerType: ALERT, DEBUG, ERROR, FATAL, INFO, OKAY
• functions
  • append_file, read_binary_file, read_text_file, write_file
  • check_internet, check_internet_ipv4, check_internet_ipv6
  • clamp
  • dir_is_empty
  • get_interfaces
  • get_timestamp
  • ip_in_network
  • is_ip, is_ipv4, is_ipv6
  • is_pid
  • noop
  • normalize_float, normalize_integer
  • normalize_ip
  • normalize_path
  • normalize_text
  • path_is_dir, path_is_file, path_is_link
  • ping_ipv4, ping_ipv6
  • printe, printo
  • randomize_ip
  • resolve_dns, resolve_dns_ipv4, resolve_dns_ipv6
  • run
  • user_is_admin
• classes
  • Chrono: delta, expired, precision, remaining, reset, timeout
  • Latch: path, pid, socket, status, off, on
  • Logger: mute, path, stderr, timestamp, type
  • Reader: bye, off, on, poll, status, wait
  • Terminal: brush, clear_line, clear_screen, clear_scrollback, disable_alt_screen, disable_alt_scroll, disable_cursor, enable_alt_screen, enable_alt_scroll, enable_cursor, height, move_by, move_home, move_to, mute, reset_terminal, restore_cursor, save_cursor, stderr, stream, width, write
  • Spinner: bye, colors, current, hide, glyphs, message, off, on, status, stderr, tempo

DAY = 86400.0

HOUR = 3600.0

MINUTE = 60.0

SECOND = 1.0

WEEK = 604800.0

They represent units of time expressed in seconds. Using them makes the code clearer and avoids magic numbers.

They are float because the second is the smallest time unit used in this project. Smaller values can be expressed as fractions of a second.

import eva, time

print('hello world!')

time.sleep(eva.SECOND / 2)

print('goodbye!')

LoggerTimestamp.COMPACT

LoggerTimestamp.DATE

LoggerTimestamp.FULL

LoggerTimestamp.HUMAN

LoggerTimestamp.ISO

LoggerTimestamp.SHORT

LoggerTimestamp.TIME

This enum defines the valid values for the Logger.timestamp property.

Each enum value matches a property of the object returned by the get_timestamp function.

Not all properties of that object are included in the enum. Only the most useful ones are defined here.

import eva

info = eva.Info(timestamp=eva.LoggerTimestamp.SHORT)
info('hello world!')

alert = eva.Alert()
alert.timestamp = eva.LoggerTimestamp.TIME
alert('goodbye!')

LoggerType.ALERT

LoggerType.DEBUG

LoggerType.ERROR

LoggerType.FATAL

LoggerType.INFO

LoggerType.OKAY

This enum defines the valid types used to create a Logger instance.

Each type has its own fixed name and color to make it easy to recognize in the terminal.

Each type also has a default output. For ALERT, INFO and OKAY the output is standard output. For DEBUG, ERROR and FATAL the output is standard error.

The output of a Logger instance can be changed, but its name and color are fixed.

import eva

logger = eva.Logger(eva.LoggerType.OKAY)

print(logger.stderr)
print(logger.type)

logger.stderr = not logger.stderr

print(logger.stderr)
print(logger.type)

append_file(path, content, /, *, eol=False, lock=False)

read_binary_file(path, /, *, lock=False)

read_text_file(path, /, *, lock=False)

write_file(path, content, /, *, exclusive=False, lock=False)

They allow reading from and writing to files.

The path parameter is the path of the file to read or write.

The append_file and write_file functions are used to write data. Both create the file if it does not exist. The content value must be bytes or str.

The append_file function adds new content to the file, with a newline at the end if the eol parameter is True.

The write_file function fully replaces the previous content. Setting the exclusive parameter to True prevents writing if the file already exists.

The read_binary_file and read_text_file functions are used to read data. The first returns the content as a bytes object. The second returns it as str and assumes UTF-8 encoding with replace as the error handler.

In multithreaded or multiprocess programs, concurrent reads or writes on the same file may corrupt its contents. To avoid this, the lock parameter can be set to True, which applies an operating-system-level lock (fcntl.flock) before the read or write operation.

No exceptions are raised if an error happens during a read or write. Instead, the functions return False as a warning. On success, read functions return the content and write functions return True.

import eva

path = '/tmp/file-does-not-exist.txt'

print(eva.read_text_file(path))

path = '/tmp/file.txt'

eva.write_file(path, b'hello ')

eva.append_file(path, 'world!', eol=True)

print(eva.read_text_file(path))

check_internet(*, ipv4=True, ipv6=True, retries=None, timeout=None)

check_internet_ipv4(*, retries=None, timeout=None)

check_internet_ipv6(*, retries=None, timeout=None)

They check if there is Internet connectivity.

To do this, they send ping requests to well-known DNS servers. These servers are stored as a fixed list of IP addresses and cannot be changed. The list includes both IPv4 and IPv6 addresses. The order is chosen at random on each run to avoid bias.

The ipv4 and ipv6 parameters determine which protocols are tested. Setting a parameter to True enables the protocol, while False skips it.

The retries parameter sets the maximum number of ping attempts per protocol before connectivity is considered missing. The default is one attempt. Each attempt uses a different IP address. The IP list is walked in order until a ping reply is received. If none is received, it starts again from the first IP until all attempts are used.

The timeout parameter sets the maximum time, in seconds, to wait for each ping reply. The default is one second. See the documentation of ping_ipv4 and ping_ipv6 for more details about this parameter.

Values such as zero retries, a zero second timeout, or both protocols disabled are allowed.

The check_internet_ipv4 and check_internet_ipv6 functions are simple wrappers with the protocol parameters fixed.

True is returned only if at least one ping is successful for each enabled protocol. Otherwise the check fails and False is returned. It also fails if both protocols are disabled or if the number of retries is zero.

import eva

both = eva.check_internet()
ipv4 = eva.check_internet_ipv4(retries=3)
ipv6 = eva.check_internet_ipv6(timeout=eva.SECOND / 2)

print(both, ipv4, ipv6)

clamp(value, /, minimum, maximum)

It limits value to the range defined by minimum and maximum, making sure it does not go outside these bounds.

In most cases value, minimum, and maximum are numbers, usually float or int, but this is not required. They only need to be comparable with each other.

A ValueError exception is raised if minimum is greater than maximum.

If the value is already inside the range, it is returned as is. Otherwise, the nearest limit is returned.

import eva, ipaddress, math

print(eva.clamp(1, minimum=12, maximum=123))

print(eva.clamp('foo', minimum='bar', maximum='quux'))

print(eva.clamp(math.inf, minimum=1234, maximum=math.inf))

address = ipaddress.ip_address('192.0.2.123')
minimum = ipaddress.ip_address('192.0.2.20')
maximum = ipaddress.ip_address('192.0.2.30')

print(eva.clamp(address, minimum, maximum))

print(eva.clamp(str(address), str(minimum), str(maximum)))

dir_is_empty(path, /)

It checks whether the path parameter is a directory (or a symbolic link to one) and is empty.

It returns True if the path is a directory and it could be confirmed to be empty. It returns False in all other cases (not empty, not a directory, does not exist, or could not be checked).

import eva

path = '/tmp/dir-does-not-exist'

print(eva.dir_is_empty(path))

path = '/tmp/dir-exists-and-is-empty'

print(eva.dir_is_empty(path))

get_interfaces(*, down=True, loopback=True, up=True)

It gets the network interfaces that are currently available on the system.

By default, all interfaces are returned, but some can be excluded using the parameters. Setting down to False excludes inactive interfaces, loopback excludes the loopback interface, and up excludes active ones.

It is allowed to exclude both active and inactive interfaces at the same time.

It returns a tuple with the interface names. The tuple may be empty.

import eva

interfaces = eva.get_interfaces(loopback=False)

for name in interfaces:
    print('name: ', name)

print('total: ', len(interfaces))

get_timestamp(epoch=None, /, *, utc=False)

It returns an object with several formatted representations of a given moment in time.

The epoch parameter specifies the moment to format. It must be a float representing the number of seconds since Unix Time, so microseconds can be included. If this parameter is not provided, the current system time is used.

The utc parameter is a boolean. When it is True, all representations are formatted using Coordinated Universal Time. By default, the system local time zone is used.

The returned value is a SimpleNamespace object. Its properties contain the formatted representations. Except for epoch, which is a float with seconds and microseconds, all values are str based on the format "%Y/%m/%d-%H:%M:%S.%f%z". The full property is the complete timestamp in that format, human is the same but without microseconds, compact uses a two-digit year and removes microseconds and all date and time separators, short also uses a two-digit year but only removes the time zone and the microseconds, iso is the extended ISO 8601 format with microseconds, date contains only the date part, and time contains only the time part. The other properties are the individual components (year, month, day, hour, minute, second, microsecond, and timezone).

import eva

now = eva.get_timestamp()

for key, value in vars(now).items():
    print(f'{key}: "{value}"')

print()

utc = eva.get_timestamp(now.epoch, utc=True)

for key, value in vars(utc).items():
    print(f'{key}: "{value}"')

ip_in_network(address, network, /)

It checks whether an IP address is inside a network prefix.

The address parameter is the IP address to check. It can also be given as a CIDR prefix, in which case the host address is extracted from it.

The network parameter is the network to check against. It should be given as a CIDR prefix. If it is just an IP address, then a "/32" prefix is assumed for IPv4 and "/128" for IPv6.

A ValueError exception is raised if the IP or the network range is not valid.

It returns True if the IP belongs to that network, or False otherwise.

import eva

address = '192.0.2.123'
network = '192.0.2.0/24'

print(eva.ip_in_network(address, network))

address, network = network, address

print(eva.ip_in_network(address, network))

is_ip(address, /, *, host=False, ipv4=True, ipv6=True, network=False)

is_ipv4(address, /, *, host=False, network=False)

is_ipv6(address, /, *, host=False, network=False)

They check whether the value of address can be treated as a valid IP address.

It is common to think of an IP address as a series of numbers joined by delimiter characters. These functions apply that rule on purpose. As a result, some forms that may be accepted in other contexts are rejected here. This includes integers, hexadecimal strings, values with leading zeros, IPv6 addresses with embedded interfaces, or IPv4 addresses mapped into IPv6, among others. Any value of this kind, or similar ones, is considered invalid.

If the host parameter is True, then CIDR prefixes are not allowed.

If the network parameter is True and the value is a CIDR prefix, then it is also checked that it strictly represents a network address (all host bits set to zero).

The ipv4 and ipv6 parameters enable the protocols to test when they are True. The value must be valid for at least one of them. It is allowed to disable both protocols.

The is_ipv4 and is_ipv6 functions are simple wrappers with the protocol parameters fixed.

It returns True if the value is accepted as a valid IP, or False otherwise.

import eva

print(eva.is_ip('192.0.2.123'))
print(eva.is_ip('192.0.2.123/32'))
print(eva.is_ip('192.0.2.0/24', network=True))
print(eva.is_ip('192.0.2.123', host=True, network=True))
print(eva.is_ip('2001:db8::1234'))
print(eva.is_ipv6('2001:db8::1234'))

print(eva.is_ip('192.0.2.123/32', host=True))
print(eva.is_ip('192.0.2.123/24', network=True))
print(eva.is_ipv4('2001:db8::1234'))

is_pid(pid, /, *, check=False)

It checks whether the value of pid can be treated as a valid process ID.

If the check parameter is True, it also checks that the process exists at that moment.

It returns True if the value is considered a valid PID, or False otherwise.

import eva, os, random

pids = (
0,
1,
os.getpid(),
random.randint(1, 1_000_000),
10**100
)

for pid in pids:
    print(eva.is_pid(pid))

print()

for pid in pids:
    print(eva.is_pid(pid, check=True))

noop(*args, **kwargs)

This is a typical "no operation" function. It accepts any kind of arguments, does nothing, and returns nothing (technically it returns None, like any function without an explicit return).

import eva, threading

eva.noop()
eva.noop('foo')
eva.noop('foo', bar='quux')

args = ('foo', 'bar', 'quux')
thread = threading.Thread(args=args, target=eva.noop)
thread.start()

normalize_float(number, /, *, maximum=None, minimum=None, padding=False, precision=None)

normalize_integer(number, /, *, maximum=None, minimum=None)

These two functions normalize the value of number. For normalize_float the value is expected to be a float, and for normalize_integer an int, but this is not strict because the value is converted to the target type as part of the normalization.

The maximum and minimum parameters can be used to keep the normalized value within these limits. One, both, or neither can be set to leave the value unbounded. These parameters are expected to be of the type to which the value is being normalized.

If the padding parameter is set to True, the returned value is a str padded with as many zeros in the decimal part as precision specifies.

The precision parameter controls rounding of the decimal to the specified number of digits. The function uses the "half-up" rounding method when possible, if that fails, it falls back to the built-in round function, with a maximum of 15 decimal digits.

A ValueError exception is raised if normalization fails. For example, if any parameter is NaN, or if minimum is greater than maximum.

The function returns the normalized value based on the given parameters.

import eva

print(eva.normalize_float(-0.0))
print(eva.normalize_float(123.0, padding=True, precision=3))
print(eva.normalize_integer(12, minimum=123, maximum=1234))

number = 123.456789
print(eva.normalize_float(number, minimum=1234.5))
print(eva.normalize_float(number, maximum=12.0))
print(eva.normalize_float(number, precision=1))

number = 1234.56789
print(eva.normalize_integer(number))
print(eva.normalize_integer(number, minimum=1234))

normalize_ip(address, /, *, exploded=False, host=False, network=False, upper=False)

It normalizes an IP address.

The address parameter is the address to normalize. It can be IPv4 or IPv6, with or without a CIDR prefix.

If the exploded parameter is set to True, the long form is used. Otherwise, the compressed form is used.

If the host parameter is set to True, the CIDR prefix is removed. Otherwise, if a prefix is present, it is kept.

If the network parameter is set to True, and address was given with a CIDR prefix, it is normalized as a network address (all host bits set to zero).

If the upper parameter is set to True, the result is returned in upper case. Otherwise, it is in lower case.

The exploded and upper parameters only make sense for IPv6. They have no effect on IPv4.

A ValueError exception is raised if the IP in address is considered invalid.

The function returns the normalized value based on the given parameters.

import eva

address = '192.0.2.123/24'
print(eva.normalize_ip(address))
print(eva.normalize_ip(address, host=True))
print(eva.normalize_ip(address, network=True))

for address in ('192.0.2.123', '2001:db8::1234'):
    print(eva.normalize_ip(address, exploded=True, upper=True))

normalize_path(*args, absolute=False, leading=False, resolve=False, trailing=False)

It normalizes a path by resolving "." and ".." and by removing redundant separators.

The positional parameters are the parts of the path to normalize. They are expected to be str, path-like objects, or even bytes (in that case they are converted to str using surrogateescape as the error handler). At least one is required. Only the first one can be an absolute path. The rest, if any, are treated as relative paths to join. None of these values can be empty or contain a null value.

If the absolute parameter is set to True, the result is normalized as an absolute path.

If the leading parameter is set to True, a path that starts with "//" is reduced to a single separator.

If the resolve parameter is set to True, symbolic links are resolved. This also forces the path to become absolute, even if absolute is False.

If the trailing parameter is set to True, a directory separator is always added at the end. Otherwise, it is always removed.

A ValueError exception is raised if the path is considered invalid for any reason.

The function returns the normalized path as a str based on the given parameters.

import eva, os

print(eva.normalize_path('foo//bar//'))

print(eva.normalize_path('/foo', 'bar', '.', 'QUUX', '..'))

os.chdir('/tmp')
print(eva.normalize_path('foo/bar', absolute=True))

print(eva.normalize_path('/bin', resolve=True))

print(eva.normalize_path('//foo/bar', leading=True))

print(eva.normalize_path('/is-a-dir', trailing=True))

normalize_text(text, /, *, full=False, printable=None, reduce=None, strip=None, uniform=None)

This function normalizes the text parameter into a str, with a set of optional actions that can also be applied. These actions can be enabled or disabled using their own parameters. text is expected to be a str, but this is not strict because the value is converted as part of the normalization.

The full parameter enables all actions set to None when True, or disables them when False. A value of None indicates that the action is disabled by default and depends on this parameter. When an action is set explicitly, its own value overrides full. Thus, full acts as a global switch for actions that are not explicitly defined.

If the printable parameter is set to True, only printable characters are allowed in the text. The characters \t and \n are also treated as printable. All other characters are replaced with the replacement character \uFFFD.

If the reduce parameter is set to True, any character treated as whitespace is reduced to a single one when it appears in a repeated sequence.

If the strip parameter is set to True, any character treated as whitespace is removed from the start and the end of the text.

If the uniform parameter is set to True, all characters treated as whitespace are converted to the normal space character (\x20).

The actions are applied in this order: uniform, reduce, strip and printable.

The function returns the normalized text as a str based on the given parameters.

import eva

text = ' Foo \t bar\n\nquux\x00. '

print(eva.normalize_text(text))

print(eva.normalize_text(text, full=True))

print(eva.normalize_text(text, full=True, strip=False))

path_is_dir(path, /, *, follow)

path_is_file(path, /, *, follow)

path_is_link(path, /, *, follow)

They try to check whether path is a directory, a regular file, or a symbolic link.

The follow parameter decides whether symbolic links in the path should be followed before checking its type. When it is True, links are followed. When it is False, the path is checked as it is, without following links.

The follow parameter is basically useless in path_is_link when it is set to True (since it is assumed that at the end of a chain of symbolic links there is something that is not a link). At most, it can be used to detect whether path is a closed loop of symbolic links.

They return True if the type of the path could be confirmed, or False otherwise or if an error occurs.

import eva

print(eva.path_is_dir('/bin', follow=False))

print(eva.path_is_dir('/bin', follow=True))

print(eva.path_is_link('/bin', follow=False))

print(eva.path_is_file('/bin/ls', follow=False))

ping_ipv4(address, /, *, retries=None, timeout=None)

ping_ipv6(address, /, *, retries=None, timeout=None)

They send a ping request to the remote host given in address.

It is important to note that the request is done by calling the Linux ping command, with a careful set of arguments to make it safe and reliable. For example, address is not used directly to build the command. It is first resolved and validated to get the IP used in the call.

The retries parameter sets the maximum number of attempts before giving up. The default is one attempt.

The timeout parameter sets the maximum time, in seconds, to wait for each attempt. The default is one second. The ping command may not accept very large values, so this parameter is deliberately limited to one hour.

Values such as zero retries or a zero second timeout are allowed. An infinite timeout is also allowed, because the ping command supports it.

The function returns True if the request succeeds, or False otherwise. It also returns False if it fails because the ping command is not available or did not run correctly (it is present on most Linux systems and supports the same parameters).

import eva

print(eva.ping_ipv4('microsoft.com', retries=2))
print(eva.ping_ipv4('google.com', timeout=eva.SECOND / 2))

print(eva.ping_ipv6('google.com'))
print(eva.ping_ipv6('::1'))

printe(*args, **kwargs)

printo(*args, **kwargs)

They are wrappers around the built-in print function, with the special feature that they use Spinner.hide and Terminal.brush to cooperate and stay in sync with other parts of the terminal.

Except that their file parameter is fixed and cannot be changed (sys.__stderr__ for printe and sys.__stdout__ for printo), all other features are the same as the built-in function, including parameters, behavior and return value.

import eva

eva.printo('hello ', end='', flush=True)
eva.printe('world!')

with eva.Spinner.hide():
    with eva.Terminal.brush:
        print('goodbye!')

randomize_ip(address, /, *, network=False, seed=None)

It picks, in an arbitrary way, one of the possible IPs from the given network address.

It is important to note that, if the CIDR prefix length allows it, the network address and the broadcast address are removed from the range for IPv4, and the "Subnet Router anycast" address is removed for IPv6.

The address parameter is the network address, with a CIDR prefix, from which the IP is taken. Any value accepted by is_ip is allowed. This means an IPv4 or IPv6 address, with or without a CIDR prefix. It is allowed for the host bits not to be all zero, since only the network bits are relevant.

If the network parameter is set to True, the returned value includes both the IP and the CIDR prefix. Otherwise, only the IP is returned.

If the seed parameter is given, the choice is not random but deterministic based on this seed. In this case, it must be of type bytes or str.

It is allowed to pass an address without a CIDR prefix, or one with a "/32" prefix for IPv4 or "/128" for IPv6. In all these cases, the same IP given as input is always returned.

See normalize_ip to know which exceptions may be raised.

The function returns an arbitrary IP address, with or without a CIDR prefix.

import eva

address = '192.0.2.123/24'
print(eva.randomize_ip(address))
print(eva.randomize_ip(address, network=True))

address = '2001:db8::/32'
print(eva.randomize_ip(address, seed='foo'))
print(eva.randomize_ip(address))
print(eva.randomize_ip(address, seed=b'foo'))

resolve_dns(domain, /, *, ipv4=True, ipv6=True, shuffle=False)

resolve_dns_ipv4(domain, /, *, shuffle=False, zen=True)

resolve_dns_ipv6(domain, /, *, shuffle=False, zen=True)

They resolve the domain given in domain to get its IP addresses.

The ipv4 and ipv6 parameters choose which protocols are used. Set them to True to enable a protocol, or to False to skip it.

If the shuffle parameter is set to True, the IPs that are found are returned in a random order, to help avoid bias.

The zen parameter is a boolean that turns zen mode on or off. When it is enabled, the return value is simplified to a more minimal form. When it is disabled, the return value contains more detail but is also more complex to use.

It is allowed to disable both protocols or to pass an IP address in domain.

The resolve_dns_ipv4 and resolve_dns_ipv6 functions are simple wrappers with the protocol parameters fixed.

The function returns a tuple with the resolved IPs, or, if zen mode is enabled, a str containing the first IP. If domain cannot be resolved, it returns an empty tuple, or False in zen mode. The result can be evaluated in a boolean context, because an empty tuple is falsy, whereas a str containing an IP is truthy.

import eva

domain = 'yahoo.com'

for address in eva.resolve_dns(domain):
    print(address)

print()

print(eva.resolve_dns_ipv4(domain))
print(eva.resolve_dns_ipv6(domain))

run(command, /, *, binary=False, cwd=None, stdin=None, timeout=None, zen=True)

This function runs a system command. It is basically a wrapper around subprocess.run, used as syntactic sugar.

The command parameter specifies the command to run. It can be provided as given, as the path to the executable (absolute or relative), with or without arguments.

If the binary parameter is set to True, the captured standard output and error are handled as bytes. Otherwise they are returned as str, using UTF-8 encoding and replace as the error handler.

The cwd parameter specifies the working directory, normalized as an absolute path, in which the command is run. If it is not provided, the current directory is used.

The stdin parameter specifies the standard input for the command. It can be either bytes or str.

The timeout parameter specifies a maximum time limit in seconds before the command is stopped. If it is not provided, or if its value is infinite, there is no limit.

The zen parameter is a boolean that turns zen mode on or off. When it is enabled, the return value is simplified to a more minimal form. When it is disabled, the return value contains more detail but is also more complex to use.

The return value is a SimpleNamespace object with these fields: code the command exit code, exception which is None if no exception was raised or the exception if one was raised, stderr and stdout with the captured output according to the binary parameter, and success which is True if the execution finished correctly or False otherwise. In zen mode, the return value is just the same as success. To be precise, success is False if, for any reason, the command was not run and finished correctly, or if code is not zero. Otherwise it is True.

import eva

print(eva.run('echo foobar'))
print(eva.run('false'))

print()

result = eva.run('tr a X', stdin='foobar', zen=False)
for key, value in vars(result).items():
    print(f'{key}: "{value}"')

print()

result = eva.run('sleep 2', timeout=eva.SECOND, zen=False)
for key, value in vars(result).items():
    print(f'{key}: "{value}"')

user_is_admin()

Checks whether the user executing the current process has administrator privileges.

Returns True if the user has administrator privileges, or False otherwise.

import eva

if eva.user_is_admin():
    print('I am the root user')
else:
    print('try running as root')

Chrono(*, precision=False, timeout=False)

This class can be used as a stopwatch or as a timer. It provides access to the elapsed time or the remaining time until a limit is reached. It works with a monotonic counter, which makes it immune to changes in the system clock. The class uses a time resolution in seconds of type float, where the decimal part represents fractions of this base unit.

The stopwatch starts at the moment of instantiation, but it can be reset later when needed, which allows the instance to be reused.

The precision and timeout parameters allow setting these properties at instantiation time.

delta RO

It provides the elapsed time according to precision. This is the base property for using the instance as a stopwatch.

expired RO

This is a boolean that tells whether the time limit set by timeout has been reached. precision is taken into account for the calculation. It is always False if timeout is not defined.

precision RW

It specifies the rounding precision used when reading delta and remaining, as well as the one used to evaluate expired. It must be an int to indicate the number of decimal places, or False to disable this feature.

remaining RO

It provides the remaining time, according to precision, until the timeout limit is reached. Its value is zero once the limit has been reached or passed, and it is always infinite if the limit is not defined (the infinite value simplifies calculations with other float values and also serves as a sentinel).

timeout RW

It specifies the time limit on which expired and remaining depend. Setting False disables this feature. This is the base property for using the instance as a timer.

reset()

Resets the initial reference time used for the calculations.

import eva, time

chrono = eva.Chrono(precision=3, timeout=eva.SECOND / 2)

while not chrono.expired:
    time.sleep(eva.SECOND / 10)
    print(chrono.delta, chrono.remaining)

Latch(path, /, *, auto=False, pid=None, socket=False)

This class implements an inter-process locking mechanism based on a given resource, so that only one instance can acquire it (the first one that succeeds) and then claim ownership of it.

Its most common use is to prevent two or more processes of the same daemon from running at the same time. Otherwise it would be redundant and even harmful. The other processes, since they cannot acquire the resource, understand that they have no right to run their role.

Typically this mechanism is based on an operating-system lock on a given file: the lock is acquired, kept for as long as needed, and then released. This class implements exactly this idea by using a lock of type fcntl.flock.

However, although this kind of lock is very common and usually enough, this class also supports an optional alternative lock based on a TCP socket. When it is enabled, it works as follows: the file path is no longer the exact resource that is locked, but instead a deterministic IPv4 address and port in the loopback network are computed from it, and the code tries to claim that socket. No file is touched, and the path does not even need to exist. It literally acts as a seed to select the IP and port in a "random" way.

The range used is 127.0.0.0/8. From this range, the network address (127.0.0.0), localhost (127.0.0.1) and the broadcast address (127.255.255.255) are deliberately avoided. For the port, privileged ones (below 1024) are avoided.

The path parameter specifies the resource on which the lock is based. It is exactly the path to the file to be locked. This path is canonicalized as an absolute path with no symbolic links.

The auto parameter is a boolean that tells whether the lock should be attempted automatically at instantiation time.

The pid parameter specifies the PID number that is written into the file once the lock is acquired. If it is omitted, the PID of the current process is used. This number must be valid, otherwise a ValueError is raised. There does not need to be a running process with that PID until the exact moment when the lock is attempted. This parameter is irrelevant for socket-based locking.

If the socket parameter is set to True, the instance uses socket-based locking.

Although the class has the on and off methods to acquire and release the lock manually, it is recommended to avoid them and use the class as a context manager in a with statement instead.

path RO

The path to the file used as the locking resource that was passed at instantiation.

pid RO

The PID number that was passed at instantiation. If the lock is file-based and is successfully acquired, this PID is written into the file.

socket RO

A boolean that tells whether the selected lock is socket-based.

status RO

A boolean that tells whether the instance has acquired the lock. It is False both when it failed and when it has not been tried yet.

off()

Releases the lock. This method is idempotent and raises no exceptions.

on()

Tries to acquire the lock. This method is idempotent and raises no exceptions. If the lock is file-based, the value of pid must match a process that really exists at that moment, otherwise the lock is not attempted. It returns True if the lock was acquired, or False otherwise.

path = '/tmp/foobar.pid'

with eva.Latch(path) as l1:
    with eva.Latch(path) as l2:
        with eva.Latch(path, socket=True) as l3:
            with eva.Latch(path, socket=True) as l4:
                print(l1.status, l2.status, l3.status, l4.status)

Logger(kind, /, *, mute=False, path=False, stderr=None, timestamp=False)

This class manages the emission of log messages. It can show them on the terminal and optionally write them to a file. The class is designed for multithreaded environments, so writes to the terminal and to the file are locked to avoid corruption.

Each instance has a set of properties that define how the message is emitted.

The most important property of an instance is its type. Each type has a fixed color and a default destination, which can be either standard output or standard error.

The emitted message is shown on the terminal prefixed with a timestamp and the type. This output is properly formatted with colors, indentation, alignment, message normalization and multiline wrapping according to the terminal width. If the destination is not interactive, it is not formatted and is sent as plain text.

The kind parameter sets the type of message the instance identifies with. It must be one of the values defined in LoggerType.

The mute, path, stderr and timestamp parameters allow setting these properties at instantiation time.

Direct instantiation of this class is not required. Each type has its own subclass: Alert, Debug, Error, Fatal, Info and Okay. These subclasses inherit from the main class and fix the kind argument. Nothing else.

To emit a message, the instance must be called as a function, since it implements the __call__ dunder. The message to emit is simply its first parameter. Since the message is a str, it can be formatted by interpolating the other parameters (if any) using f-strings. If there are no parameters other than the message, it is emitted as is. It is allowed to omit the message and call the instance with no parameters, in which case an empty message is emitted. The return value of each call is a boolean that tells whether the message could be written to the file set by path, or True if path is disabled.

mute RW

A boolean that tells whether emitted messages should be shown on the terminal. Even when it is set to True, messages are still written to the file according to path.

path RW

The path to the file where emitted messages are written. This path is normalized and converted to an absolute path. Messages are written as plain text, prefixed by the LoggerTimestamp.ISO timestamp and the instance type. This property must be set to False to disable this feature.

stderr RW

A boolean that tells whether emitted messages should go to standard error instead of standard output. Each LoggerType already has a default destination, but it can be changed with this property.

timestamp RW

The timestamp format that should be shown on the terminal when a message is emitted. It accepts any value from LoggerTimestamp. This property must be set to False to disable this feature.

type RO

The type of the instance. It can be any value from LoggerType. It cannot be changed after the object has been created.

import eva

logger = eva.Logger(eva.LoggerType.INFO)
logger('hello world!')

info = eva.Info()
info('hello {}!', 'world')

error = eva.Error(path='/tmp/error.log')
error('something went wrong')

okay = eva.Okay(timestamp=eva.LoggerTimestamp.TIME)
okay('everything is alright')

debug = eva.Debug(path='/tmp/debug.log')
debug.mute = True
debug('my type is {type}', type=debug.type)

Reader()

This class allows capturing standard input by reading it in the background. It uses a secondary thread to monitor it and distribute the data to the active instances.

The secondary thread mechanism avoids blocking the main thread forever. Instead, there is at most a short polling wait before trying again, which lets the thread breathe and handle other tasks. In fact, if standard input handling is delegated to another thread, this polling wait does not even affect the main thread.

The secondary thread acts as an engine. It only starts when at least one instance becomes active, and it keeps running until the last active instance is deactivated. When it starts, and if the terminal is interactive, the terminal input mode is set to cbreak. The previous terminal mode is first saved so it can be restored when the engine stops.

While the engine is running, it periodically checks whether there is new data on standard input. If so, the data is distributed by appending it to the buffer of each active instance. Each buffer is a FIFO queue, and consuming it produces a captured chunk, not the entire buffer so far. Each chunk can be at least 1 byte and at most 1024 bytes.

Although the class has the on and off methods to manually enable and disable reading, it is recommended to avoid them and use the class as a context manager in a with statement instead.

status RO

A boolean that shows the state of the instance. It is True when it is registered as active, or False otherwise.

wait RW

A class property (not an instance property) that sets the maximum wait time in seconds for each poll. Its default value is 0.25 seconds.

bye()

A class method (not an instance method) that deactivates all active instances. It literally calls the off method of each instance.

off()

Deactivates the instance. From that point on, its buffer is cleared and the instance no longer receives what the engine captures. The engine is stopped if this instance was the last one still active. This method is idempotent and raises no exceptions.

on()

Activates the instance. From that point on, whatever the engine captures is appended to its buffer. The engine is started if this instance is the first one to become active. This method is idempotent and raises no exceptions.

poll()

Removes and returns one chunk from its buffer. Since it is a FIFO queue, this chunk is the oldest one, not the newest. The chunk is a bytes object and is returned immediately. If the buffer is empty or the instance is not active, it waits for at most the number of seconds set by wait. When this time limit is reached, the return value is None to signal that there is still nothing captured. This method can also return False to signal that standard input has been fully consumed, so the instance can be deactivated. This happens when standard input is not an interactive terminal.

import eva

with eva.Reader() as reader:
    print('press any key... (ESC to exit)')

    while True:
        chunk = reader.poll()

        if chunk in (False, b'\x1B'):
            print('goodbye!')
            break
        elif chunk is not None:
            print(chunk)

Terminal(*, mute=False, stderr=False)

This class provides an abstraction over the terminal as syntactic sugar. It simplifies common actions that may be needed when working with a terminal (mainly ANSI sequences), such as clearing the screen or part of it, moving the cursor, hiding it, and so on. The class tries to be agnostic to whether the output is really a terminal or not.

It is safe for multi-threaded environments. This is done by acquiring a lock in the instances. The one that owns it has the right to manipulate the terminal until it releases it.

Before writing to the terminal, this class uses Spinner.hide to temporarily silence any spinner that may be active.

The mute and stderr parameters allow setting these properties at instantiation time.

brush RO

This is a class property (not an instance property) that publicly exposes the threading.RLock object used to synchronize writing to the terminal. Each instance gets the right to write to it by acquiring this lock. Classes or functions such as Logger, printe, printo, or Spinner use it directly or indirectly to synchronize. So, when using them, there is no need to use this lock manually. In fact, in general it is better to avoid it.

height RO

width RO

These properties return the height and width of the terminal, in units commonly known as "rows" and "columns". No exception is raised if these values cannot be obtained. In that case, very low default values such as 1 are returned instead. In such a case, these values should be seen as sentinel values.

mute RW

A boolean that, when set to True, makes any write from the instance to the terminal be ignored. It mainly affects the write method and, by extension, any other method that sends ANSI sequences to the terminal. It is therefore a way to mute the instance.

stderr RW

A boolean that, when set to True, makes the terminal target be standard error, or False to make it standard output.

stream RO

Returns the target stream of the instance. It is literally sys.__stderr__ if its stderr property is True, or sys.__stdout__ otherwise.

clear_line(*, end=True, start=True)

clear_screen(*, end=True, start=True)

They allow clearing the current line, the screen, or parts of them. ANSI sequences are used for this. If the start parameter is True, the section from the beginning to the cursor is cleared. If the end parameter is True, the section from the cursor to the end is cleared. The corresponding section is not cleared if its parameter is False.

clear_scrollback()

Clears the scrollback history. ANSI sequences are used for this.

disable_alt_screen()

enable_alt_screen()

Enables or disables the alternate screen of the terminal. ANSI sequences are used for this.

disable_alt_scroll()

enable_alt_scroll()

Enables or disables the alternate scroll mode of the terminal. ANSI sequences are used for this.

disable_cursor()

enable_cursor()

Enables or disables cursor visibility. ANSI sequences are used for this.

move_by(*, x=None, y=None)

move_home()

move_to(*, x=None, y=None)

They move the cursor position. ANSI sequences are used for this. The x and y parameters are the horizontal and vertical axes where to place the cursor. If they are 0, the cursor does not move on that axis. The move_by method moves the cursor relative to its current position, so its parameters can be negative. The move_to method moves the cursor to an absolute position in the terminal, so it does not accept negative values. The move_home method moves the cursor to its home position (there is an ANSI sequence for this, but move_to(x=1, y=1) also achieves something similar).

reset_terminal()

Resets the terminal, restoring its default state. ANSI sequences are used for this.

restore_cursor()

save_cursor()

They save and restore the cursor position using ANSI sequences. The save_cursor function saves the current cursor position. The cursor can then be moved back to that saved position using restore_cursor.

write(content, /, *, flush=True)

Writes to the terminal. The content parameter sets the data to write and must be a str. The flush parameter is a boolean that, when True, makes the write immediate.

import eva

term = eva.Terminal()
term.save_cursor()
term.enable_alt_screen()
term.clear_screen()
term.disable_cursor()

term.move_home()
term.write('foo')
term.move_to(y=2)
term.write('bar')
term.move_by(x=-3, y=1)
term.write('quux')

term.write('\nhello world!\n')

with eva.Terminal.brush:
    print('goodbye!')

term.enable_cursor()
input('(press ENTER to exit)')
term.clear_screen()
term.disable_alt_screen()
term.restore_cursor()

Spinner(*, auto=False, colors=None, glyphs=None, message=None, tempo=None)

This class manages spinner-style animations. During long-running tasks, these animations are useful to show visual feedback in the terminal.

A background thread is used to manage active spinners. This avoids blocking the main thread and lets it handle other tasks.

The background thread acts as an engine. It only starts when one of the instances is activated, and it keeps running until the last active instance is deactivated.

While the engine is running, the animation of the last activated spinner is shown in the terminal. There is therefore a LIFO stack of active instances, and the most recent one is the one shown. This means several spinners can be active at the same time, but only the top one is visible. The engine handles this.

Each instance can define its own look and behavior.

The animation hides the cursor and, for each frame, shows one colored character and a message on the current cursor line. To avoid corrupting the terminal, this animation is coordinated with other classes or functions such as Logger, printe, printo, or Terminal.

Although the class has the on and off methods to activate and deactivate spinners manually, it is recommended to avoid them and use the class as a context manager in a with statement instead.

The auto parameter is a boolean that tells whether the spinner should be activated automatically at instantiation time.

The colors, glyphs, message and tempo parameters allow setting these properties at instantiation time.

colors RW

This is a tuple with the animation colors. It can be edited by setting a list or a tuple. An empty tuple is allowed. Each element is an 8-bit ANSI color as an int, so there are up to 256 possible colors, from 0 to 255. Each frame uses one of them to color the glyph.

current RO

This is a class property (not an instance property) that holds the priority instance, that is, the spinner that is currently being shown. Its value is False if no spinner is active. Because an instance is truthy, this property can be used as a boolean to know whether any spinner is being shown.

glyphs RW

This is a tuple with the animation glyphs. It can be edited by setting a list or a tuple. It can also be set to a str, which is then split into individual characters. An empty tuple is allowed. Each element is a single character. Each frame uses one of them to animate the spinner.

message RW

Sets the message that goes with the animation. It is usually a str, but it can also be a function. In that case, the function is called on each frame and its return value is used as the message. In all cases, an empty string is allowed.

status RO

A boolean that shows the state of the instance. It is True when it is registered as active, or False otherwise.

stderr RW

This is a class property (not an instance property) that sets the output target of the animation. It accepts True (its default value) to send the animation to standard error. It is False for standard output. The animation is only shown if the target is an interactive terminal.

tempo RW

Defines the speed of the animation. It sets the interval in seconds between each frame. Its default value is 0.25 seconds.

bye()

A class method (not an instance method) that deactivates all active instances. It literally calls the off method of each instance.

hide()

This is a class method (not an instance method) that temporarily hides spinner animations. It is used as a context manager with the with statement. Classes or functions such as Logger, printe, printo, or Terminal already use it to coordinate. So when using them, there is no need to call this method manually. In general, it is better to avoid using it directly.

off()

Deactivates the instance. The animation engine is stopped if this instance was the last one still active. This method is idempotent and raises no exceptions.

on()

Activates the instance. The engine is started if this instance is the first one to be registered as active. This method is idempotent and raises no exceptions. In practice, calling this method on a spinner that is already active makes it the priority one. That is, it is moved in the LIFO stack of active spinners and becomes the visible one if it was not.

import eva, random, time

nested1 = eva.Spinner(message='my turn', tempo=eva.SECOND / 10)

nested2 = eva.Spinner()
nested2.colors = (0, 1, 15)
nested2.message = lambda: random.randint(0, 9)

something_that_takes_time = lambda: time.sleep(eva.SECOND * 3)

with eva.Spinner(glyphs='◳◲◱◰'):
    something_that_takes_time()

    with nested1:
        something_that_takes_time()

        with nested2:
            something_that_takes_time()
            eva.printo('hello world!')

        something_that_takes_time()

    with eva.Spinner.hide():
        print('goodbye!')

    something_that_takes_time()