Joule Application Programming Interface (API)

Note

The Joule Application Programming Interface (API) uses asynchronous coroutines and must be run inside an event loop. To run the examples in this documentation interactively use the asyncio module:

Asyncio REPL

$> python3 -m asyncio
asyncio REPL 3.X.X
...
>>> import asyncio
>>>

A Jupyter Notebook that provides an overview of the API functionality is available from GitHub. To run this notebook follow the commands below to install Jupyter and matplotlib, retrieve the notebook file, and start the Jupyter server:

SHELL

pip install jupyterlab matplotlib # prefix with sudo for system-wide install
wget https://raw.githubusercontent.com/wattsworth/joule/master/API_demo.ipynb
jupyter lab --ip=0.0.0.0 # add the --ip flag to allow external connections

Joule Nodes

A Node represents a Joule instance and is the only means to access API methods. Do not create manually create a Node object. Instead, use joule.api.get_node to create a connection to a specific node. Joule modules have a node object created automatically that refers to the Joule node running it.

Module Access

# Inside a joule.client.ReaderModule class
async def run(self, parsed_args, output):
    node_info = await self.node.info()
    # other code ...

See [#working-with-nodes] for more information on managing access to Joule nodes using the API. These tools are also available through the CLI.

The rest of this section describes class methods separated by category.

`annotation_create(annotation, stream)` `async`

Add an annotation. Create a new :class:joule.api.Annotation object locally and associate it with a data stream. The stream may be specified by path, ID, or a :class:joule.api.DataStream object

Parameters:

Name	Type	Description	Default
`annotation`	`Annotation`	the annotation to add	required
`stream`	`DataStream \| str \| int`	reference to the DataStream to annotate	required

Returns:

Type	Description
`Annotation`	Annotation with ID populated

Example

>>> from joule import utilities
>>> event_note = Annotation(title='Event Annotation',
                    start=utilities.human_to_timestamp('now'))
>>> await node.annotation_create(event_note, '/path/to/stream')
>>> range_note = Annotation(title='Range Annotation',
                    start=utilities.human_to_timestamp('1 minute ago'),
                    end=utilities.human_to_timestamp('now'))
>>> await node.annotation_create(range_note, '/path/to/stream')

`annotation_delete(annotation)` `async`

Remove an annotation. The annotation may be specified by ID or :class:joule.api.Annotation object.

Example

>>> await node.annotation_delete(5) # assuming 5 is a valid annotation ID

`annotation_get(stream, start=None, end=None)` `async`

Retrieve annotations for a particular stream. Specify timestamps to only retrieve annotations over a particular interval. The stream may be specified by path, ID, or a :class:joule.api.DataStream object

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds).. default to the beginning of the data	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). defaults to the end of the data	`None`

Returns:

Type	Description
`List[Annotation]`	list of Annotations that match the specified criteria

Example

>>> annotations = await node.annotation_get("/path/to/stream")
>>> print(annotations[0].name)
"Demo Annotation"

`annotation_update(annotation)` `async`

Update an annotation with new title or content. The time ranges (start,end) may not be changed

Example

>>> await node.annotation_update(annotation) # assuming annotation already exists on the stream

`data_consolidate(stream, max_gap, start=None, end=None)` `async`

Remove interval breaks less than max_gap microseconds in a data stream. Useful for merging data streams with unintentional gaps due to importing data at different times. NOTE: This operation does not recompute the higher level decimations needed to display the data stream in Lumen. If data does not appear in Lumen after consolidating a large number of small intervals, run :meth:Node.data_decimate

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the beginning of the data.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the end of the data.	`None`
`max_gap`	`int`	merge intervals of data that are less than this many microseconds apart.	required

Returns:

Type	Description
`List`	the list of remaining intervals after consolidating

`data_decimate(stream)` `async`

Recompute the decimations for a data stream.

Note

This can take a long time for large streams**

`data_delete(stream, start=None, end=None)` `async`

Delete data from a stream. Specify timestamp bounds to delete a particular range or omit to delete all data. Deleting a range of data creates an interval break in the stream as show in the example below.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the beginning of the data.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the end of the data.	`None`

Example

>>> await node.data_intervals("/parent/my_folder/stream")
[[1551759387204004, 1551863787204004]]
>>> left=1551795387204004  # Tue, 05 Mar 2019 08:16:27
>>> right=1551831387204004 # Tue, 05 Mar 2019 19:16:27
>>> await node.data_delete("/parent/my_folder/stream", start=left, end=right)
>>> await node.data_intervals("/parent/my_folder/stream")
[[1551759387204004, 1551791787204004],
[1551831387204004, 1551863787204004]]

`data_drop_decimations(stream)` `async`

Remove the decimated copies of this stream. This can be useful for reducing the size of the database if Lumen visualizations are not needed for the data stream.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required

`data_intervals(stream, start=None, end=None)` `async`

Retrieve list of data intervals. See :ref:sec-intervals for details on data intervals. Specify timestamp bounds to list intervals over a particular range or omit to list all intervals.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the beginning of the data.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Defaults to the end of the data.	`None`

