Query functions

A Jupyter notebook would be able to import and use SQLAlchemy functions freely, but that would require users to understand a lot about the internals of the aggregation system’s data structures. It would effectively mean that the tables and models that we’ve created become part of the public API, and any changes to them may mean incrementing the major version number and documenting changes for end-users.

Instead, let’s create some functions that return DataPoint records for users to interact with. This way, only the DataPoint objects and the function signatures are part of the API that we must maintain for people. We can always add more functions later, as we discover additional requirements.

To begin with, the most important feature that we need is the ability to find data records, ordered by the time they were collected. This lets users write some analysis code to analyze the values of the sensors over time. We may also want to filter this by the sensor type, the deployment identifier, and a date range.

We have to decide what form we want the function to have. Should it return a list or tuple of objects or an iterator? A tuple would allow us to easily count the number of items we retrieved and to iterate over the list multiple times. On the other hand, an iterator would allow us to minimize RAM use, which may help us support much larger data sets, but restricts us to only being able to iterate over the data once. We’ll create iterator functions, as they allow for more efficient code. The iterators can be converted to tuples by the calling code, so our users are able to choose to iterate over a tuple if they prefer.

This function acts as a (synchronous) context manager, setting up a database connection and an associated session and both returning that session and setting it as the value of the db_session_var context variable before entering the body of the associated with block. It also unsets this session, commits any changes, and closes the session when the context manager exits. This ensures that there are no lingering locks in the database, that data is persisted, and that if functions that use the db_session_var variable can only be used inside the body of this context manager.

We also need to write the function that returns DataPoint objects for the user to analyze. Eventually, we’ll have to deal with performance issues due to processing large amounts of data, but the first code you write to solve a problem should not be optimized, a naïve implementation is both easier to understand and more likely not to suffer from being too clever. We’ll look at some techniques for optimization in the next chapter.

This function gets the session from the context variable set up by with_database(...), builds a query object, and then runs that object’s all method using an executor, giving way to other tasks while the all method runs. Iterating over the query object rather than calling query.all() would cause database operations to be triggered as the loop runs, so we must be careful to only set up the query in asynchronous code and delegate the all() function call to the executor. The result of this is a list of SQLAlchemy’s lightweight result named tuples in the rows variable, which we can then iterate over yielding the matching DataPoint object.

As rows variable contains a list of all the result objects, we know that all the data has been processed by the database and parsed SQLAlchemy in the executor before execution passes back to our get_data() function . This means that we’re using all the RAM needed to store the full results set before the first DataPoint object is available to the end-user. Storing all this data when we don’t know that we need all of it is a little memory and time inefficient, but elaborate methods to paginate the data in the iterator would be an example of premature optimization. Don’t change this from the naïve approach until it becomes a problem.

Merely counting the data points isn’t an interesting way of analyzing the data. We can start trying to make sense of the data by plotting values on a scatter plots. Let’s start with a simple sanity check, plotting the value of the RelativeHumidity sensor against date (Listing 9-5). This is a good one to start with, as the stored data is a floating-point number rather than a dictionary-based structure, so we don’t need to parse the values.

The matplotlib library is perhaps the most popular plotting library in Python. Its plot_date(...) function is a great fit for plotting a series of values against time. It takes a list of values for the x axis and a corresponding list of values for the y axis, as well as the style to be used when plotting a point³ and a flag to set which axis contains the date values. Our get_data(...) function doesn’t return what we need for the x and y parameters directly, it returns an async iterator of data point objects.

Filtering data

It would be nice to filter the data in the query stage, rather than just discarding all sensor data that doesn’t meet our criteria when we are iterating through them. Right now, every piece of data is selected, a result object is created, then a DataPoint object, and only then are irrelevant entries skipped. To this end, we can add an additional parameter to the get_data(...) method that determines if a filter on sensor_data will be applied to the generated query.

async def get_data(sensor_name: t.Optional[str] = None) -> t.AsyncIterator[DataPoint]:

    db_session = db_session_var.get()

    loop = asyncio.get_running_loop()

    query = db_session.query(datapoint_table)

    if sensor_name:

        query = query.filter(datapoint_table.c.sensor_name == sensor_name)

    query = query.order_by(datapoint_table.c.collected_at)

This approach saves a lot of overhead, as it means that only the relevant sensor data points are passed to the end-user, but also it’s a more natural interface. Users expect to be able to specify what data they want, not to get absolutely all data and manually filter it. The version of the function in Listing 9-6 takes less than one second to execute with my sample data set (compared to over 3 seconds for the previous version) but shows the same chart.

from apd.aggregation.query import with_database, get_data

from matplotlib import pyplot as plt

