OSS Announcement

4-produce-events-from-time-series-and-binary-data-streams-with-payloads.py

# Callback triggered for each new data frame.
def on_data_frame_handler(topic, stream: StreamConsumer, df: pd.DataFrame):
		
	# We filter rows where the driver was speeding.
	above_speed_limit = df[df["speed"] > 130]

	# If there is a record of speeding, we sent a ticket.
	if df.shape[0] > O:

		# We find the moment with the highest speed.
		max_speed_moment = df['speed'].idxmax()
		speed = df.loc[max_speed_moment]
		time = df.loc[max_speed_moment]["time"]

		# We create a document that will be consumed by the ticket service.
		speeding_ticket = {
			'vehicle': stream.stream_id,
			'time': time,
			'speed': speed,
			'fine': (speed - 130) * 100,
			'photo_proof': df.loc[max_speed_moment]["camera_frame"]
		}

		topic_producer.get_or_create_stream(stream.stream_id) \
			.events \
			.add_timestamp_in_nanoseconds(time) \
			.add_value("ticket", json.dumps(speeding_ticket)) \
			.publish()