Example

>>> await node.data_intervals("/parent/my_folder/stream") # unbroken data stream
[[1551759387204004, 1551863787204004]]

>>> await node.data_intervals("/parent/my_folder/stream") # one segment of missing data
[[1551730769556442, 1551757125630201],
[1551758221956119, 1551811071052711]]

>>> await node.data_intervals("/parent/my_folder/stream") # no stream data
[]

`data_read(stream, start=None, end=None, max_rows=None)` `async`

Read historic data from a stream. Specify timestamp bounds for a particular range or omit to read all historic data. This method returns a pipe which should be used to read the data. The pipe must be closed after use. See :ref:pipes for details on Joule Pipes.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	Timestamp in UNIX microseconds. Omit to read from beginning of the stream.	`None`
`end`	`Optional[int]`	Timestamp in UNIX microseconds. Omit to read until the end of the stream.	`None`
`max_rows`	`Optional[int]`	Return a decimated view of the data with at most this many rows, decimations are provided in powers of 4.	`None`

Returns:

Name	Type	Description
`a`	`Pipe`	class:`joule.models.Pipe` connection to the specified stream.

Example

>>> pipe = await node.data_read("/parent/my_folder/stream")
>>> data = await pipe.read() # for large data run in a loop
array([(1551730769556442,      0), (1551730769657062,     10),
    (1551730769757882,     20), ..., (1551751850688250, 661380),
    (1551751850795677, 661390), (1551751850896756, 661400)],
    dtype=[('timestamp', '<i8'), ('data', '<i4')])
>>> pipe.consume(len(data)) # flush the pipe
>>> await pipe.close() # close the data connection

`data_stream_annotation_delete(stream, start=None, end=None)` `async`

Remove annotations from this stream. If start and/or end is specified only remove annotations within this interval. If either start or end are omitted the range defaults to the beginning or end of the stream respectively.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to the target DataStream	required
`start`	`Optional[int]`	only remove annotations after (>=) to UNIX timestamp	`None`
`end`	`Optional[int]`	only remove annotation before (<=) to UNIX timestamp	`None`

`data_stream_create(stream, folder)` `async`

Create a stream and place in the specified folder. Folder may be specified by object, path or numeric ID. See :class:joule.api.DataStream for details on creating DataStream objects. Raises :exc:joule.errors.ApiError if the stream or folder specification is invalid. If the folder is specified by path it will be created if it does not exist.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream`	new DataStream object	required
`folder`	`Folder \| str \| int`	reference to desired parent Folder	required

Returns:

Type	Description
`DataStream`	newly created DataStream object with ID field populated

Example

>>> new_stream = joule.api.DataStream(name="New Stream")
>>> new_stream.elements = [joule.api.Element(name="Element1")]
>>> await node.data_stream_create(new_stream,"/parent/new_folder")
<joule.api.DataStream id=2628 name='New Stream' description='' datatype='float32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>

`data_stream_delete(stream)` `async`

Delete a stream. DataStream may be specified by a :class:joule.api.DataStream object, a path, or numeric ID. Raises :exc:joule.errors.ApiError if the stream specification is invalid or if the stream is locked. To remove data within a stream see :meth:Node.data_delete.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to target DataStream	required

Example

>>> new_stream = joule.api.DataStream(name="New Stream")
>>> new_stream.elements = [joule.api.Element(name="Element1")]
>>> await node.data_stream_create(new_stream,"/parent/new_folder")
<joule.api.DataStream id=2628 name='New Stream' description='' datatype='float32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>

`data_stream_get(stream)` `async`

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to the target DataStream	required

Returns:

Type	Description
`DataStream`	requested DataStream object

Example

>>> stream = await node.data_stream_get("/parent/my_folder/stream")
<joule.api.DataStream id=2627 name='stream' description='' datatype='int32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>

>>> stream = await node.data_stream_get(2627)
<joule.api.DataStream id=2627 name='stream' description='' datatype='int32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>

>>> await node.data_stream_get("/does/not/exist") # raises ApiError
joule.errors.ApiError: stream does not exist [404]

`data_stream_info(stream)` `async`

Get information about a stream as a joule.api.DataStreamInfo object. DataStream may be specified by a joule.api.DataStream object, a path, or numeric ID. Raises :exc:joule.errors.ApiError if the stream specification is invalid.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to target DataStream	required

Returns:

Type	Description
`DataStreamInfo`	information about the target DataStream

Example

>>> await node.data_stream_info("/parent/my_folder/stream")
<joule.api.DataStreamInfo start=1551730769556442 end=1551751402742424
rows=61440, total_time=20633185982>

`data_stream_move(stream, folder)` `async`

Move a DataStream into a different Folder. The stream and folder may be specified by objects, paths, or numeric ID's. The stream name must be unique in the destination and not be locked (actively in use by a module or statically configured). Raises :exc:joule.errors.ApiError if stream or folder specifications are invalid or the requested move cannot be performed. The destination is automatically created if it does not exist.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to target DataStream	required
`folder`	`Folder \| str \| int`	reference to desired parent Folder	required

Example

>>> await node.data_stream_move("/parent1/my_folder/stream","/parent2")
>>> parent2 = await node.folder_get("/parent2")
>>> parent2.data_streams
[<joule.api.DataStream id=2627 name='stream' description='' datatype='int32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>]

>>> await node.data_stream_move("/does/not/exist","/parent2") # raises ApiError
joule.errors.ApiError: stream does not exist [404]

`data_stream_update(stream)` `async`

Update stream and element attributes. Raises :exc:joule.errors.ApiError if stream is locked or the specification is invalid. The datatype and number of elements may not be changed.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream`	DataStream object with modified attributes	required