async def plot():

    points = [(dp.collected_at, dp.data) async for dp in get_data(sensor_name="RelativeHumidity")]

    x, y = zip(*points)

    plt.plot_date(x, y, "o", xdate=True)

with with_database("postgresql+psycopg2://apd@localhost/apd") as session:

    await plot()

plt.show()

Listing 9-6

Delegating filtering to the get_data function

This plotting function is short and not overly complex; it represents quite a natural interface for loading data from the database. The downside is that having multiple deployments mixed in together results in unclear charts, where there are multiple data points for a given time. Matplotlib supports calling plot_date(...) multiple times with different logical result sets, which are then displayed using different colors. Our users can achieve this by creating multiple point lists as they iterate over the results of the get_data(...) call, as shown in Listing 9-7.

import collections

from apd.aggregation.query import with_database, get_data

from matplotlib import pyplot as plt

async def plot():

    legends = collections.defaultdict(list)

    async for dp in get_data(sensor_name="RelativeHumidity"):

        legends[dp.deployment_id].append((dp.collected_at, dp.data))

    for deployment_id, points in legends.items():

        x, y = zip(*points)

        plt.plot_date(x, y, "o", xdate=True)

with with_database("postgresql+psycopg2://apd@localhost/apd") as session:

    await plot()

plt.show()

Listing 9-7

Plotting all sensor deployments independently

This one again makes the interface unnatural; it would be more logical for end-users to iterate over deployments and then iterate over sensor data values, rather than iterate over all data points and organize them into lists manually. An alternative would be to create a new function that lists all the deployment ids, then allow get_data(...) to filter by deployment_id . This would allow us to loop over the individual deployments and make a new get_data(...) call to get only that deployment’s data. This is demonstrated in Listing 9-8.

async def get_deployment_ids():

    db_session = db_session_var.get()

    loop = asyncio.get_running_loop()

    query = db_session.query(datapoint_table.c.deployment_id).distinct()

    return [row.deployment_id for row in await loop.run_in_executor(None, query.all)]

async def get_data(

    sensor_name: t.Optional[str] = None,

    deployment_id: t.Optional[UUID] = None,

) -> t.AsyncIterator[DataPoint]:

    db_session = db_session_var.get()

    loop = asyncio.get_running_loop()

    query = db_session.query(datapoint_table)

    if sensor_name:

        query = query.filter(datapoint_table.c.sensor_name == sensor_name)

    if deployment_id:

        query = query.filter(datapoint_table.c.deployment_id == deployment_id)

    query = query.order_by(

        datapoint_table.c.collected_at,

    )

Listing 9-8

Extended data collection functions for deployment_id filtering

This new function can be used to loop over multiple calls to get_data(...), rather than the plot function looping and sorting the resulting data points into independent lists. Listing 9-9 demonstrates a very natural interface to looping over all the deployments for a single sensor, which behaves identically to the previous version.

import collections

from apd.aggregation.query import with_database, get_data, get_deployment_ids

from matplotlib import pyplot as plt

async def plot(deployment_id):

    points = []

    async for dp in get_data(sensor_name="RelativeHumidity", deployment_id=deployment_id):

        points.append((dp.collected_at, dp.data))

    x, y = zip(*points)

    plt.plot_date(x, y, "o", xdate=True)

with with_database("postgresql+psycopg2://apd@localhost/apd") as session:

    deployment_ids = await get_deployment_ids()

    for deployment in deployment_ids:

        await plot(deployment)

plt.show()

Listing 9-9

Plotting all deploymens using the new helper functions

This approach allows the end-user to interrogate each deployment individually, so only the relevant data for a combination of sensor and deployment is loaded into RAM at once. It’s a perfectly appropriate API to offer the end-user.

Multilevel iterators

We previously reworked the interface for filtering by sensor name to do the filtering in the database to avoid iterating over unnecessary data. Our new deployment id filter isn’t used to exclude data we don’t need, it’s used to make it easier to loop over each logical group independently. We don’t need to use a filter here, we’re using one to make the interface more natural.

If you’ve worked with the itertools module in the standard library much, you may have used the groupby(...) function . This takes an iterator and a key function and returns an iterator of iterators, the first being the value of the key function and the second being a run of values that match the given result of the key function. This is the same problem we’ve been trying to solve by listing our deployments and then filtering the database query.

The key function given to groupby(...) is often a simple lambda expression, but it can be any function, such as one of the functions from the operator module. For example, operator.attrgetter("deployment_id") is equivalent to lambda obj: obj.deployment_id, and operator.itemgetter(2) is equivalent to lambda obj: obj[2].

