-
Notifications
You must be signed in to change notification settings - Fork 0
5. Adapters
To get various data sources into and out of the graph, various Input and Output Adapters are available, such as CSV, Parquet, and database adapters (amongst others). Users can also write their own input and output adapters, as explained below.
There are two types of Input Adapters: Historical (aka Simulated) adapters and Realtime Adapters. Historical adapters are used to feed in historical timeseries data into the graph from some data source which has timeseries data. Realtime Adapters are used to feed in live event based data in realtime, generally events created from external sources on separate threads.
There is not distinction of Historical vs Realtime output adapters since outputs need not care if the generated timeseries data which are wired into them are generated from realtime or historical inputs.
In CSP terminology, a single adapter corresponds to a single timeseries edge in the graph. There are common cases where a single data source may be used to provide data to multiple adapter (timeseries) instances, for example a single CSV file with price data for many stocks can be read once but used to provide data to many individual, one per stock. In such cases an AdapterManager is used to coordinate management of the single source (CSV file, database, Kafka connection, etc) and provided data to individual adapters.
Note that adapters can be quickly written and prototyped in python, and if needed can be moved to a c+ implementation for more efficiency.
The Kafka adapter is a user adapter to stream data from a Kafka bus as a reactive time series. It leverages the librdkafka C/C++ library internally.
The KafkaAdapterManager instance represents a single connection to a broker.
A single connection can subscribe and/or publish to multiple topics.
KafkaAdapterManager(
broker,
start_offset: typing.Union[KafkaStartOffset,timedelta,datetime] = None,
group_id: str = None,
group_id_prefix: str = '',
max_threads=100,
max_queue_size=1000000,
auth=False,
security_protocol='SASL_SSL',
sasl_kerberos_keytab='',
sasl_kerberos_principal='',
ssl_ca_location='',
sasl_kerberos_service_name='kafka',
rd_kafka_conf_options=None,
debug: bool = False,
poll_timeout: timedelta = timedelta(seconds=1)
):-
broker: name of the Kafka broker, such asprotocol://host:port -
start_offset: signify where to start the stream playback from (defaults toKafkaStartOffset.LATEST). Can be one of theKafkaStartOffsetenum types or:-
datetime: to replay from the given absolute time -
timedelta: this will be taken as an absolute offset from starttime to playback from
-
-
group_id: if set, this adapter will behave as a consume-once consumer.start_offsetmay not be set in this case since adapter will always replay from the last consumed offset. -
`group_id_prefix: when not passing an explicit group_id, a prefix can be supplied that will be use to prefix the UUID generated for the group_id
-
max_threads: maximum number of threads to create for consumers. The topics are round-robin'd onto threads to balance the load. The adapter won't create more threads than topics. -
max_queue_size: maximum size of the (internal to Kafka) message queue. If the queue is full, messages can be dropped, so the default is very large.
In order to publish or subscribe, you need to define a MsgMapper. These are the supported message types:
JSONTextMessageMapper(datetime_type = DateTimeType.UNKNOWN)ProtoMessageMapper(datetime_type = DateTimeType.UNKNOWN)
You should choose the DateTimeType based on how you want (when publishing) or expect (when subscribing) your datetimes to be represented on the wire.
The supported options are:
UINT64_NANOSUINT64_MICROSUINT64_MILLISUINT64_SECONDS
The enum is defined in csp/adapters/utils.py.
Note the JSONTextMessageMapper currently does not have support for lists.
To subscribe to json data with lists, simply subscribe using the RawTextMessageMapper and process the text into json (e.g. via json.loads).
Once you have an KafkaAdapterManager object and a MsgMapper object, you can subscribe to topics using the following method:
KafkaAdapterManager.subscribe(
ts_type: type,
msg_mapper: MsgMapper,
topic: str,
key=None,
field_map: typing.Union[dict,str] = None,
meta_field_map: dict = None,
push_mode: csp.PushMode = csp.PushMode.LAST_VALUE,
adjust_out_of_order_time: bool = False
):-
ts_type: the timeseries type you want to get the data on. This can be acsp.Structor basic timeseries type -
msg_mapper: theMsgMapperobject discussed above -
topic: the topic to subscribe to -
key: The key to subscribe to. IfNone, then this will subscribe to all messages on the topic. Note that in this "wildcard" mode, all messages will tick as "live" as replay in engine time cannot be supported -
field_map: dictionary of{message_field: struct_field}to define how the subscribed message gets mapped onto the struct -
meta_field_map: to extract meta information from the kafka message, provide a meta_field_map dictionary of meta field info → struct field name to place it into. The following meta fields are currently supported:-
"partition": which partition the message came from -
"offset": the kafka offset of the given message -
"live": whether this message is "live" and not being replayed -
"timestamp": timestamp of the kafka message -
"key": key of the message
-
-
push_mode:csp.PushMode(LAST_VALUE, NON_COLLAPSING, BURST) -
adjust_out_of_order_time: in some cases it has been seen that kafka can produce out of order messages, even for the same key. This allows the adapter to be more laz and allow it through by forcing time to max(time, prev time)
Similarly, you can publish on topics using the following method:
KafkaAdapterManager.publish(
msg_mapper: MsgMapper,
topic: str,
key: str,
x: ts['T'],
field_map: typing.Union[dict,str] = None
):-
msg_mapper: same as above -
topic: same as above -
key: key to publish to -
x: the timeseries to publish -
field_map: dictionary of {struct_field: message_field} to define how the struct gets mapped onto the published message. Note this dictionary is the opposite of the field_map in subscribe()
If you are having issues, such as not getting any output or the application simply locking up, start by ensuring that you are logging the adapter's status() with a csp.print/log call and set debug=True.
Then follow the known issues below.
-
Reason:
GSSAPI Error: Unspecified GSS failure. Minor code may provide more information (No Kerberos credentials available)- Resolution: Kafka uses kerberos tickets for authentication. Need to set-up kerberos token first
-
Message received on unknown topic: errcode: Broker: Group authorization failed error: FindCoordinator response error: Group authorization failed.- Resolution: Kafka broker running on windows are case sensitive to kerberos token. When creating Kerberos token with kinit, make sure to use principal name with case sensitive user id.
-
authentication: SASL handshake failed (start (-4)): SASL(-4): no mechanism available: No worthy mechs found (after 0ms in state AUTH_REQ)- Resolution: cyrus-sasl-gssapi needs to be installed on the box for Kafka kerberos authentication
-
Message error on topic "an-example-topic". errcode: Broker: Topic authorization failed error: Subscribed topic not available: an-example-topic: Broker: Topic authorization failed)- Resolution: The user account does not have access to the topic
The ParquetReader adapter is a generic user adapter to stream data from Apache Parquet files as a CSP time series.
ParquetReader adapter supports only flat (non hierarchical) parquet files with all the primitive types that are supported by the CSP framework.
ParquetReader(
self,
filename_or_list,
symbol_column=None,
time_column=None,
tz=None
):
"""
:param filename_or_list: The specifier of the file/files to be read. Can be either:
- Instance of str, in which case it's interpreted os a path of single file to be read
- A callable, in which case it's interpreted as a generator function that will be called like f(starttime, endtime) where starttime and endtime
are the start and end times of the current engine run. It's expected to generate a sequence of filenames to read.
- Iterable container, for example a list of files to read
:param symbol_column: An optional parameter that specifies the name of the symbol column if the file if there is any
:param time_column: A mandatory specification of the time column name in the parquet files. This column will be used to inject the row values
from parquet at the given timestamps.
:param tz: The pytz timezone of the timestamp column, should only be provided if the time_column in parquet file doesn't have tz info.
"""def subscribe(
self,
symbol,
typ,
field_map=None,
push_mode: csp.PushMode = csp.PushMode.NON_COLLAPSING
):
"""Subscribe to the rows corresponding to a given symbol
This form of subscription can be used only if non empty symbol_column was supplied during ParquetReader construction.
:param symbol: The symbol to subscribe to, for example 'AAPL'
:param typ: The type of the CSP time series subscription. Can either be a primitive type like int or alternatively a type
that inherits from csp.Struct, in which case each instance of the struct will be constructed from the matching file columns.
:param field_map: A map of the fields from parquet columns for the CSP time series. If typ is a primitive, then field_map should be
a string specifying the column name, if typ is a csp Struct then field_map should be a str->str dictionary of the form
{column_name:struct_field_name}. For structs field_map can be omitted in which case we expect a one to one match between the given Struct
fields and the parquet files columns.
:param push_mode: A push mode for the output adapter
"""
def subscribe_all(
self,
typ,
field_map=None,
push_mode: csp.PushMode = csp.PushMode.NON_COLLAPSING
):
"""Subscribe to all rows of the input files.
:param typ: The type of the CSP time series subscription. Can either be a primitive type like int or alternatively a type
that inherits from csp.Struct, in which case each instance of the struct will be constructed from the matching file columns.
:param field_map: A map of the fields from parquet columns for the CSP time series. If typ is a primitive, then field_map should be
a string specifying the column name, if typ is a csp Struct then field_map should be a str->str dictionary of the form
{column_name:struct_field_name}. For structs field_map can be omitted in which case we expect a one to one match between the given Struct
fields and the parquet files columns.
:param push_mode: A push mode for the output adapter
"""Parquet reader provides two subscription methods.
subscribe produces a time series only of the rows that correspond to the given symbol,
**subscribe_all**produces a time series of all rows in the parquet files.
The ParquetWriter adapter is a generic user adapter to stream data from CSP time series to Apache Parquet files.
ParquetWriter adapter supports only flat (non hierarchical) parquet files with all the primitive types that are supported by the csp framework.
Any time series of Struct objects will be flattened to multiple columns.
ParquetWriter(
self,
file_name: Optional[str],
timestamp_column_name,
config: Optional[ParquetOutputConfig] = None,
filename_provider: Optional[csp.ts[str]] = None
):
"""
:param file_name: The path of the output parquet file name. Must be provided if no filename_provider specified. If both file_name and filename_provider are specified then file_name will be used as the initial output file name until filename_provider provides a new file name.
:param timestamp_column_name: Required field, if None is provided then no timestamp will be written.
:param config: Optional configuration of how the file should be written (such as compression, block size,...).
:param filename_provider: An optional time series that provides a times series of file paths. When a filename_provider time series provides a new file path, the previous open file name will be closed and all subsequent data will be written to the new file provided by the path. This enable partitioning and splitting the data based on time.
"""def publish_struct(
self,
value: ts[csp.Struct],
field_map: Dict[str, str] = None
):
"""Publish a time series of csp.Struct objects to file
:param value: The time series of Struct objects that should be published.
:param field_map: An optional dict str->str of the form {struct_field_name:column_name} that maps the names of the
structure fields to the column names to which the values should be written. If the field_map is non None, then only
the fields that are specified in the field_map will be written to file. If field_map is not provided then all fields
of a structure will be written to columns that match exactly the field_name.
"""
def publish(
self,
column_name,
value: ts[object]
):
"""Publish a time series of primitive type to file
:param column_name: The name of the parquet file column to which the data should be written to
:param value: The time series that should be published
"""Parquet writer provides two publishing methods.
publish_struct is used to publish time series of csp.Struct objects while publish is used to publish primitive time series.
The columns in the written parquet file is a union of all columns that were published (the order is preserved).
A new row is written to parquet file whenever any of the inputs ticks.
For the given row, any column that corresponds to a time series that didn't tick, will have null values.
import tempfile
from datetime import datetime, timedelta
import csp
from csp.adapters.parquet import ParquetOutputConfig, ParquetReader, ParquetWriter
class Dummy(csp.Struct):
int_val: int
float_val: float
@csp.graph
def write_struct(file_name: str):
st = datetime(2020, 1, 1)
curve = csp.curve(Dummy, [(st + timedelta(seconds=1), Dummy(int_val=1, float_val=1.0)),
(st + timedelta(seconds=2), Dummy(int_val=2, float_val=2.0)),
(st + timedelta(seconds=3), Dummy(int_val=3, float_val=3.0))])
writer = ParquetWriter(file_name=file_name, timestamp_column_name='csp_time',
config=ParquetOutputConfig(allow_overwrite=True))
writer.publish_struct(curve)
@csp.graph
def write_series(file_name: str):
st = datetime(2020, 1, 1)
curve_int = csp.curve(int, [(st + timedelta(seconds=i), i * 5) for i in range(10)])
curve_str = csp.curve(str, [(st + timedelta(seconds=i), f'str_{i}') for i in range(10)])
writer = ParquetWriter(file_name=file_name, timestamp_column_name='csp_time',
config=ParquetOutputConfig(allow_overwrite=True))
writer.publish('int_vals', curve_int)
writer.publish('str_vals', curve_str)
@csp.graph
def writer_graph(struct_file_name: str, series_file_name: str):
write_struct(struct_file_name)
write_series(series_file_name)
@csp.graph
def reader_graph(series_file_name: str):
reader = ParquetReader(series_file_name, time_column='csp_time')
csp.print('Read as struct', reader.subscribe_all(Dummy))
csp.print('Read as single int column', reader.subscribe_all(int, 'int_val'))
csp.print('Read as single float column', reader.subscribe_all(float, 'float_val'))
if __name__ == '__main__':
with tempfile.NamedTemporaryFile(suffix='.parquet') as struct_file:
struct_file.file.close()
with tempfile.NamedTemporaryFile(suffix='.parquet') as series_file:
series_file.file.close()
g = csp.run(writer_graph, struct_file.name, series_file.name,
starttime=datetime(2020, 1, 1), endtime=timedelta(minutes=1))
g = csp.run(reader_graph, struct_file.name,
starttime=datetime(2020, 1, 1), endtime=timedelta(minutes=1))
The DBReader adapter is a generic user adapter to stream data from a database as a reactive time series. It leverages sqlalchemy internally in order to be able to access various DB backends.
Please refer to the SQLAlchemy Docs for information on how to create sqlalchemy connections.
The DBReader instance represents a single connection to a database. From a single reader you can subscribe to various streams, either the entire stream of data (which would basically represent the result of a single join) or if a symbol column is declared, subscribe by symbol which will then demultiplex rows to the right adapter.
DBReader(self, connection, time_accessor, table_name=None, schema_name=None, query=None, symbol_column=None, constraint=None):
"""
:param connection: sqlalchemy engine or (already connected) connection object.
:param time_accessor: TimeAccessor object
:param table_name: name of table in database as a string
:param query: either string query or sqlalchemy query object. Ex: "select * from users"
:param symbol_column: name of symbol column in table as a string
:param constraint: additional sqlalchemy constraints for query. Ex: constraint = db.text('PRICE>:price').bindparams(price = 100.0)
"""- connection: seqlalchemy engine or existing connection object.
- time_accessor: see below
- table_name: either table or query is required. If passing a table_name then this table will be queried against for subscribe calls
- query: (optional) if table isn't supplied user can provide a direct query string or sqlalchemy query object. This is useful if you want to run a join call. For basic single-table queries passing table_name is preferred
-
symbol_column: (optional) in order to be able to demux rows bysome column, pass
symbol_column. Example case for this is if database has data stored for many symbols in a single table, and you want to have a timeseries tick per symbol. -
constraint: (optional) additional sqlalchemy constraints for query. Ex:
constraint = db.text('PRICE>:price').bindparams(price= 100.0)
All data fed into csp must be time based.
TimeAccessor is a helper class that defines how to extract timestamp information from the results of the data.
Users can define their own TimeAccessor implementation or use pre-canned ones:
-
TimestampAccessor( self, time_column, tz=None): use this if there exists a single datetime column already. Provide the column name and optionally the timezone of the column (if its timezone-less in the db) -
DateTimeAccessor(self, date_column, time_column, tz=None): use this if there are two separate columns for date and time, this accessor will combine the two columns to create a single datetime. Optionally pass tz if time column is timezone-less in the db
User implementations would have to extend TimeAccessor interface.
In addition to defining how to convert db columns to timestamps, accessors are also used to augment the query to limit the data for the graph's start and end times.
Once you have a DBReader object created, you can subscribe to time_series from it using the following methods:
subscribe(self, symbol, typ, field_map=None)subscribe_all(self, typ, field_map=None)
Both of these calls expect typ to be a csp.Struct type.
field_map is a dictionary of { db_column : struct_column } mappings that define how to map the database column names to the fields on the struct.
subscribe is used to subscribe to a stream for the given symbol (symbol_column is required when creating DBReader)
subscribe_all is used to retrieve all the data resulting from the request as a single timeseries.
The Symphony adapter allows for reading and writing of messages from the Symphony message platform using requests and the Symphony SDK.
The Slack adapter allows for reading and writing of messages from the Slack message platform using the Slack Python SDK.
There are two main categories of writing input adapters, historical and realtime. When writing historical adapters you will need to implement a "pull" adapter, which pulls data from a historical data source in time order, one event at a time. There are also ManagedSimAdapters for feeding multiple "managed" pull adapters from a single source (more on that below). When writing realtime adapters, you will need to implement a "push" adapter, which will get data from a separate thread that drives external events and "pushes" them into the engine as they occur.
When writing input adapters it is also very important to denote the difference between "graph building time" and "runtime" versions of your adapter.
For example, csp.adapters.csv has a CSVReader class that is used at graph building time.
Graph build time components solely describe the adapter.
They are meant to do little else than keep track of the type of adapter and its parameters, which will then be used to construct the actual adapter implementation when the engine is constructed from the graph description.
It is the runtime implementation that actual runs during the engine execution phase to process data.
For clarity of this distinction, in the descriptions below we will denote graph build time components with --graph-- and runtime implementations with --impl--.
There are two flavors of historical input adapters that can be written. The simplest one is a PullInputAdapter. A PullInputAdapter can be used to convert a single source into a single timeseries. The csp.curve implementation is a good example of this. Single source to single timeseries adapters are of limited use however, and the more typical use case is for AdapterManager based input adapters to service multiple InputAdapters from a single source. For this one would use an AdapterManager to coordinate processing of the data source, and ManagedSimInputAdapter as the individual timeseries providers.
To write a Python based PullInputAdapter one must write a class that derives from csp.impl.pulladapter.PullInputAdapter. The derived type should the define two methods:
-
def start(self, start_time, end_time): this will be called at the start of the engine with the start/end times of the engine. start_Time and end_time will be tz-unaware datetime objects in UTC time. At this point the adapter should open its resource and seek to the requested starttime. -
def next(self): this method will be repeatedly called by the engine. The adapter should return the next event as a time,value tuple. If there are no more events, then the method should return None
The PullInputAdapter that you define will be used as the runtime --impl–-. You also need to define a --graph-- time representation of the time series edge. In order to do this you should define a csp.impl.wiring.py_pull_adapter_def. The py_pull_adapter_def creates a --graph-- time representation of your adapter:
def py_pull_adapter_def(name, adapterimpl, out_type, **kwargs)-
name: string name for the adapter -
adapterimpl: a derived implementation of csp.impl.pulladapter.PullInputAdapter -
out_type: the type of the output, should be ats[]type. Note this can use tvar types if a subsequent argument defines the tvar -
kwargs: **kwargs here be passed through as arguments to the PullInputAdapter implementation
Note that the **kwargs passed to py_pull_adapter_def should be the names and types of the variables, like arg1=type1, arg2=type2. These are the names of the kwargs that the returned input adapter will take and pass through to the PullInputAdapter implementation, and the types expected for the values of those args.
csp.curve is a good simple example of this:
import copy
from csp.impl.pulladapter import PullInputAdapter
from csp.impl.wiring import py_pull_adapter_def
from csp import ts
from datetime import timedelta
class Curve(PullInputAdapter):
def __init__(self, typ, data):
''' data should be a list of tuples of (datetime, value) or (timedelta, value)'''
self._data = data
self._index = 0
super().__init__()
def start(self, start_time, end_time):
if isinstance(self._data[0][0], timedelta):
self._data = copy.copy(self._data)
for idx, data in enumerate(self._data):
self._data[idx] = (start_time + data[0], data[1])
while self._index < len(self._data) and self._data[self._index][0] < start_time:
self._index += 1
super().start(start_time, end_time)
def next(self):
if self._index < len(self._data):
time, value = self._data[self._index]
if time <= self._end_time:
self._index += 1
return time, value
return None
curve = py_pull_adapter_def('curve', Curve, ts['T'], typ='T', data=list)Now curve can be called in graph code to create a curve input adapter:
x = csp.curve(int, [ (t1, v1), (t2, v2), .. ])
csp.print('x', x)See example "e_14_user_adapters_01_pullinput.py for
Step 1) PullInputAdapter impl
Similar to the Python PullInputAdapter API is the c++ API which one can leverage to improve performance of an adapter implementation.
The --impl-- is very similar to python pull adapter.
One should derive from PullInputAdapter<T>, a templatized base class (templatized on the type of the timeseries) and define these methods:
-
start(DateTime start, DateTime end): similar to python API start, called when engine starts. Open resource and seek to start time here -
stop(): called on engine shutdown, cleanup resource -
bool next(DateTime & t, T & value): if there is data to provide, sets the next time and value for the adapter and returns true. Otherwise, return false
Step 2) Expose creator func to python
Now that we have a c++ impl defined, we need to expose a python creator for it. Define a method that conforms to the signature
csp::InputAdapter * create_my_adapter(
csp::AdapterManager * manager,
PyEngine * pyengine,
PyTypeObject * pyType,
PushMode pushMode,
PyObject * args)-
manager: will be nullptr for pull adapters -
pyengine: PyEngine engine wrapper object -
pyType: this is the type of the timeseries input adapter to be created as a PyTypeObject. one can switch on this type using switchPyType to create the properly typed instance -
pushMode: the csp PushMode for the adapter (pass through to base InputAdapter) -
args: arguments to pass to the adapter impl
Then simply register the creator method:
REGISTER_INPUT_ADAPTER(_my_adapter, create_my_adapter)
This will register methodname onto your python module, to be accessed as your module.methodname. Note this uses csp/python/InitHelpers which is used in the _cspimpl module. To do this in a separate python module, you need to register InitHelpers in that module.
Step 3) Define your --graph–- time adapter
One liner now to wrap your impl in a graph time construct using csp.impl.wiring.input_adapter_def:
my_adapter = input_adapter_def('my_adapter', my_module._my_adapter, ts[int], arg1=int, arg2={str:'foo'})my_adapter can now be called with arg1, arg2 to create adapters in your graph. Note that the arguments are typed using v=t syntax. v=(t,default) is used to define arguments with defaults.
Also note that all input adapters implicitly get a push_mode argument that is defaulted to csp.PushMode.LAST_VALUE.
In most cases you will likely want to expose a single source of data into multiple input adapters. For this use case your adapter should define an AdapterManager --graph-- time component, and AdapterManagerImpl --impl-- runtime component. The AdapterManager --graph-- time component just represents the parameters needed to create the --impl-- AdapterManager. Its the --impl-- that will have the actual implementation that will open the data source, parse the data and provide it to individual Adapters.
Similarly you will need to define a derived ManagedSimInputAdapter --impl-- component to handle events directed at an individual time series adapter.
NOTE It is highly recommended not to open any resources in the --graph-- time component. graph time components can be pruned and/or memoized into a single instance, opening resources at graph time shouldn't be necessary.
The graph time AdapterManager doesn't need to derive from any interface. It should be initialized with any information the impl needs in order to open/process the data source (ie csv file, time column, db connection information, etc etc). It should also have an API to create individual timeseries adapters. These adapters will then get passed the adapter manager --impl-- as an argument where they are created, so that they can register themselves for processing. The AdapterManager also needs to define a _create method. The _create is the bridge between the --graph-- time AdapterManager representation and the runtime --impl-- object. _create will be called on the --graph-- time AdapterManager which will in turn create the --impl-- instance. _create will get two arguments, engine (this represents the runtime engine object that will run the graph) and a memo dict which can optionally be used for any memoization that on might want.
Lets take a look at CSVReader as an example:
# GRAPH TIME
class CSVReader:
def __init__(self, filename, time_converter, delimiter=',', symbol_column=None):
self._filename = filename
self._symbol_column = symbol_column
self._delimiter = delimiter
self._time_converter = time_converter
def subscribe(self, symbol, typ, field_map=None):
return CSVReadAdapter(self, symbol, typ, field_map)
def _create(self, engine, memo):
return CSVReaderImpl(engine, self)-
__init__: as you can see, all__init__does is keep the parameters that the impl will need. -
subscribe: API to create an individual timeseries / edge from this file for the given symbol. typ denotes the type of the timeseries to create (iets[int]) and field_map is used for mapping columns onto csp.Struct types. Note that subscribe returns a CSVReadAdapter instance. CSVReadAdapter is the --graph-- time representation of the edge (similar to how we defined csp.curve above). We pass itselfas its first argument, which will be used to create the AdapterManager --impl-- -
\_create: the method to create the --impl-- object from the given --graph-- time representation of the manager
The CSVReader would then be used in graph building code like so:
reader = CSVReader('my_data.csv', time_formatter, symbol_column='SYMBOL', delimiter='|')
# aapl will represent a ts[PriceQuantity] edge that will tick with rows from
# the csv file matching on SYMBOL column AAPL
aapl = reader.subscribe('AAPL', PriceQuantity)The AdapterManager --impl-- is responsible for opening the data source, parsing and processing through all the data and managing all the adapters it needs to feed. The impl class should derive from csp.impl.adaptermanager.AdapterManagerImpl and implement the following methods:
-
start(self,starttime,endtime): this is called when the engine starts up. At this point the impl should open the resource providing the data and seek to starttime. starttime/endtime will be tz-unaware datetime objects in UTC time -
stop(self): this is called at the end of the run, resources should be cleaned up at this point -
process_next_sim_timeslice(self, now): this method will be called multiple times through the run. The initial call will provide now with starttime. The impl's responsibility is to process all data at the given timestamp (more on how to do this below). The method should return the next time in the data source, or None if there is no more data to process. The method will be called again with the provided timestamp as "now" in the next iteration. NOTE that process_next_sim_timeslice is required to move ahead in time. In most cases the resource data can be supplied in time order, if not it would have to be sorted up front.
process_next_sim_timeslice should parse data for a given time/row of data and then push it through to any registered ManagedSimInputAdapter that matches on the given row
Users will need to define ManagedSimInputAdapter derived types to represent the individual timeseries adapter --impl-- objects. Objects should derive from csp.impl.adaptermanager.ManagedSimInputAdapter.
ManagedSimInputAdapter.__init__ takes two arguments:
-
typ: this is the type of the timeseries, ie int for ats[int] -
field_map: Optional, field_map is a dictionary used to map source column names → csp.Struct field names.
ManagedSimInputAdapter defines a method push_tick() which takes the value to feed the input for given timeslice (as defined by "now" at the adapter manager level).
There is also a convenience method called process_dict() which will take a dictionary of {column : value} entries and convert it properly into the right value based on the given field_map.
As with the csp.curve example, we need to define a graph-time construct that represents a ManagedSimInputAdapter edge. In order to define this we use py_managed_adapter_def. py_managed_adapter_defis AdapterManager "aware" and will properly create the AdapterManager --impl-- the first time its encountered. It will then pass the manager impl as an argument to the ManagedSimInputAdapter.
def py_managed_adapter_def(name, adapterimpl, out_type, manager_type, **kwargs):
"""
Create a graph representation of a python managed sim input adapter.
:param name: string name for the adapter
:param adapterimpl: a derived implementation of csp.impl.adaptermanager.ManagedSimInputAdapter
:param out_type: the type of the output, should be a ts[] type. Note this can use tvar types if a subsequent argument defines the tvar
:param manager_type: the type of the graph time representation of the AdapterManager that will manage this adapter
:param kwargs: **kwargs will be passed through as arguments to the ManagedSimInputAdapter implementation
the first argument to the implementation will be the adapter manager impl instance
"""Putting this all together lets take a look at a CSVReader implementation and step through what's going on:
import csv as pycsv
from datetime import datetime
from csp import ts
from csp.impl.adaptermanager import AdapterManagerImpl, ManagedSimInputAdapter
from csp.impl.wiring import pymanagedadapterdef
# GRAPH TIME
class CSVReader:
def __init__(self, filename, time_converter, delimiter=',', symbol_column=None):
self._filename = filename
self._symbol_column = symbol_column
self._delimiter = delimiter
self._time_converter = time_converter
def subscribe(self, symbol, typ, field_map=None):
return CSVReadAdapter(self, symbol, typ, field_map)
def _create(self, engine, memo):
return CSVReaderImpl(engine, self)Here we define CSVReader, our AdapterManager --graph-- time representation.
It holds the parameters that will be used for the impl, it implements a subscribe() call for users to create timeseries and defines a _create method to create a runtime --impl–- instance from the graphtime representation.
Note how on line 17 we pass self to the CSVReadAdapter, this is what binds the input adapter to this AdapterManager
# RUN TIME
class CSVReaderImpl(AdapterManagerImpl): # 1
def __init__(self, engine, adapterRep): # 2
super().__init__(engine) # 3
# 4
self._rep = adapterRep # 5
self._inputs = {} # 6
self._csv_reader = None # 7
self._next_row = None # 8
# 9
def start(self, starttime, endtime): # 10
self._csv_reader = pycsv.DictReader( # 11
open(self._rep._filename, 'r'), # 12
delimiter=self._rep._delimiter # 13
) # 14
self._next_row = None # 15
# 16
for row in self._csv_reader: # 17
time = self._rep._time_converter(row) # 18
self._next_row = row # 19
if time >= starttime: # 20
break # 21
# 22
def stop(self): # 23
self._csv_reader = None # 24
# 25
def register_input_adapter(self, symbol, adapter): # 26
if symbol not in self._inputs: # 27
self._inputs[symbol] = [] # 28
self._inputs[symbol].append(adapter) # 29
# 30
def process_next_sim_timeslice(self, now): # 31
if not self._next_row: # 32
return None # 33
# 34
while True: # 35
time = self._rep._time_converter(self._next_row) # 36
if time > now: # 37
return time # 38
self.process_row(self._next_row) # 39
try: # 40
self._next_row = next(self._csv_reader) # 41
except StopIteration: # 42
return None # 43
# 44
def process_row(self, row): # 45
symbol = row[self._rep._symbol_column] # 46
if symbol in self._inputs: # 47
for input in self._inputs.get(symbol, []): # 48
input.process_dict(row) # 49CSVReaderImpl is the runtime --impl–-. It gets created when the engine is being built from the described graph.
-
lines 10-21 - start(): this is the start method that gets called with the time range the graph will be run against. Here we open our resource (pycsv.DictReader) and scan t through the data until we reach the requested starttime.
-
lines 23-24 - stop(): this is the stop call that gets called when the engine is done running and is shutdown, we free our resource here
-
lines 26-29: the CSVReader allows one to subscribe to many symbols from one file. symbols are keyed by a provided SYMBOL column. The individual adapters will self-register with the CSVReaderImpl when they are created with the requested symbol. CSVReaderImpl keeps track of what adapters have been registered for what symbol in its self._inputs map
-
lines 31-43: this is main method that gets invoked repeatedly throughout the run. For every distinct timestamp in the file, this method will get invoked once and the method is expected to go through the resource data for all points with time now, process the row and push the data to any matching adapters. The method returns the next timestamp when its done processing all data for "now", or None if there is no more data. NOTE that the csv impl expects the data to be in time order. process_next_sim_timeslice must advance time forward.
-
lines 45-49: this method takes a row of data (provided as a dict from DictReader), extracts the symbol and pushes the row through to all input adapters that match
class CSVReadAdapterImpl(ManagedSimInputAdapter): # 1
def __init__(self, managerImpl, symbol, typ, field_map): # 2
managerImpl.register_input_adapter(symbol, self) # 3
super().__init__(typ, field_map) # 4
# 5
CSVReadAdapter = py_managed_adapter_def( # 6
'csvadapter',
CSVReadAdapterImpl,
ts['T'],
CSVReader,
symbol=str,
typ='T',
fieldMap=(object, None)
)- line 3: this is where the instance of an adapter --impl-- registers itself with the CSVReaderImpl.
- line 6+: this is where we define CSVReadAdapter, the --graph-- time representation of a CSV adapter, returned from CSVReader.subscribe
See example "e_14_user_adapters_02_adaptermanager_siminput" for another example of how to write a managed sim adapter manager.
To write a Python based PushInputAdapter one must write a class that derives from csp.impl.pushadapter.PushInputAdapter. The derived type should the define two methods:
-
def start(self, start_time, end_time): this will be called at the start of the engine with the start/end times of the engine. start_time and end_time will be tz-unaware datetime objects in UTC time (generally these aren't needed for realtime adapters). At this point the adapter should open its resource / connect the data source / start any driver threads that are needed. -
def stop(self): This method well be called when the engine is done running. At this point any open threads should be stopped and resources cleaned up.
The PushInputAdapter that you define will be used as the runtime --impl–-. You also need to define a --graph-- time representation of the time series edge. In order to do this you should define a csp.impl.wiring.py_push_adapter_def. The py_push_adapter_def creates a --graph-- time representation of your adapter:
def py_push_adapter_def(name, adapterimpl, out_type, **kwargs)
-
name: string name for the adapter -
adapterimpl: a derived implementation of csp.impl.pushadapter.PushInputAdapter -
out_type: the type of the output, should be a ts[] type. Note this can use tvar types if a subsequent argument defines the tvar -
kwargs: **kwargs here be passed through as arguments to the PushInputAdapter implementation
Note that the **kwargs passed to py_push_adapter_def should be the names and types of the variables, like arg1=type1, arg2=type2. These are the names of the kwargs that the returned input adapter will take and pass through to the PushInputAdapter implementation, and the types expected for the values of those args.
Example e_14_user_adapters_03_pushinput.py demonstrates a simple example of this
from csp.impl.pushadapter import PushInputAdapter
from csp.impl.wiring import py_push_adapter_def
import csp
from csp import ts
from datetime import datetime, timedelta
import threading
import time
# The Impl object is created at runtime when the graph is converted into the runtime engine
# it does not exist at graph building time!
class MyPushAdapterImpl(PushInputAdapter):
def __init__(self, interval):
print("MyPushAdapterImpl::__init__")
self._interval = interval
self._thread = None
self._running = False
def start(self, starttime, endtime):
""" start will get called at the start of the engine, at which point the push
input adapter should start its thread that will push the data onto the adapter. Note
that push adapters will ALWAYS have a separate thread driving ticks into the csp engine thread
"""
print("MyPushAdapterImpl::start")
self._running = True
self._thread = threading.Thread(target=self._run)
self._thread.start()
def stop(self):
""" stop will get called at the end of the run, at which point resources should
be cleaned up
"""
print("MyPushAdapterImpl::stop")
if self._running:
self._running = False
self._thread.join()
def _run(self):
counter = 0
while self._running:
self.push_tick(counter)
counter += 1
time.sleep(self._interval.total_seconds())
# MyPushAdapter is the graph-building time construct. This is simply a representation of what the
# input adapter is and how to create it, including the Impl to create and arguments to pass into it
MyPushAdapter = py_push_adapter_def('MyPushAdapter', MyPushAdapterImpl, ts[int], interval=timedelta)Note how line 41 calls self.push_tick. This is the call to get data from the adapter thread ticking into the csp engine
Now MyPushAdapter can be called in graph code to create a timeseries that is sourced by MyPushAdapterImpl
@csp.graph
def my_graph():
# At this point we create the graph-time representation of the input adapter. This will be converted
# into the impl once the graph is done constructing and the engine is created in order to run
data = MyPushAdapter(timedelta(seconds=1))
csp.print('data', data)If you dont need as much control as PushInputAdapter provides, or if you have some existing source of data on a thread you can't control, another option is to use the higher-level abstraction csp.GenericPushAdapter. csp.GenericPushAdapter wraps a csp.PushInputAdapter implementation internally and provides a simplified interface. The downside of csp.GenericPushAdapter is that you lose some control of when the input feed starts and stop.
Lets take a look at the example found in "e_14_generic_push_adapter"
# This is an example of some separate thread providing data
class Driver:
def __init__(self, adapter : csp.GenericPushAdapter):
self._adapter = adapter
self._active = False
self._thread = None
def start(self):
self._active = True
self._thread = threading.Thread(target=self._run)
self._thread.start()
def stop(self):
if self._active:
self._active = False
self._thread.join()
def _run(self):
print("driver thread started")
counter = 0
# Optionally, we can wait for the adapter to start before proceeding
# Alternatively we can start pushing data, but push_tick may fail and return False if
# the csp engine isn't ready yet
self._adapter.wait_for_start()
while self._active and not self._adapter.stopped():
self._adapter.push_tick(counter)
counter += 1
time.sleep(1)
@csp.graph
def my_graph():
adapter = csp.GenericPushAdapter(int)
driver = Driver(adapter)
# Note that the driver thread starts *before* the engine is started here, which means some ticks may potentially get dropped if the
# data source doesn't wait for the adapter to start. This may be ok for some feeds, but not others
driver.start()
# Lets be nice and shutdown the driver thread when the engine is done
csp.schedule_on_engine_stop(driver.stop)In this example we have this dummy Driver class which simply represents some external source of data which arrives on a thread that's completely independent of the engine. We pass along a csp.GenericInputAdapter instance to this thread, which can then call adapter.push_tick to get data into the engine (see line 27).
On line 24 we can also see an optional feature which allows the unrelated thread to wait for the adapter to be ready to accept data before ticking data onto it. If push_tick is called before the engine starts / the adapter is ready to receive data, it will simply drop the data. Note that GenericPushAadapter.push_tick will return a bool to indicate whether the data was successfully pushed to the engine or not.
In most cases you will likely want to expose a single source of data into multiple input adapters. For this use case your adapter should define an AdapterManager --graph-- time component, and AdapterManagerImpl --impl-- runtime component. The AdapterManager --graph-- time component just represents the parameters needed to create the --impl-- AdapterManager. Its the --impl-- that will have the actual implementation that will open the data source, parse the data and provide it to individual Adapters.
Similarly you will need to define a derived PushInputAdapter --impl-- component to handle events directed at an individual time series adapter.
NOTE It is highly recommended not to open any resources in the --graph-- time component. Graph time components can be pruned and/or memoized into a single instance, opening resources at graph time shouldn't be necessary.
The graph time AdapterManager doesn't need to derive from any interface. It should be initialized with any information the impl needs in order to open/process the data source (ie activemq connection information, server host port, multicast channels, config files, etc etc). It should also have an API to create individual timeseries adapters. These adapters will then get passed the adapter manager --impl-- as an argument when they are created, so that they can register themselves for processing. The AdapterManager also needs to define a _create method. The _create is the bridge between the --graph-- time AdapterManager representation and the runtime --impl-- object. _create will be called on the --graph-- time AdapterManager which will in turn create the --impl-- instance. _create will get two arguments, engine (this represents the runtime engine object that will run the graph) and memo dict which can optionally be used for any memoization that on might want.
Lets take a look at the example found in "e_14_user_adapters_04_adaptermanager_pushinput"
# This object represents our AdapterManager at graph time. It describes the manager's properties
# and will be used to create the actual impl when its time to build the engine
class MyAdapterManager:
def __init__(self, interval: timedelta):
"""
Normally one would pass properties of the manager here, ie filename,
message bus, etc
"""
self._interval = interval
def subscribe(self, symbol, push_mode=csp.PushMode.NON_COLLAPSING):
""" User facing API to subscribe to a timeseries stream from this adapter manager """
# This will return a graph-time timeseries edge representing and edge from this
# adapter manager for the given symbol / arguments
return MyPushAdapter(self, symbol, push_mode=push_mode)
def _create(self, engine, memo):
""" This method will get called at engine build time, at which point the graph time manager representation
will create the actual impl that will be used for runtime
"""
# Normally you would pass the arguments down into the impl here
return MyAdapterManagerImpl(engine, self._interval)- __init__ - as you can see, all __init__ does is keep the parameters that the impl will need.
- subscribe - API to create an individual timeseries / edge from this file for the given symbol. The interface defined here is up to the adapter writer, but generally "subscribe" is recommended, and it should take any number of arguments needed to define a single stream of data. MyPushAdapter is the --graph-- time representation of the edge, which will be described below. We pass it self as its first argument, which will be used to create the AdapterManager --impl--
- _create - the method to create the --impl-- object from the given --graph-- time representation of the manager
MyAdapterManager would then be used in graph building code like so:
adapter_manager = MyAdapterManager(timedelta(seconds=0.75))
data = adapter_manager.subscribe('AAPL', push_mode=csp.PushMode.LAST_VALUE)
csp.print(symbol + " last_value", data)The AdapterManager --impl-- is responsible for opening the data source, parsing and processing all the data and managing all the adapters it needs to feed. The impl class should derive from csp.impl.adaptermanager.AdapterManagerImpl and implement the following methods:
- start(self,starttime,endtime): this is called when the engine starts up. At this point the impl should open the resource providing the data and start up any thread(s) needed to listen to and react to external data. starttime/endtime will be tz-unaware datetime objects in UTC time, though typically these aren't needed for realtime adapters
-
stop(self): this is called at the end of the run, resources should be cleaned up at this point -
process_next_sim_timeslice(self, now): this is used by sim adapters, for realtime adapter managers we simply return None
In the example manager, we spawn a processing thread in the start() call.
This thread runs in a loop until it is shutdown, and will generate random data to tick out to the registered input adapters.
Data is passed to a given adapter by calling push_tick()
Users will need to define PushInputAdapter derived types to represent the individual timeseries adapter --impl-- objects. Objects should derive from csp.impl.pushadapter.PushInputAdapter.
PushInputAdapter defines a method push_tick() which takes the value to feed the input timeseries.
Similar to the stand alone PushInputAdapter described above, we need to define a graph-time construct that represents a PushInputAdapter edge. In order to define this we use py_push_adapter_def again, but this time we pass the adapter manager --graph-- time type so that it gets constructed properly. When the PushInputAdapter instance is created it will also receive an instance of the adapter manager --impl–-, which it can then self-register on/
def py_push_adapter_def (name, adapterimpl, out_type, manager_type=None, memoize=True, force_memoize=False, **kwargs):
"""
Create a graph representation of a python push input adapter.
:param name: string name for the adapter
:param adapterimpl: a derived implementation of csp.impl.pushadapter.PushInputAdapter
:param out_type: the type of the output, should be a ts[] type. Note this can use tvar types if a subsequent argument defines the tvar
:param manager_type: the type of the graph time representation of the AdapterManager that will manage this adapter
:param kwargs: **kwargs will be passed through as arguments to the ManagedSimInputAdapter implementation
the first argument to the implementation will be the adapter manager impl instance
"""Continuing with the --graph-- time AdapterManager described above, we now define the impl:
# This is the actual manager impl that will be created and executed during runtime
class MyAdapterManagerImpl(AdapterManagerImpl):
def __init__(self, engine, interval):
super().__init__(engine)
# These are just used to simulate a data source
self._interval = interval
self._counter = 0
# We will keep track of requested input adapters here
self._inputs = {}
# Our driving thread, all realtime adapters will need a separate thread of execution that
# drives data into the engine thread
self._running = False
self._thread = None
def start(self, starttime, endtime):
""" start will get called at the start of the engine run. At this point
one would start up the realtime data source / spawn the driving thread(s) and
subscribe to the needed data """
self._running = True
self._thread = threading.Thread(target=self._run)
self._thread.start()
def stop(self):
""" This will be called at the end of the engine run, at which point resources should be
closed and cleaned up """
if self._running:
self._running = False
self._thread.join()
def register_input_adapter(self, symbol, adapter):
""" Actual PushInputAdapters will self register when they are created as part of the engine
This is the place we gather all requested input adapters and their properties
"""
if symbol not in self._inputs:
self._inputs[symbol] = []
# Keep a list of adapters by key in case we get duplicate adapters (should be memoized in reality)
self._inputs[symbol].append(adapter)
def process_next_sim_timeslice(self, now):
""" This method is only used by simulated / historical adapters, for realtime we just return None """
return None
def _run(self):
""" Our driving thread, in reality this will be reacting to external events, parsing the data and
pushing it into the respective adapter
"""
symbols = list(self._inputs.keys())
while self._running:
# Lets pick a random symbol from the requested symbols
symbol = symbols[random.randint(0, len(symbols) - 1)]
adapters = self._inputs[symbol]
data = MyData(symbol=symbol, value=self._counter)
self._counter += 1
for adapter in adapters:
adapter.push_tick(data)
time.sleep(self._interval.total_seconds())Then we define our PushInputAdapter --impl--, which basically just
self-registers with the adapter manager --impl-- upon construction. We
also define our PushInputAdapter --graph-- time construct using py_push_adapter_def.
# The Impl object is created at runtime when the graph is converted into the runtime engine
# it does not exist at graph building time. a managed sim adapter impl will get the
# adapter manager runtime impl as its first argument
class MyPushAdapterImpl(PushInputAdapter):
def __init__(self, manager_impl, symbol):
print(f"MyPushAdapterImpl::__init__ {symbol}")
manager_impl.register_input_adapter(symbol, self)
super().__init__()
MyPushAdapter = py_push_adapter_def('MyPushAdapter', MyPushAdapterImpl, ts[MyData], MyAdapterManager, symbol=str)And then we can run our adapter in a csp graph
@csp.graph
def my_graph():
print("Start of graph building")
adapter_manager = MyAdapterManager(timedelta(seconds=0.75))
symbols = ['AAPL', 'IBM', 'TSLA', 'GS', 'JPM']
for symbol in symbols:
# your data source might tick faster than the engine thread can consume it
# push_mode can be used to buffered up tick events will get processed
# LAST_VALUE will conflate and only tick the latest value since the last cycle
data = adapter_manager.subscribe(symbol, csp.PushMode.LAST_VALUE)
csp.print(symbol + " last_value", data)
# BURST will change the timeseries type from ts[T] to ts[[T]] (list of ticks)
# that will tick with all values that have buffered since the last engine cycle
data = adapter_manager.subscribe(symbol, csp.PushMode.BURST)
csp.print(symbol + " burst", data)
# NON_COLLAPSING will tick all events without collapsing, unrolling the events
# over multiple engine cycles
data = adapter_manager.subscribe(symbol, csp.PushMode.NON_COLLAPSING)
csp.print(symbol + " non_collapsing", data)
print("End of graph building")
csp.run(my_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True)Do note that realtime adapters will only run in realtime engines (note the realtime=True argument to csp.run).
Output adapters are used to define graph outputs, and they differ from input adapters in a number of important ways.
Output adapters also differ from terminal nodes, e.g. regular csp.node instances that do not define outputs, and instead consume and emit their inputs inside their csp.ticked blocks.
For many use cases, it will be sufficient to omit writing an output adapter entirely. Consider the following example of a terminal node that writes an input dictionary timeseries to a file.
@csp.node
def write_to_file(x: ts[Dict], filename: str):
if csp.ticked(x):
with open(filename, "a") as fp:
fp.write(json.dumps(x))This is a perfectly fine node, and serves its purpose. Unlike input adapters, output adapters do not need to differentiate between historical and realtime mode. Input adapters drive the execution of the graph, whereas output adapters are reactive to their input nodes and subject to the graph's execution.
However, there are a number of reasons why you might want to define an output adapter instead of using a vanilla node. The most important of these is when you want to share resources across a number of output adapters (e.g. with a Manager), or between an input and an output node, e.g. reading data from a websocket, routing it through your csp graph, and publishing data to the same websocket connection. For most use cases, a vanilla csp node will suffice, but let's explore some anyway.
To write a Python based OutputAdapter one must write a class that derives from csp.impl.outputadapter.OutputAdapter.
The derived type should define the method:
-
def on_tick(self, time: datetime, value: object): this will be called when the input to the output adapter ticks.
The OutputAdapter that you define will be used as the runtime --impl–-. You also need to define a --graph-- time representation of the time series edge. In order to do this you should define a csp.impl.wiring.py_output_adapter_def. The py_output_adapter_def creates a --graph-- time representation of your adapter:
def py_output_adapter_def(name, adapterimpl, **kwargs)
-
name: string name for the adapter -
adapterclass: a derived implementation ofcsp.impl.outputadapter.OutputAdapter -
kwargs: **kwargs here be passed through as arguments to the OutputAdapter implementation
Note that the **kwargs passed to py_output_adapter_def should be the names and types of the variables, like arg1=type1, arg2=type2.
These are the names of the kwargs that the returned output adapter will take and pass through to the OutputAdapter implementation, and the types expected for the values of those args.
Here is a simple example of the same filewriter from above:
from csp.impl.outputadapter import OutputAdapter
from csp.impl.wiring import py_output_adapter_def
from csp import ts
import csp
from json import dumps
from datetime import datetime, timedelta
class MyFileWriterAdapterImpl(OutputAdapter):
def __init__(self, filename: str):
super().__init__()
self._filename = filename
def start(self):
self._fp = open(self._filename, "a")
def stop(self):
self._fp.close()
def on_tick(self, time, value):
self._fp.write(dumps(value) + "\n")
MyFileWriterAdapter = py_output_adapter_def(
name='MyFileWriterAdapter',
adapterimpl=MyFileWriterAdapterImpl,
input=ts['T'],
filename=str,
)Now our adapter can be called in graph code:
@csp.graph
def my_graph():
curve = csp.curve(
data=[
(timedelta(seconds=0), {"a": 1, "b": 2, "c": 3}),
(timedelta(seconds=1), {"a": 1, "b": 2, "c": 3}),
(timedelta(seconds=1), {"a": 1, "b": 2, "c": 3}),
],
typ=object,
)
MyFileWriterAdapter(curve, filename="testfile.jsonl")As explained above, we could also do this via single node (this is probably the best version between the three):
@csp.node
def dump_json(data: ts['T'], filename: str):
with csp.state():
s_file=None
with csp.start():
s_file = open(filename, "w")
with csp.stop():
s_file.close()
if csp.ticked(data):
s_file.write(json.dumps(data) + "\n")
s_file.flush()TODO
Adapter managers function the same way for output adapters as for input adapters, i.e. to manage a single shared resource from the manager across a variety of discrete output adapters.
As a as last example, lets tie everything together and implement a managed push input adapter combined with a managed output adapter.
This example is available in e_14_user_adapters_05_adaptermanager_inputoutput .
First, we will define our adapter manager. In this example, we're going to cheat a little bit and combine our adapter manager (graph time) and our adapter manager impl (run time).
class MyAdapterManager(AdapterManagerImpl):
'''
This example adapter will generate random `MyData` structs every `interval`. This simulates an upstream
data feed, which we "connect" to only a single time. We then multiplex the results to an arbitrary
number of subscribers via the `subscribe` method.
We can also receive messages via the `publish` method from an arbitrary number of publishers. These messages
are demultiplexex into a number of outputs, simulating sharing a connection to a downstream feed or responses
to the upstream feed.
'''
def __init__(self, interval: timedelta):
self._interval = interval
self._counter = 0
self._subscriptions = {}
self._publications = {}
self._running = False
self._thread = None
def subscribe(self, symbol):
'''This method creates a new input adapter implementation via the manager.'''
return _my_input_adapter(self, symbol, push_mode=csp.PushMode.NON_COLLAPSING)
def publish(self, data: ts['T'], symbol: str):
'''This method creates a new output adapter implementation via the manager.'''
return _my_output_adapter(self, data, symbol)
def _create(self, engine, memo):
# We'll avoid having a second class and make our AdapterManager and AdapterManagerImpl the same
super().__init__(engine)
return self
def start(self, starttime, endtime):
self._running = True
self._thread = threading.Thread(target=self._run)
self._thread.start()
def stop(self):
if self._running:
self._running = False
self._thread.join()
# print closing of the resources
for name in self._publications.values():
print("closing asset {}".format(name))
def register_subscription(self, symbol, adapter):
if symbol not in self._subscriptions:
self._subscriptions[symbol] = []
self._subscriptions[symbol].append(adapter)
def register_publication(self, symbol):
if symbol not in self._publications:
self._publications[symbol] = "publication_{}".format(symbol)
def _run(self):
'''This method runs in a background thread and generates random input events to push to the corresponding adapter'''
symbols = list(self._subscriptions.keys())
while self._running:
# Lets pick a random symbol from the requested symbols
symbol = symbols[random.randint(0, len(symbols) - 1)]
data = MyData(symbol=symbol, value=self._counter)
self._counter += 1
for adapter in self._subscriptions[symbol]:
# push to all the subscribers
adapter.push_tick(data)
time.sleep(self._interval.total_seconds())
def _on_tick(self, symbol, value):
'''This method just writes the data to the appropriate outbound "channel"'''
print("{}:{}".format(self._publications[symbol], value))This adapter manager is a bit of a silly example, but it demonstrates the core concepts.
The adapter manager will demultiplex a shared stream (in this case, the stream defined in _run is a random sequence of MyData structs) between all the input adapters it manages.
The input adapter itself will do nothing more than let the adapter manager know that it exists:
class MyInputAdapterImpl(PushInputAdapter):
'''Our input adapter is a very simple implementation, and just
defers its work back to the manager who is expected to deal with
sharing a single connection.
'''
def __init__(self, manager, symbol):
manager.register_subscription(symbol, self)
super().__init__()Similarly, the adapter manager will multiplex the output adapter streams, in this case combining them into streams of print statements.
And similar to the input adapter, the output adapter does relatively little more than letting the adapter manager know that it has work available, using its triggered on_tick method to call the adapter manager's _on_tick method.
class MyOutputAdapterImpl(OutputAdapter):
'''Similarly, our output adapter is simple as well, deferring
its functionality to the manager
'''
def __init__(self, manager, symbol):
manager.register_publication(symbol)
self._manager = manager
self._symbol = symbol
super().__init__()
def on_tick(self, time, value):
self._manager._on_tick(self._symbol, value)
As a last step, we need to ensure that the runtime adapter implementations are registered with our graph:
_my_input_adapter = py_push_adapter_def(name='MyInputAdapter', adapterimpl=MyInputAdapterImpl, out_type=ts[MyData], manager_type=MyAdapterManager, symbol=str)
_my_output_adapter = py_output_adapter_def(name='MyOutputAdapter', adapterimpl=MyOutputAdapterImpl, manager_type=MyAdapterManager, input=ts['T'], symbol=str)To test this example, we will:
- instantiate our manager
- subscribe to a certain number of input adapter "streams" (which the adapter manager will demultiplex out of a single random node)
- print the data
- sink each stream into a smaller number of output adapters (which the adapter manager will multiplex into print statements)
@csp.graph
def my_graph():
adapter_manager = MyAdapterManager(timedelta(seconds=0.75))
data_1 = adapter_manager.subscribe("data_1")
data_2 = adapter_manager.subscribe("data_2")
data_3 = adapter_manager.subscribe("data_3")
csp.print("data_1", data_1)
csp.print("data_2", data_2)
csp.print("data_3", data_3)
# pump two streams into 1 output and 1 stream into another
adapter_manager.publish(data_1, "data_1")
adapter_manager.publish(data_2, "data_1")
adapter_manager.publish(data_3, "data_3")Here is the result of a single run:
2023-02-15 19:14:53.859951 data_1:MyData(symbol=data_1, value=0)
publication_data_1:MyData(symbol=data_1, value=0)
2023-02-15 19:14:54.610281 data_3:MyData(symbol=data_3, value=1)
publication_data_3:MyData(symbol=data_3, value=1)
2023-02-15 19:14:55.361157 data_3:MyData(symbol=data_3, value=2)
publication_data_3:MyData(symbol=data_3, value=2)
2023-02-15 19:14:56.112030 data_2:MyData(symbol=data_2, value=3)
publication_data_1:MyData(symbol=data_2, value=3)
2023-02-15 19:14:56.862881 data_2:MyData(symbol=data_2, value=4)
publication_data_1:MyData(symbol=data_2, value=4)
2023-02-15 19:14:57.613775 data_1:MyData(symbol=data_1, value=5)
publication_data_1:MyData(symbol=data_1, value=5)
2023-02-15 19:14:58.364408 data_3:MyData(symbol=data_3, value=6)
publication_data_3:MyData(symbol=data_3, value=6)
2023-02-15 19:14:59.115290 data_2:MyData(symbol=data_2, value=7)
publication_data_1:MyData(symbol=data_2, value=7)
2023-02-15 19:14:59.866160 data_2:MyData(symbol=data_2, value=8)
publication_data_1:MyData(symbol=data_2, value=8)
2023-02-15 19:15:00.617068 data_1:MyData(symbol=data_1, value=9)
publication_data_1:MyData(symbol=data_1, value=9)
2023-02-15 19:15:01.367955 data_2:MyData(symbol=data_2, value=10)
publication_data_1:MyData(symbol=data_2, value=10)
2023-02-15 19:15:02.118259 data_3:MyData(symbol=data_3, value=11)
publication_data_3:MyData(symbol=data_3, value=11)
2023-02-15 19:15:02.869170 data_2:MyData(symbol=data_2, value=12)
publication_data_1:MyData(symbol=data_2, value=12)
2023-02-15 19:15:03.620047 data_1:MyData(symbol=data_1, value=13)
publication_data_1:MyData(symbol=data_1, value=13)
closing asset publication_data_1
closing asset publication_data_3
Although simple, this examples demonstrates the utility of the adapters and adapter managers. An input resource is managed by one entity, distributed across a variety of downstream subscribers. Then a collection of streams is piped back into a single entity.