Example

>>> stream = await node.data_stream_get("/parent/my_folder/stream")
>>> stream.elements # Element name is "Element1"
[<joule.api.Element id=3192 index=0, name='Element1' units=None
plottable=True display_type='CONTINUOUS'>]
>>> stream.elements[0].name="New Name"
>>> await node.data_stream_update(stream) # send updated values to Joule
>>> updated_stream = await node.data_stream_get(stream) # refresh local copy
>>> updated_stream.elements # Element name is now "New Name"
[<joule.api.Element id=3192 index=0, name='New Name' units=None
plottable=True display_type='CONTINUOUS'>]

`data_subscribe(stream)` `async`

Read live data from a stream. The stream must be actively produced by a module. This method returns a pipe which should be used to read the data. The pipe must be closed after use. See :ref:pipes for details on Joule Pipes.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required

Returns:

Type	Description
`Pipe`	a Pipe connection to the live DataStream

Example

>>> pipe = await node.data_read("/live/stream")
>>> data = await pipe.read() # run in a loop for continuous updates
array([(1551730769556442,      0), (1551730769657062,     10),
    (1551730769757882,     20), ..., (1551751850688250, 661380),
    (1551751850795677, 661390), (1551751850896756, 661400)],
    dtype=[('timestamp', '<i8'), ('data', '<i4')])
>>> pipe.consume(len(data)) # flush the pipe
>>> await pipe.close() # close the data connection

`data_write(stream, start=None, end=None, merge_gap=0)` `async`

Write data to a stream. The stream must not be an active destination from any other source. Optionally specify start and end timestamps to remove existing data over the interval you plan to write. Omit to merge the new data with the existing data. This method returns a pipe which should be used to write the data. The pipe must be closed after use. See :ref:pipes for details on Joule Pipes.

Parameters:

Name	Type	Description	Default
`stream`	`DataStream \| str \| int`	reference to a DataStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Omit to merge with existing data	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Omit to merge with existing data	`None`
`merge_gap`	`int`	merge with surrounding intervals if gap is less than this	`0`

Example

>>> import numpy as np
>>> pipe_out = await node.data_write("/parent/my_folder/stream")
>>> for i in range(4):
...   await pipe_out.write(np.array([[joule.utilities.time_now(), i]]))
>>> await pipe_out.close()
>>> pipe_in = await node.data_read("/parent/my_folder/stream") # read the data back
>>> await pipe_in.read()
array([(1551758297114942, 0.), (1551758297115062, 1.),
    (1551758297115090, 2.), (1551758297115111, 3.)],
    dtype=[('timestamp', '<i8'), ('data', '<f4')])

`db_connect()` `async`

Create a connection to the node database. Note the node's pg_hba.conf must allow remote connections to the database. The connection provides readonly access to the data and metadata schemas but does allow creation of new schemas or table creation in the public schema.

Returns:

Type	Description
`Engine`	authenticated connection to the database as the `joule_module` user.

`db_connection_info()` `async`

Connection information necessary to connect to the node database. This is useful if the IP address or domain name must be changed before connecting to the databse. Returns a :class:joule.utilities.ConnectionInfo object.

Returns:

Type	Description
`ConnectionInfo`	information necessary to establish a connection as the `joule_module` user

`event_stream_count(stream, start=None, end=None, json_filter=None, include_on_going_events=True)` `async`

The parameters are the same as :meth:Node.event_stream_read_list.

Returns:

Type	Description
`int`	The number of Events matching the specified criteria

`event_stream_create(stream, folder)` `async`