For this example, we’ll define a key function that returns the value of an integer modulo 3 and a data() generator function that yields a fixed series of numbers, printing its status as it goes. This allows us to see clearly when the underlying iterator is advanced.

import itertools

import typing as t

def mod3(n: int) -> int:

    return n % 3

def data() -> t.Iterable[int]:

    for number in [0, 1, 4, 7, 2, 6, 9]:

        print(f"Yielding {number}")

        yield number

We can loop over the contents of the data() generator and print the value of the mod3 function, which lets us see that the first group has one item, then there’s a group of three items, then a group of one, then a group of two.

>>> print([mod3(number) for number in data()])

data() is starting

Yielding 0

Yielding 1

Yielding 4

Yielding 7

Yielding 2

Yielding 6

Yielding 9

data() is complete

[0, 1, 1, 1, 2, 0, 0]

Setting up a groupby does not consume the underlying iterable; each item it generates is processed as the groupby is iterated over. To work correctly, the groupby only needs to decide if the current item is in the same group as the previous one or if a new group has started, it doesn’t analyze the iterable as a whole. Items with the same value for the key function are only grouped together if they are a contiguous block in the input iterator, so it’s common to ensure that the underlying iterator is sorted to avoid splitting groups up.

By creating a groupby over our data with the mod3(...) key function, we can create a two-level loop, first iterating over the values of the key function, then iterating over the values from data() that produce that key value.

>>> for val, group in itertools.groupby(data(), mod3):

...     print(f"Starting new group where mod3(x)=={val}")

...     for number in group:

...         print(f"x=={number} mod3(x)=={mod3(val)}")

...     print(f"Group with mod3(x)=={val} is complete")

...

data() is starting

Yielding 0

Starting new group where mod3(x)==0

x==0 mod3(x)==0

Yielding 1

Group with mod3(x)==0 is complete

Starting new group where mod3(x)==1

x==1 mod3(x)==1

Yielding 4

x==4 mod3(x)==1

Yielding 7

x==7 mod3(x)==1

Yielding 2

Group with mod3(x)==1 is complete

Starting new group where mod3(x)==2

x==2 mod3(x)==2

Yielding 6

Group with mod3(x)==2 is complete

Starting new group where mod3(x)==0

x==6 mod3(x)==0

Yielding 9

x==9 mod3(x)==0

data() is complete

Group with mod3(x)==0 is complete

From the output of the print statements, we can see that the groupby only ever pulls one item at a time, but manages the iterators it provides in such a way that looping over the values is natural. Whenever the inner loop requests a new item, the groupby function requests a new item from the underlying iterator and then decides its behavior based on that value. If the key function reports the same value as the previous item, it yields the new value to the inner loop; otherwise, it signals that the inner loop is complete and holds the value until the next inner loop starts.

The iterators behave just as we’d expect if we had concrete lists of items; there is no requirement to iterate over the inner loop if we don’t need to. If we don’t iterate over the inner loop completely before advancing the outer loop, the groupby object will transparently advance the source iterable as though we had. In the following example, we skip the group of three where mod3(...)==1, and we can see that the underlying iterator is advanced three times by the groupby object:

>>> for val, group in itertools.groupby(data(), mod3):

...     print(f"Starting new group where mod3(x)=={val}")

...     if val == 1:

...         # Skip the ones

...         print("Skipping group")

...         continue

...     for number in group:

...         print(f"x=={number} mod3(x)=={mod3(val)}")

...     print(f"Group with mod3(x)=={val} is complete")

...

data() is starting

Yielding 0

Starting new group where mod3(x)==0

x==0 mod3(x)==0

Yielding 1

Group with mod3(x)==0 is complete

Starting new group where mod3(x)==1

Skipping group

Yielding 4

Yielding 7

Yielding 2

Starting new group where mod3(x)==2

x==2 mod3(x)==2

Yielding 6

Group with mod3(x)==2 is complete

Starting new group where mod3(x)==0

x==6 mod3(x)==0

Yielding 9

x==9 mod3(x)==0

data() is complete

Group with mod3(x)==0 is complete

The behavior is intuitive when we’re using it, but it can be hard to follow how it’s implemented. Figure 9-1 shows a pair of flow charts, one for the outer loop and one for each individual inner loop.

If we had a standard iterator (as opposed to an asynchronous iterator), we could sort the data by deployment_id and use itertools.groupby(...) to simplify our code to handle multiple deployments without needing to query for the individual deployments. Rather than making a new get_data(...) call for each, we could iterate over the groups and handle the internal iterator in the same way we already do, using list comprehensions and zip(...).

Unfortunately, there is no fully asynchronous equivalent of groupby at the time of writing. While we can write a function that returns an async iterator whose values are UUID and async iterator of DataPoint pairs, there is no way of grouping these automatically.

At the risk of writing clever code, we can write an implementation of groupby that works with asynchronous code ourselves using closures. It would expose multiple iterators to the end-user that work on the same underlying iterator, in just the same way as itertools.groupby(...). It would be better to use a library function for this if one were available.

Each time we find a new value of the key function, we need to return a new generator function that maintains a reference to the underlying source iterator. This way, when someone advances an item iterator, it can choose to either yield the data point it receives or to indicate that it’s the end of the item iterator, as the groupby function does. Equally, if we advance the outer iterator before an item iterator has been consumed, it needs to “fast-forward” through the underlying iterator until the start of a new group is found.

The code in Listing 9-10 is a single function that delegates to our get data function and wraps it in the appropriate groupby logic, as opposed to a generic function that can adapt any iterator.

async def get_data_by_deployment(

    *args, **kwargs

) -> t.AsyncIterator[t.Tuple[UUID, t.AsyncIterator[DataPoint]]]:

    """Return an Async Iterator that contains two-item pairs.

    These pairs are a string (deployment_id), and an async iterator that contains

    the datapoints with that deployment_id.

    Usage example:

        async for deployment_id, datapoints in get_data_by_deployment():

            print(deployment_id)

            async for datapoint in datapoints:

                print(datapoint)

            print()

    """

    # Get the data, using the arguments to this function as filters

    data = get_data(*args, **kwargs)

    # The two levels of iterator share the item variable, initialise it # with the first item from the iterator. Also set last_deployment_id

    # to None, so the outer iterator knows to start a new group.

    last_deployment_id: t.Optional[UUID] = None

    try:

        item = await data.__anext__()

    except StopAsyncIteration:

        # There were no items in the underlying query, return immediately

        return

    async def subiterator(group_id: UUID) -> t.AsyncIterator[DataPoint]:

        """Using a closure, create an iterator that yields the current

        item, then yields all items from data while the deployment_id matches

        group_id, leaving the first that doesn't match as item in the enclosing

        scope."""

        # item is from the enclosing scope

        nonlocal item

        while item.deployment_id == group_id:

            # yield items from data while they match the group_id this iterator represents

            yield item

            try:

                # Advance the underlying iterator

                item = await data.__anext__()

            except StopAsyncIteration:

                # The underlying iterator came to an end, so end the subiterator too

                return

    while True:

        while item.deployment_id == last_deployment_id:

            # We are trying to advance the outer iterator while the

            # underlying iterator is still part-way through a group.# Speed through the underlying until we hit an item where

            # the deployment_id is different to the last one (or,

            # is not None, in the case of the start of the iterator)

            try:

                item = await data.__anext__()

            except StopAsyncIteration:

                # We hit the end of the underlying iterator: end this # iterator too

                return

        last_deployment_id = item.deployment_id

        # Instantiate a subiterator for this group

        yield last_deployment_id, subiterator(last_deployment_id)

Listing 9-10

An implementation of get_data_by_deployment that acts like an asynchronous groupby

This uses await data.__anext__() to advance the underlying data iterator, rather than an async for loop, to make the fact that the iterator is consumed in multiple places more obvious.

An implementation of this generator coroutine is in the code for this chapter. I’d encourage you to try adding print statements and breakpoints to it, to help understand the control flow. This code is more complex than most Python code you’ll need to write (and I’d caution you against introducing this level of complexity into production code; having it as a self-contained dependency is better), but if you can understand how it works, you’ll have a thorough grasp on the details of generator functions, asynchronous iterators, and closures. As asynchronous code is used more in production code, libraries to offer this kind of complex manipulation of iterators are sure to become available.

Displaying multiple sensors

We’ve now got three functions that help us connect to the database and iterate over DataPoint objects in a sensible order and with optional filtering. So far we’ve been using the matplotlib.pyplot.plot_dates(...) function to convert pairs of sensor values and dates to a single chart. This is a helper function that makes it easier to generate a plot by making various drawing functions available in a global namespace. It is not the recommended approach when making multiple charts.

We want to be able to loop over each of our sensor types and generate a chart for each. If we were to use the pyplot API, we would be constrained to using a single plot, with the highest values skewing the axes to make the lowest impossible to read. Instead, we want to generate an independent plot for each and show them side by side. For this, we can use the matplotlib.pyplot.figure(...) and figure.add_subplot(...) functions. A subplot is an object which behaves broadly like matplotlib.pyplot but representing a single plot inside a larger grid of plots. For example, figure.add_subplot(3,2,4) would be the fourth plot in a three-row, two-column grid of plots.