Create an event stream and place in the specified folder. Folder may be specified by object, path or numeric ID. See :class:joule.api.EventStream for details on creating EventStream objects. Raises :exc:joule.errors.ApiError if the stream or folder specification is invalid. If the folder is specified by path it will be created if it does not exist.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream`	the EventStream to create	required
`folder`	`Folder \| str \| int`	reference to the desired parent Folder	required

Returns:

Type	Description
`EventStream`	configured EventStream with ID populated

Example

>>> new_stream = joule.api.EventStream(name="New Stream")
>>> await node.event_stream_create(new_stream,"/parent/new_folder")
<joule.api.EventStream id=38 name='New Stream' description=''>

`event_stream_delete(stream)` `async`

Delete a stream. EventStream may be specified by a :class:joule.api.EventStream object, a path, or numeric ID. Raises :exc:joule.errors.ApiError if the stream specification is invalid or if the stream is locked. To remove data within a stream see :meth:Node.event_stream_remove.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required

Example

>>> folder = await node.folder_get("/parent/my_folder")
>>> folder.event_streams # my_folder has one stream
[<joule.api.EventStream id=39 name='stream' description='example'>
>>> await node.event_stream_delete("/parent/my_folder/stream") # delete the stream
>>> folder = await node.folder_get("/parent/my_folder")
>>> folder.event_streams # my_folder is now empty
[]

`event_stream_get(stream, create=False, description='', chunk_duration='', chunk_duration_us=None, event_fields=None)` `async`

Retrieve the specified stream. EventStream may be specified by a :class:joule.api.EventStream object, a path, or a numeric ID. Raises :exc:joule.errors.ApiError if stream specification is invalid. If stream is a path and create=True then the EventStream is created if it does not exist using the description, chunk_duration specification and event_fields parameters. If the stream exists these parameters are ignored. See :meth:Node.event_stream_update for modifying existing streams. See :meth:Node.event_stream_create for more information creating streams.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to target stream	required
`create`	`bool`	create stream if it does not exist	`False`
`description`	`str`	additional information about this stream	`''`
`chunk_duration`	`str`	TimescaleDB chunk duration specified using time syntax (TODO)	`''`
`chunk_duration_us`	`Optional[int]`	TimescaleDB chunk duration specified in microseconds	`None`
`event_fields`		dictionary of event fields, see event field documentation (TODO)	`None`

Returns:

Type	Description
`EventStream`	the newly created or pre-existing EventStream object

Example

>>> await node.event_stream_get("/folder/event_stream", create=True,
    description="example event stream",
    event_fields={"Property A": "string", "Property B": "numeric"})

>>> await node.event_stream_get("/folder/event_stream") # generates exception if stream does not exist

`event_stream_info(stream)` `async`

Get information about a stream as a :class:joule.api.EventStreamInfo object. EventStream may be specified by a :class:joule.api.EventStream object, a path, or numeric ID. Raises :exc:joule.errors.ApiError if the stream specification is invalid.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required

Returns:

Type	Description
`EventStreamInfo`	information about the EventStream

Example

>>> await node.event_stream_info("/parent/my_folder/stream")
<joule.api.EventStreamInfo start=1643858892000000 end=1643891982000000
    events=10, total_time=33090000000>

`event_stream_move(stream, folder)` `async`

Move a stream into a different folder. The stream and folder may be specified by objects, paths, or numeric ID's. The stream name must be unique in the destination. Raises :exc:joule.errors.ApiError if stream or folder specifications are invalid or the requested move cannot be performed. The destination is automatically created if it does not exist.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to EventStream	required
`folder`	`Folder \| str \| int`	reference to desired parent Folder	required

Example

>>> await node.event_stream_move("/parent1/my_folder/stream","/parent2")
>>> parent2 = await node.folder_get("/parent2")
>>> parent2.event_streams
[<joule.api.DataStream id=2627 name='stream' description='' datatype='int32'
is_configured=False is_source=False is_destination=False locked=False decimate=True>]

>>> await node.event_stream_move("/does/not/exist","/parent2") # raises ApiError
joule.errors.ApiError: stream does not exist [404]

`event_stream_read(stream, start=None, end=None, json_filter=None, include_on_going_events=True, block_size=None)`

Read events from an existing stream using an asynchronous generator. Specify block_size to iterate over lists of events, otherwise iterates over each event individually. Recommended for large streams where all events might not fit into memory.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read from start of stream.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read to end of stream.	`None`
`limit`		Limit query to first N result.	required
`json_filter`	`Optional[Dict[str, str]]`	Select events based on content, see below.	`None`
`include_on_going_events`		Set to `False` to only return events that begin within the specified interval. By default this function returns all events that are active within the specified interval.	`True`
`block_size`	`Optional[int]`	if > 1, yields a list of Events, otherwise yields a single Event	`None`

Yields:

Type	Description
`AsyncGenerator[Union[Event, List[Event]], None]`	list of Events with max length = `block_size` if specified, otherwise a single Event at a time

Example

Iterate over each event in a stream, or iterate over blocks of events (more efficient)

>>> async for event in node.event_stream_read(stream):
...     print(f"processing {event}...")
>>> async for events in node.event_stream_read(stream, block_size=2):
...     assert len(events)<=2 # maximum length is block_size
...     print(f"processing {events}...")

`event_stream_read_list(stream, start=None, end=None, json_filter=None, include_on_going_events=True, limit=10000)` `async`

Read events from an existing stream. Returns a list of :class:joule.api.Event objects. If the expected number of events is very large consider using the async generator form of this function above.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read from start of stream.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read to end of stream.	`None`
`limit`		Limit query to first N result.	`10000`
`json_filter`	`Optional[Dict[str, str]]`	Select events based on content, see below.	`None`
`include_on_going_events`		Set to `False` to only return events that begin within the specified interval. By default this function returns all events that are active within the specified interval.	`True`

Returns:

Type	Description
`List[Event]`	A list of Events matching the specified criteria

JSON Filter Format

Specify a list of lists where each item is a tuple of [key,comparison,value] Outer lists are OR conditions, inner lists are AND conditions. Comparison must be one of the following

String Comparison: is, not, like, unlike
Numeric Comparison: gt, gte, lt, lte, eq, neq

Example

Retrieve the first 100 events where `name` is LIKE `sample` OR `value` is between 0 AND 10:

>>>  #       ((-------- test ----------)  OR (------test------ AND ------test-------))
>>> filter = [[['name','like','sample']]   , [['value','gt',0],    ['value','lt',10]]]
>>> await node.event_stream_read_list(stream,limit=100,json_filter=json.dumps(filter))

`event_stream_remove(stream, start=None, end=None, json_filter=None)` `async`

Remove events from an existing stream. EventStream may be specified by a :class:joule.api.EventStream object, a path, or numeric ID. Specify start and/or end timestamps to remove over events over a specific time range. Use json_filter to remove events based on content.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required
`start`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read from start of stream.	`None`
`end`	`Optional[int]`	UNIX timestamp (microseconds). Omit to read to end of stream.	`None`
`json_filter`		Select events based on content, see below.	`None`

`event_stream_update(stream)` `async`

Update the event stream name and/or description. Raises :exc:joule.errors.ApiError if the specification is invalid.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream`	EventStream object with updated attributes	required

Example

>>> stream = await node.event_stream_get("/parent/my_folder/stream")
>>> stream.elements # Element name is "Element1"
[<joule.api.Element id=3192 index=0, name='Element1' units=None
plottable=True display_type='CONTINUOUS'>]
>>> stream.elements[0].name="New Name"
>>> await node.data_stream_update(stream) # send updated values to Joule
>>> updated_stream = await node.data_stream_get(stream) # refresh local copy
>>> updated_stream.elements # Element name is now "New Name"
[<joule.api.Element id=3192 index=0, name='New Name' units=None
plottable=True display_type='CONTINUOUS'>]

`event_stream_write(stream, events)` `async`

Add events to an existing stream. EventStream may be specified by a :class:joule.api.EventStream object, a path, or numeric ID. Events is a list of :class:joule.api.Event objects.

Parameters:

Name	Type	Description	Default
`stream`	`EventStream \| str \| int`	reference to an EventStream	required
`events`	`List[Event]`	Event objects to add to the stream	required

Returns:

Type	Description
`List[Event]`	list of Event objects with ID populated

Example

>>> from joule.utilities import time_now
>>> e1 = Event(start_time=time_now()-1e6, end_time = time_now(), content={'name': 'event1'})
>>> await asyncio.sleep(2)
>>> e2 = Event(start_time=time_now()-1e6, end_time = time_now(), content={'name': 'event2'})
>>> await node.event_stream_write("/plugs/events",[e1,e2])

`folder_delete(folder, recursive=False)` `async`

Parameters:

Name	Type	Description	Default
`folder`	`Folder \| str \| int`	reference to target folder	required
`recursive`	`bool`	if `True` remove child folders, if `False` raise an exception if child folders exist	`False`

`folder_get(folder)` `async`

Retrieve the specified folder. Folder may be specified by a :class:joule.api.Folder object, a path, or numeric ID. Raises :exc:joule.errors.ApiError if folder specification is invalid.

Parameters:

Name	Type	Description	Default
`folder`	`Folder \| str \| int`	reference to target Folder	required

Returns:

Type	Description
`Folder`	requested Folder object

Example

>>> folder = await node.folder_get("/parent/my_folder")  # query by path
<joule.api.Folder id=2729 name='archive' description=None locked=True>

>>> live = await node.folder_get(2730) # query by ID
<joule.api.Folder id=2730 name='my_folder' description=None locked=True>

>>> await node.folder_get("/does/not/exist") # raises ApiError
joule.errors.ApiError: folder does not exist [404]

`folder_move(source, destination)` `async`

Move the source folder into the destination folder. The source and destination may be specified by joule.api.Folder objects, paths, or numeric ID's. The source folder name must be unique in the destination and not be locked (actively in use by a module or statically configured) Raises :exc:joule.errors.ApiError if folder specifications are invalid or the requested move cannot be performed. The destination is automatically created if it does not exist.

Parameters:

Name	Type	Description	Default
`source`	`Folder \| str \| int`	reference to the target folder	required
`destination`	`Folder \| str \| int`	reference to the desired parent folder	required

Example

>>> await node.folder_move("/parent1/my_folder","/parent2")
>>> parent2 = await node.folder_get("/parent2")
>>> parent2.children
[<joule.api.Folder id=3321 name='my_folder' description=None locked=False>]

>>> await node.folder_move("/parent1/missing_folder","/parent2") # rasises ApiError
joule.errors.ApiError: folder does not exist [404]