Right now, our plot(...) function assumes that the data it is working with is a number, which can be passed directly to matplotlib for display on our chart. Many of our sensors have different data formats though, such as the temperature sensor which has a dictionary of temperature and the unit being used as its value attribute. These different values need to be converted to numbers before they can be plotted.

We can refactor our plotting function out to a utility function in apd.aggregation to vastly simplify our Jupyter notebooks, but we need to ensure that it can be used with other formats of sensor data. Each plot needs to provide some configuration for the sensor to be graphed, a subplot object to draw the plot in, and a mapping from deployment ids to a user-facing name for populating the plot’s legend. It should also accept the same filtering arguments as get_data(...), to allow users to constrain their charts by date or deployment id.

The config data class also needs some string parameters, such as the title of the chart, the axis labels, and the sensor_name that needs to be passed to get_data(...) in order to find the data needed for this chart. Once we have the Config class defined, we can create two config objects that represent the two sensors which use raw floating-point numbers as their value type and a function to return all registered configs.

If we use this new cleaner function to add a new config object for temperature, we see the chart in Figure 9-2. It’s clear from this data we can see that the temperature sensor is not entirely reliable: the temperature in my office rarely exceeds the melting point of steel.

Processing data

An advantage of the approach that we’ve taken is that we can perform relatively arbitrary transforms on the data that we’re given, allowing us to discard data points that we consider to be incorrect. It’s often better to discard data when analyzing than during collection, as bugs in the function to check a data point’s validity won’t cause data loss if it’s only checked during analysis. We can always delete incorrect data after the fact, but we can never recollect data that we chose to ignore.

One way of fixing this problem with the temperature sensor would be to make the clean iterator look at a moving window on the underlying data rather than just one DataPoint at a time. This way, it can use the neighbors of a sensor value to discard values that are too different.

The collections.deque type is useful for this, as it offers a structure with an optional maximum size, so we can add each temperature we find to the deque, but when reading it, we only see the last n entries that were added. A deque can have items added or removed from either the left or right edges, so it’s essential to be consistent about adding and popping from the same end when using it as a limited window.

This cleaner function produces a much smoother temperature trend, as demonstrated in Figure 9-3. The cleaner filters out any data points where the temperature could not be found as well as any severe errors. It is retaining fine detail of temperature trends; as the window contains the last three data points recorded (even those which were not excluded from the data set), a sudden change in temperature will start to be reflected in the output data so long as it persists for at least two consecutive readings.

Interactivity with Jupyter widgets

So far, our code to generate the charts has no interactivity available to the end-user. We are currently displaying all data points ever recorded, but it would be handy to be able to filter to only show a time period without needing to modify the code to generate the chart.

To do this, we add an optional dependency on ipywidgets, using the extras_require functionality of setup.cfg, and reinstall the apd.aggregation package in our environment using pipenv install -e .[jupyter].

With this installed, we can request that Jupyter create interactive widgets for each argument and call the function with the user-selected values. Interactivity allows the person viewing the notebook to choose arbitrary input values without needing to modify the code for the cell or even understand the code.

Figure 9-4 shows an example of a function which adds two integers and which has been connected to Jupyter’s interactivity support. In this case, the two integer arguments are given a default value of 100 and are rendered as sliders. Users can manipulate these sliders, and the result of the function is recomputed automatically.

Multiply nested synchronous and asynchronous code

We can’t pass coroutines to the interactive(...) function as it’s defined to expect a standard, synchronous function. It's a synchronous function itself, so it’s not even possible for it to await the result of a coroutine call. Although IPython and Jupyter allow await constructs in places where they aren’t usually permitted, this is done by wrapping the cell in a coroutine⁷ and scheduling it as a task; it is not deep magic that truly marries synchronous and asynchronous code, it’s a hack for convenience.

Our plotting code involves awaiting the plot_sensor(...) coroutine , so Jupyter must wrap the cell into a coroutine. Coroutines can only be called by coroutines or directly on an event loop’s run(...) function, so asynchronous code generally grows to the point that the entire application is asynchronous. It’s a lot easier to have a group of functions that are all synchronous or all asynchronous than it is to mix the two approaches.

We can’t do that here because we need to provide a function to interactive(...), over which we have no control of the implementation. The way we get around this problem is that we must convert the coroutine into a new synchronous method. We don’t want to rewrite all the code to a synchronous style just to accommodate the interactive(...) function, so a wrapper function to bridge the gap is a better fit.

The coroutine requires access to an event loop that it can use to schedule tasks and which is responsible for scheduling it. The existing event loop we have won’t do, as it is busy executing the coroutine that’s waiting for interactive(...) to return. If you recall, it’s the await keyword that implements cooperative multitasking in asyncio, so our code will only ever switch between different tasks when it hits an await expression.

If we are running a coroutine, we can await another coroutine or task, which allows the event loop to execute other code. Execution won’t return to our code until the function that was being awaited has completed execution, but other coroutines can run in the meantime. We can call synchronous code like interactive(...) from an asynchronous context, but that code can introduce blocking. As this blocking is not blocking on an await statement, execution cannot be passed to another coroutine during this period. Calling any synchronous function from an asynchronous function is equivalent to guaranteeing that a block of code does not contain an await statement, which guarantees that no other coroutine’s code will be run.

Until now, we have used the asyncio.run(...) function to start a coroutine from synchronous code and block waiting for its result, but we’re already inside a call to asyncio.run(main()) so we cannot do this again.⁸ As the interactive(...) call is blocking without an await expression, our wrapper will be running in a context where it’s guaranteed that no coroutine code can run. Although the wrapper function that we use to convert our asynchronous coroutine to a synchronous function must arrange for that coroutine to be executed, it cannot rely on the existing event loop to do this.

To make this explicit, imagine a function that takes two functions as arguments, as shown in Listing 9-18. These functions both return an integer. This function invokes both of the functions that were passed as arguments, adds the results, and then returns the sum of those integers. If all the functions involved are synchronous, there are no problems.

import typing as t

def add_number_from_callback(a: t.Callable[[], int], b: t.Callable[[], int]) -> int:

    return a() + b()

def constant() -> int:

    return 5

print(add_number_from_callback(constant, constant))

Listing 9-18

Example of calling only synchronous functions from a synchronous context

We can even call this add_number_from_callback(...) function from an asynchronous context and get the right result, with the caveat that add_number_from_callback(...) blocks the entire process, potentially negating the benefits of asynchronous code.

async def main() -> None:

    print(add_number_from_callback(constant, constant))

asyncio.run(main())

Our particular invocation is low risk because we know that there are no IO requests which could potentially block for a long time. However, we might want to add a new function that returns a number from a HTTP request. If we already had a coroutine to get the result of a HTTP request, we might want to use this rather than reimplementing this as a synchronous function. An example of a coroutine to get the number (in this case from the random.org random number generator service) is as follows:

import aiohttp

async def async_get_number_from_HTTP_request() -> int:

    uri = "https://www.random.org/integers/?num=1&min=1&max=100&col=1" "&base=10&format=plain"

    async with aiohttp.ClientSession() as http:

        response = await http.get(uri)

        return int(await response.text())

As this is a coroutine, we can’t pass it directly to the add_number_from_callback(...) function. If we were to try, we’d see Python error TypeError: unsupported operand type(s) for +: 'int' and 'coroutine'.⁹

You might write a wrapper function for async_get_number_from_HTTP_request to create a new task that we can wait for, but that would submit the coroutine to the existing event loop, which we’ve already decided isn’t a possible solution. We would have no way of awaiting this task, as it’s not valid to use await in a synchronous function, and it’s not valid to call asyncio.run(...) in a nested fashion. The only way of waiting for this would be to loop doing nothing until the task is complete, but this loop prevents the event loop from scheduling the task, resulting in a contradiction.

def get_number_from_HTTP_request() -> int:

    task = asyncio.create_task(async_get_number_from_HTTP_request())

    while not task.done():

        pass

    return task.result()

The main() task constantly loops over the task.done() check, never hitting an await statement and so never giving way to the async_get_number_from_HTTP_request() task. This function results in a deadlock.

Tip

It’s also possible to create blocking asynchronous code with any long-running loop that doesn’t contain an explicit await statement or an implicit one such as async for and async with.

You shouldn’t need to write a loop that checks for another coroutine’s data, as we’ve done here. You should await that coroutine rather than looping. If you do ever need a loop with no awaits inside, you can explicitly give the event loop a chance to switch into other tasks by awaiting a function that does nothing, such as await asyncio.sleep(0), so long as you’re looping in a coroutine rather than a synchronous function that a coroutine called.

We can’t convert the entire call stack to the asynchronous idiom, so the only remaining way around this problem is to start a second event loop, allowing the two tasks to run in parallel. We’ve blocked our current event loop, but we can start a second one to execute the asynchronous HTTP code.

This approach makes it possible to call async code from synchronous contexts, but all tasks scheduled in the main event loop are still blocked waiting for the HTTP response. This only solves the problem of deadlocks when mixing synchronous and asynchronous code; the performance penalty is still in place. You should avoid mixing synchronous and asynchronous code wherever possible. The resulting code is difficult to understand, can introduce deadlocks, and negates the performance benefits of asyncio.