`folder_root()` `async`

Retrieve the node's root folder. This returns the entire database structure as shown in the example below.

Returns:

Type	Description
`Folder`	root folder object which includes the full folder hierarchy

Example

>>> root = await node.folder_root()
>>> root
<joule.api.Folder id=2728 name='root' description=None locked=True>
>>> root.children
[<joule.api.Folder id=3123 name='tmp' description=None locked=False>,
<joule.api.Folder id=2729 name='archive' description=None locked=True>,
<joule.api.Folder id=2730 name='live' description=None locked=True>]

`folder_update(folder)` `async`

Update folder attributes. The name and description are the only writable attributes. Raises :exc:joule.errors.ApiError if folder is locked or the specification is invalid

Example

>>> folder = await node.folder_get("/parent/my_folder")
<joule.api.Folder id=3329 name='my_folder' description=None locked=False>
>>> folder.name="new name"
>>> await node.folder_update(folder) # save the change
>>> parent = await node.folder_get("/parent")
>>> parent.children # folder 3329 has a new name
[<joule.api.Folder id=3329 name='new name' description=None locked=False>]

`module_get(module, statistics=False)` `async`

Retrieve a specific module as a :class:joule.api.Module object. If statistics is True retrieve CPU and memory statistics for each module. Collecting statistics takes additional time because the CPU usage is averaged over a short time interval. Module may be specified by object, name, or numeric ID.

Parameters:

Name	Type	Description	Default
`module`	`Module \| str \| int`	reference to a Module	required
`statistics`	`bool`	include memory and CPU usage information	`False`

Returns:

Type	Description
`Module`	the requested Module

Example

>>> my_module = await node.module_get("my module", statistics=True)
<joule.api.Module id=0 name='my module' description='adds 1 to the input' is_app=False>
>>> my_module.statistics
<joule.api.ModuleStatistics pid=1460 create_time=1551805343.4 cpu_percent=5.60 memory_percent=6.23>

`module_list(statistics=False)` `async`

Retrieve a list of the current Joule modules as :class:joule.api.Module objects. If statistics is True retrieve CPU and memory statistics for each module. Collecting statistics takes additional time because the CPU usage is averaged over a short time interval.

Parameters:

Name	Type	Description	Default
`statistics`	`bool`	include memory and CPU usage information	`False`

Returns:

Type	Description
`List[Module]`	list of the active modules

Example

>>> await node.module_list()
[<joule.api.Module id=0 name='plus1' description='adds 1 to the input' is_app=False>,
<joule.api.Module id=1 name='counter' description='counts up by 10s' is_app=False>]

`module_logs(module)` `async`

Retrieve a list of module logs. Logs are the stdout and stderr streams from the module. The easiest way to generate logs is by adding print statements to a module. The maximum number of lines is controlled by the MaxLogLines parameter in the main configuration file (see :ref:sec-system-configuration). Logs are automatically rolled if a module produces more than the maximum number of lines. The module may be specified by object, name, or numeric ID.

Parameters:

Name	Type	Description	Default
`module`	`Module \| str \| int`	reference to a Module	required

Returns:

Type	Description
`List[str]`	list of recently printed lines to stderr and stdout

Example

>>> await node.module_logs("my module")
['[2019-03-04T22:57:01.049266]: ---starting module---',
'[2019-03-04T22:59:02.089463]: serial input restarted',
'[2019-03-04T23:04:36.948160]: WARNING: temperature > 98.6']

`proxy_get(proxy)` `async`

Retrieve a specific proxy as a :class:joule.api.Proxy object. Proxy may be specified by object, name, or numeric ID.

Returns:

Type	Description
`Proxy`	requested Proxy object

Example

>>> await node.proxy_get("flask app")
<joule.api.Proxy id=0 name='flask app' proxied_url='http://localhost:8088/interface/p0'
target_url='http://localhost:5000'>

`proxy_list()` `async`

Retrieve a list of proxied URL's as :class:joule.api.Proxy objects.

Returns:

Type	Description
`List[Proxy]`	a list of proxied URL's

Example

>>> await node.proxy_list()
[<joule.api.Proxy id=0 name='flask_app' proxied_url='http://localhost:8088/interface/p0'
target_url='http://localhost:5000'>,
<joule.api.Proxy id=1 name='intranet_host' proxied_url='http://localhost:8088/interface/p1'
target_url='http://internal.domain.com'>]

Working with Nodes

`joule.api.delete_node(node)`

Remove a node from the current user's configuration

Warning

This operation will remove the key material for the node and cannot be reversed

Parameters:

Name	Type	Description	Default
`node`	`str \| BaseNode`	the node to remove	required

`joule.api.get_node(name='')`

Connect to a Joule node. Retrieve a list of node names with CLI tool joule node list

Parameters:

Name	Type	Description	Default
`name`	`str`	omit to use the default node	`''`

Returns:

Type	Description
`BaseNode`	a connection to the Joule node

`joule.api.get_nodes()`

Returns:

Type	Description
`List[BaseNode]`	all nodes that the current user has access to

`joule.api.save_node(node)`

Persist any changes to node attributes (name, URL, or key) Args: node: modified node to save

Asyncio REPL

>>> import joule
>>> my_node = joule.api.get_node()

API Models

API Folder model. See :ref:sec-node-folder-actions for details on using the API to manipulate folders. Folders are locked if they contain locked streams (streams that are active or statically configured). Folder objects should not be created directly. They should be received via API calls.

Parameters:

Name	Type	Description	Default
`name`	`str`	folder name, must be unique in the parent	`''`
`description`	`str`	optional field	required
`locked`	`bool`	folder may not be moved, deleted, or changed	required
`data_streams`	`List[DataStream]`	data streams in the folder	required
`event_streams`	`List[EventStream]`	event streams in the folder	required
`children`	`List[Folder]`	subfolders in the folder	required

API EventStream model. See :ref:sec-node-event-stream-actions for details on using the API to manipulate event streams.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name, must be unique in the parent	`''`
`description`	`str`	optional field	`''`

API EventStreamInfo model. Received from :meth:Node.event_stream_info and should not be created directly.

Parameters:

Name	Type	Description	Default
`start`	`int`	timestamp in UNIX microseconds of the beginning of the first event	required
`end`	`int`	timestamp in UNIX microsseconds of the end of the last data event	required
`event_count`	`int`	number of events in the stream	required
`bytes`	`int`	approximate size of the data on disk	`0`
`total_time`	`int`	event stream duration in microseconds (start-end)	`0`

API DataStream model. See :ref:sec-node-data-stream-actions for details on using the API to manipulate data streams. Streams are locked if they are active or statically configured. When creating a stream manually, omit the ID and status attributes (is_configured, is_source, is_destination, active, and locked), these are set by the Joule server.

Parameters:

Name	Type	Description	Default
`name`	`str`	stream name, must be unique in the parent	`''`
`description`	`str`	optional field	`''`
`datatype`	`str`	element datatype	`'float32'`
`keep_us`	`int`	store the last N microseconds of data (-1 to keep all and 0 to keep none)	`-1`
`is_configured`		is the stream statically configured with a `.conf` file	required
`is_source`		is the stream an active data source	required
`is_destination`		is the stream an active data destination	required
`active`	`bool`	is the stream a source or destination	required
`locked`	`bool`	is the stream active or configured	required
`decimate`	`bool`	is the stream data decimated for visualization	required
`elements`	`List[Element]`	list of the stream elements	`None`

API DataStreamInfo model. Received from :meth:Node.data_stream_info and should not be created directly.

.. warning:: Rows and Bytes values are approximate

Parameters:

Name	Type	Description	Default
`start`	`int`	timestamp in UNIX microseconds of the first data element	required
`end`	`int`	timestamp in UNIX microsseconds of the last data element	required
`rows`	`int`	approximate rows of data in the stream	required
`bytes`	`int`	approximate size of the data on disk	`0`
`total_time`	`int`	data duration in microseconds (start-end)	`0`

API Element model. Streams have one or more elements. See :ref:sec-data-streams for details on the stream data model.

Parameters:

Name	Type	Description	Default
`id`	`int`	unique numeric ID assigned by Joule	required
`index`	`int`	column position in the data array (0 = first element)	required
`name`	`str`	element name	`''`
`units`	`str`	unit of measurement, may be any string	`''`
`plottable`	`bool`	should the element be visible in the Lumen plotting interface	`True`
`display_type [continous\|discrete\|event]`		plot type, defaults to continuous	required
`offset`	`float`	offset data visualization by `y=(x-offset)*scale_factor`	required
`scale_factor`	`float`	scale data visualation with above equation	required
`default_max`	`float`	fix auto scale max (set to None to fit plotted data)	required
`default_min`	`float`	fix auto scale min (set to None to fit plotted data)	required

API Module model. See :ref:sec-node-module-actions for details on using the API to query modules. See :ref:modules for details on writing new modules. See :ref:sec-modules for details on adding modules to Joule. Use :meth:Node.data_stream_get to retrieve the associated stream from the path string in the inputs and outputs dictionaries.

Parameters:

Name	Type	Description	Default
`id`	`int`	unique numeric ID assigned by Joule server	required
`name`	`str`	module name, must be unique	required
`description`	`str`	optional field	required
`is_app`	`bool`	whether the module provides a web interface	required
`inputs`	`Dict`	Mapping of `[pipe_name] = stream_path` for module inputs	required
`outputs`	`Dict`	Mapping of `[pipe_name] = stream_path` for output connections	required
`statistics`	`ModuleStatistics`	execution statistics, may be `None` depending on API call parameters	`None`

API ModuleStatistics model. Received by settings statistics to True when calling :meth:Node.module_get or :meth:Node.module_info and should not be created directly.

Parameters:

Name	Type	Description	Default
`pid`	`int`	process ID	required
`create_time`	`int`	process creation time in UNIX seconds	required
`cpu_percent`	`float`	approximate CPU usage	required
`memory_percent`	`float`	approximate memory usage	required