A helper function that takes a coroutine and executes it in a new thread, without involving the currently running event loop, is given as Listing 9-19. This also includes a coroutine that makes use of this wrapper to pass the HTTP coroutine as though it were a synchronous function.

def wrap_coroutine(f):

    @functools.wraps(f)

    def run_in_thread(*args, **kwargs):

        loop = asyncio.new_event_loop()

        wrapped = f(*args, **kwargs)

        with ThreadPoolExecutor(max_workers=1) as pool:

            task = pool.submit(loop.run_until_complete, wrapped)

        return task.result()

    return run_in_thread

async def main() -> None:

    print(

        add_number_from_callback(

            constant, wrap_coroutine(async_get_number_from_HTTP_request)

        )

    )

Listing 9-19

Wrapper function to start a second event loop and delegate new async tasks there

We can use this same approach to allow our plot_sensor(...) coroutine to be used in an interactive(...) function call, as shown in Listing 9-20.

import asyncio

from uuid import UUID

import ipywidgets as widgets

from matplotlib import pyplot as plt

from apd.aggregation.query import with_database

from apd.aggregation.analysis import (get_known_configs, plot_sensor, wrap_coroutine)

@wrap_coroutine

async def plot(*args, **kwargs):

    location_names = {

     UUID('53998a51-60de-48ae-b71a-5c37cd1455f2'): "Loft",

     UUID('1bc63cda-e223-48bc-93c2-c1f651779d69'): "Living Room",

     UUID('ea0683de-6772-4678-bfe7-6014f54ffc8e'): "Office",

     UUID('5aaa901a-7564-41fb-8eba-50cdd6fe9f80'): "Outside",

    }

    with with_database("postgresql+psycopg2://apd@localhost/apd") as session:

        coros = []

        figure = plt.figure(figsize = (20, 10), dpi=300)

        configs = get_known_configs().values()

        for i, config in enumerate(configs, start=1):

            plot = figure.add_subplot(2, 2, i)

            coros.append(plot_sensor(config, plot, location_names, *args, **kwargs))

        await asyncio.gather(*coros)

    return figure

start = widgets.DatePicker(

    description='Start date',

)

end = widgets.DatePicker(

    description='End date',

)

out = widgets.interactive(plot, collected_after=start, collected_before=end)

display(out)

Listing 9-20

Interactive chart filtering example, with output shown

Persistent endpoints

The next logical piece of cleanup we could do is to move the configuration of endpoints to a new database table. This would allow us to automatically generate the location_names variable, ensure the colors used on each chart are consistent across invocations, and also let us update all sensor endpoints without having to pass their URLs each time.

To do this, we’ll create a new database table and data class to represent a deployment of apd.sensors. We also need command-line utilities to add and edit the deployment metadata, utility functions to get the data, and tests for all of this.

Charting maps and geographic data

We’ve been focused on xy plots of value against time in this chapter, as it represents the test data we’ve been retrieving. Sometimes we need to plot data against other axes. The most common of these is latitude against longitude, so the plot resembles a map.

The shape of the data is only very approximately like an outline of Great Britain, which is shown in Figure 9-5. Most people who look at this plot would not recognize it as such.

The distortion is because we’ve plotted this according to the equirectangular map projection, where latitude and longitude are an equally spaced grid that does not take the shape of the earth into account. There is no one correct map projection; it very much depends on what the map’s intended use is.

We need the map to look familiar to most people, who will be very familiar with the outline of whatever country they live in. We want people who look at it to look at the data, not the unusual projection. The most commonly used projection is the Mercator projection, which the OpenStreetMap (OSM) project provides implementations for in many programming languages, including Python.¹⁰ The merc_x(...) and merc_y(...) functions to implement the projection won’t be included in the listings, as they’re rather complex mathematical functions.

With the projection functions available, we can re-run the plotting function and see that the points match up much more closely with the outline that we expect, as shown in Figure 9-6. In this case, the labels on the axes no longer show coordinates; they show meaningless numbers. We should ignore these labels for now.

New plot types

This only shows us the position of each data point, not the value associated with it. The plotting functions we’ve used so far all plot two values, the x and y coordinates. While we could label the plot points with temperatures, or color code with a scale, the resulting chart isn’t very easy to read. Instead, there are some other plot types in matplotlib that can help us: specifically tricontourf(...). The tricontour family of plotting functions take three-dimensional input of (x, y, value) and interpolate between them to create a plot with areas of color representing a range of values.

While the tricontour functions plot the color areas, we should also plot the points where the measurements were taken, albeit less prominently (Listing 9-23). This works the same way as plotting multiple data sets on a chart; we can call the various plot functions as many times as needed to display all the data; they do not need to be the same type of plot, so long as the axes are compatible.

fig, ax = plt.subplots()

lats = [ll[0] for ll in datapoints.keys()]

lons = [ll[1] for ll in datapoints.keys()]

temperatures = tuple(datapoints.values())

x = tuple(map(merc_x, lons))

y = tuple(map(merc_y, lats))

ax.tricontourf(x, y, temperatures)

ax.plot(x, y, 'wo', ms=3)

ax.set_aspect(1.0)

plt.show()

Listing 9-23

Color contours and scatter on the same plot

This is understandable once we know what we’re looking at, but we can improve it further by plotting the coastline of the island of Great Britain on the map. Given the list of coordinates representing the coastline of Great Britain,¹¹ we can make a final call to the plot function, this time specifying that we want to draw a line rather than dots. The final version (Figure 9-7) of our plot is much easier to read, especially if we enable drawing a legend by calling plt.colorbar(tcf) where tcf is the result of the ax.tricontourf(...) function call.

Tip

There are lots of GIS libraries for Python and Matplotlib that make more complex maps easier. If you’re planning on drawing lots of maps, I’d encourage you to look at Fiona and Shapely for manipulating points and polygons easily. I strongly recommend these libraries to anyone working with geographic information in Python; they’re very powerful indeed.

The basemap toolkit for matplotlib offers very flexible map drawing tools, but the maintainers have decided against distributing it like a standard Python package so I am unable to recommend it as a general solution to map drawing.

Drawing a custom map using the new configs

The changes to Config allow us to define map-based configurations, but our current data doesn’t include any data that can be drawn as a map because none of our deployments includes a location sensor. We can use the new config.get_data(...) option to generate some static data rather than real, aggregated data to demonstrate the functionality. We can also add the custom coastline line by extending the draw_map(...) function (Listing 9-26).

def get_literal_data():

    # Get manually entered temperature data, as our particular deployment

    # does not contain data of this shape

    raw_data = {...}

    now = datetime.datetime.now()

    async def points():

        for (coord, temp) in raw_data.items():

            deployment_id = uuid.uuid4()

            yield DataPoint(sensor_name="Location", deployment_id=deployment_id,

            collected_at=now, data=coord)

            yield DataPoint(sensor_name="Temperature", deployment_id=deployment_id,

            collected_at=now, data=temp)

    async def deployments(*args, **kwargs):

        yield None, points()

    return deployments

def draw_map_with_gb(plot, x, y, colour):

    # Draw the map and add an explicit coastline

    gb_boundary = [...]

    draw_map(plot, x, y, colour)

    plot.plot(

        [merc_x(coord[0]) for coord in gb_boundary],

        [merc_y(coord[1]) for coord in gb_boundary],

        "k-",

    )

country = Config(

    get_data=get_literal_data(),

    clean=get_map_cleaner_for("Temperature"),

    title="Country wide temperature",

    ylabel="",

    draw=draw_map_with_gb,

)

out = widgets.interactive(interactable_plot_multiple_charts(configs=configs + (country, )), collected_after=start, collected_before=end)

Listing 9-26

Jupyter function to draw a custom map chart along with the registered charts

Exercise 9-3: Add a Bar Chart For Cumulative Solar Power

We wrote a cleaner for the solar generation data to convert it to momentary power instead of cumulative power. This makes it much more evident when power is being generated over time, but it makes understanding the amount generated each day harder.

Write a new cleaner that returns cumulative power per day and a new draw function that displays this as a bar chart.

As always, the code accompanying this chapter includes a starting point and a sample completed version.

Summary

In this chapter, we’ve returned to Jupyter for the purpose that people are most familiar with, rather than purely as a prototyping tool. We’ve also used Matplotlib here, which many users of Jupyter will have come across already. Together, these two make a formidable tool for communicating data analysis outcomes.

We’ve written lots of helper functions to make it easy for people to build custom interfaces in Jupyter to view the data we are aggregating. This has allowed us to define a public-facing API while allowing us lots of flexibility to change the way things are implemented. A good API for end-users is vital for retaining users, so it’s worth spending the time on.

The final version of the accompanying code for this chapter includes all the functions we’ve built up, many of which contain long blocks of sample data. Some of these were too long to include in print, so I recommend that you take a look at the code samples and try them out.

Finally, we’ve looked at some more advanced uses of some technologies we’ve used already, including using the __post_init__(...) hook of data classes to preserve backward compatibility when default arguments do not suffice, and more complex combinations of synchronous and asynchronous code.