API Annotation model. See :ref:sec-node-annotation-actions for details on using the API to manipulate annotations. Annotations are associated with streams and may either coverage a range of data or a single event. If end is None the annotation marks an event, otherwise it marks a range.

Parameters:

Name	Type	Description	Default
`title`	`string`	annotation title	required
`content`	`string`	additional description (optional)	`''`
`start`	`int`	Unix microsecond timestamp	required
`end`	`int`	specify for range annotation, omit for event annotation	`None`

API Exceptions

Bases: Exception

Error generated by an API call. Catches all API related errors.

Joule Application Programming Interface (API)

Joule Nodes

annotation_create(annotation, stream) async

annotation_delete(annotation) async

annotation_get(stream, start=None, end=None) async

annotation_update(annotation) async

data_consolidate(stream, max_gap, start=None, end=None) async

data_decimate(stream) async

data_delete(stream, start=None, end=None) async

data_drop_decimations(stream) async

data_intervals(stream, start=None, end=None) async

data_read(stream, start=None, end=None, max_rows=None) async

data_stream_annotation_delete(stream, start=None, end=None) async

data_stream_create(stream, folder) async

data_stream_delete(stream) async

data_stream_get(stream) async

data_stream_info(stream) async

data_stream_move(stream, folder) async

data_stream_update(stream) async

data_subscribe(stream) async

data_write(stream, start=None, end=None, merge_gap=0) async

db_connect() async

db_connection_info() async

event_stream_count(stream, start=None, end=None, json_filter=None, include_on_going_events=True) async

event_stream_create(stream, folder) async

event_stream_delete(stream) async

event_stream_get(stream, create=False, description='', chunk_duration='', chunk_duration_us=None, event_fields=None) async

event_stream_info(stream) async

event_stream_move(stream, folder) async

event_stream_read(stream, start=None, end=None, json_filter=None, include_on_going_events=True, block_size=None)

event_stream_read_list(stream, start=None, end=None, json_filter=None, include_on_going_events=True, limit=10000) async

event_stream_remove(stream, start=None, end=None, json_filter=None) async

event_stream_update(stream) async

event_stream_write(stream, events) async

folder_delete(folder, recursive=False) async

folder_get(folder) async

folder_move(source, destination) async

folder_root() async

folder_update(folder) async

module_get(module, statistics=False) async

module_list(statistics=False) async

module_logs(module) async

proxy_get(proxy) async

proxy_list() async

Working with Nodes

joule.api.delete_node(node)

joule.api.get_node(name='')

joule.api.get_nodes()

joule.api.save_node(node)

API Models

API Exceptions

Utilities

`annotation_create(annotation, stream)` `async`

`annotation_delete(annotation)` `async`

`annotation_get(stream, start=None, end=None)` `async`

`annotation_update(annotation)` `async`

`data_consolidate(stream, max_gap, start=None, end=None)` `async`

`data_decimate(stream)` `async`

`data_delete(stream, start=None, end=None)` `async`

`data_drop_decimations(stream)` `async`

`data_intervals(stream, start=None, end=None)` `async`

`data_read(stream, start=None, end=None, max_rows=None)` `async`

`data_stream_annotation_delete(stream, start=None, end=None)` `async`

`data_stream_create(stream, folder)` `async`

`data_stream_delete(stream)` `async`

`data_stream_get(stream)` `async`

`data_stream_info(stream)` `async`

`data_stream_move(stream, folder)` `async`

`data_stream_update(stream)` `async`

`data_subscribe(stream)` `async`

`data_write(stream, start=None, end=None, merge_gap=0)` `async`

`db_connect()` `async`

`db_connection_info()` `async`

`event_stream_count(stream, start=None, end=None, json_filter=None, include_on_going_events=True)` `async`

`event_stream_create(stream, folder)` `async`

`event_stream_delete(stream)` `async`

`event_stream_get(stream, create=False, description='', chunk_duration='', chunk_duration_us=None, event_fields=None)` `async`

`event_stream_info(stream)` `async`

`event_stream_move(stream, folder)` `async`

`event_stream_read(stream, start=None, end=None, json_filter=None, include_on_going_events=True, block_size=None)`

`event_stream_read_list(stream, start=None, end=None, json_filter=None, include_on_going_events=True, limit=10000)` `async`

`event_stream_remove(stream, start=None, end=None, json_filter=None)` `async`

`event_stream_update(stream)` `async`

`event_stream_write(stream, events)` `async`

`folder_delete(folder, recursive=False)` `async`

`folder_get(folder)` `async`

`folder_move(source, destination)` `async`

`folder_root()` `async`

`folder_update(folder)` `async`

`module_get(module, statistics=False)` `async`

`module_list(statistics=False)` `async`

`module_logs(module)` `async`

`proxy_get(proxy)` `async`

`proxy_list()` `async`

`joule.api.delete_node(node)`

`joule.api.get_node(name='')`

`joule.api.get_nodes()`

`joule.api.save_node(node)`