React to data, fast.

topic = quix.open_output_topic("payment-transactions")
stream = topic.get_or_create_stream(CUSTOMER_ID)
stream.events.write(payments)

//sends parameter (payment) data

using var outputTopic = quix.OpenOutputTopic("payment-transactions");
var streamWriter = outputTopic.CreateStream("CUSTOMERID");

streamWriter.Events
    .AddTimestamp(DateTime.Now)
    .AddValue("Payment", PaymentData)
    .Write();

transactions = quix.open_input_topic("payment-transactions")
blocked_topic = quix.open_output_topic("payment-blocked")
blocked_payments = blocked_topic.create_stream("blocked_payments")
blocked_cache = quix.open_input_topic("blocked_accounts").as_cache()

// Lets get this into Pandas (changed to blocking theme)

def handle_data(data):
 if data.account_num in blocked_cache:
  blocked_payments.events.add("Block", data.transaction_num)

transactions.on_read_data += handle_data

//receives parameter data sends parameter data

using var inputTopic = quix.OpenInputTopic("payment-transactions");
using var blockedAccounts = quix.OpenInputTopic("blocked-accounts").AsCache();
using var outputTopic = quix.OpenOutputTopic("payment-blocked");

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    var streamWriter = outputTopic.CreateStream();
    streamReader.Parameters.OnRead += (paymentData) =>
    {
        if (blockedAccounts.Contains(paymentData.AccountNumber))
        {
            streamWriter.Parameters
                .Buffer
                .AddTimestamp(DateTime.Now)
                .AddValue("Block", paymentData.TransactionNumber)
                .Write();
        }
    };
};

transactions = quix.open_input_topic("payment-transactions")
fraud_topic = quix.open_output_topic("fraud-transactions")
fraud_model = pickle.load(open("fraud_model.pkl"))

def handle_data(data, fraud_stream):
 is_fraud = model.detect(data)
 fraud_stream.events.write(data.transaction)

transactions.on_read_data += handle_data

//receives parameter data sends fraud event

using var inputTopic = quix.OpenInputTopic("payment-transactions");
using var outputTopic = quix.OpenOutputTopic("fraud-transactions");
var fraudModel = System.IO.File.OpenRead("fraudModel.pkl");

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    var streamWriter = outputTopic.CreateStream();
    streamReader.Parameters.OnRead += (paymentData) =>
    {
        if (fraudModel.Detect(paymentData))
        {
            streamWriter.Events
                .AddTimestamp(DateTime.Now)
                .AddValue("PaymentData", paymentData)
                .Write();
        }
    };
};

fraud_transactions = quix.open_input_topic("fraud-transactions")
segment = segment_connector("zen-desk")
big_query.connect()

def handle_data(data):
 segment.create_ticket(data, priority="HIGH")
 big_query.write(data)

fraud_transactions.on_read_data += handle_data

//receives event data sends to segment and data lake

using var inputTopic = quix.OpenInputTopic("fraud-transactions");
var segmentConnector = new SegmentConnector("ZenDesk");
var dataLake = new BigQueryConnector();

inputTopic.OnStreamReceived += (s, streamReader) =>
{
 streamReader.Events.OnRead += (fraudData) =>
    {
        if (fraudData.FraudResult)
            segmentConnector.CreateTicket(fraudData, priority = "HIGH");
        dataLake.Write(fraudData);
    };
};

topic = quix.open_output_topic("factory")
stream = topic.create_stream(DEVICE_ID)
device = connector.get_device(DEVICE_ID)

while(device.is_running)
 data = device.get_data()
 stream.parameters.write(data)

using var outputTopic = quix.OpenOutputTopic("factory");
var streamWriter = outputTopic.CreateStream();
var device = deviceConnector.GetDevice("DEVICEID");

while (device.IsRunning)
{
    streamWriter.Parameters
        .Buffer
        .AddTimestamp(DateTime.Now)
        .AddValue("Temperature", device.Temperature)
        .AddValue("PowerOutput", device.PowerOutput)
        .AddValue("BatteryLevel", device.BatteryLevel)
        .Write();
}

// Based on machine.type look up the service interval and set machine.hours_to_service it based on machine.hours_since_service.
// Also set machine.service_type

factory = quix.open_input_topic("factory")
service_topic = quix.open_output_topic("servicing")

def handle_data(device, service_stream):
 hours_till_service = device.service_interval - device.runtime
 if hours_till_service < 100:
  service_stream.events \
   .add("DeviceId", device.id) \
   .add("ServiceParts", device.service.parts) \
   .add("ServiceType", device.service.type) \
   .add("TimeRemaining", hours_till_service) \
   .write()

factory.on_read_data += handle_data

//receives param, identifies event and sends event, tags deviceid

using var inputTopic = quix.OpenInputTopic("factory");
using var outputTopic = quix.OpenOutputTopic("servicing");

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    var streamWriter = outputTopic.CreateStream();
    streamReader.Parameters.OnRead += (deviceData) =>
    {
        var hoursTillService = deviceData.ServiceInterval - deviceData.RunTime;
        if (hoursTillService < 100)
        {
            streamWriter.Events
                .AddTimestamp(DateTime.Now)
                .AddTag("DeviceId", deviceData.Id)
                .AddValue("ServiceParts", deviceData.Service.Parts)
                .AddValue("ServiceType", deviceData.Service.Type)
                .AddValue("TimeRemaining", hoursTillService)
                .Write();
        }
    };
};

// If service is near, order service parts using POST.

factory = quix.open_input_topic("factory")
factory_events = quix.open_output_topic("factory-events")
parts_catalogue = parts_api.connect()

def handle_data(service, factory_events_stream):
 parts_required = service["ServiceParts"]
 parts_catalogue.order_parts(parts_required)
 factory_events_stream    .events \
  .add("PartsOrdered", parts_required) \
  .add("DeviceId", service["DeviceId"]) \
  .add("TimeRemaining", service["TimeRemaining"]) \
  .write()

factory.on_read_data += handle_data

//receives event, sends parameter and parts order to api, tags deviceid

using var inputTopic = quix.OpenInputTopic("servicing");
using var outputTopic = quix.OpenOutputTopic("factory-events");
var partsCatalogue = new PartsApi().Connect();

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    var streamWriter = outputTopic.CreateStream();
    streamReader.Events.OnRead += (serviceData) =>
    {
        partsCatalogue.OrderParts(serviceData.ServiceParts);

        streamWriter.Parameters
            .Buffer
            .AddTimestamp(DateTime.Now)
            .AddTag("DeviceId", serviceData.DeviceId)
            .AddValue("PartsOrdered", serviceData.ServiceParts)
            .AddValue("TimeRemaining", serviceData.TImeRemaining)
            .Write();
    };
};

factory_events = quix.open_input_topic("factory_events")
data_lake = snowflake.connect()

def handle_data(service):
 parts = service["PartsOrdered"]
 device_id = service["DeviceId"]
 time_remaining = service["TimeRemaining"]
 data_lake.persist(datetime.now(), parts, device_id, time_remaining)

factory_events.on_read_data += handle_data

//receives parameter, sends to snowflake

using var inputTopic = quix.OpenInputTopic("factory-events");
var snowflakeConnector = new SnowflakeConnector();

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    streamReader.Parameters.OnRead += (serviceData) =>
    {
        snowflakeConnector.PersistData(DateTime.Now,
            serviceData.PartsOrdered,
            serviceData.DeviceId,
            serviceData.TimeRemaining);
    };
};

topic = quix.open_output_topic("vehicle-telemetry")

telemetry_payload = {timestamp: get_date(), location: coordinates}

stream = topic.create_stream(vehicle_id)
stream.parameters.write(telemetry_payload)

using var outputTopic = quix.OpenOutputTopic("vehicle-telemetry");
var telemetryPayload = $@"[{{ 'timestamp': '{DateTime.Now}',
 'location': coords }}]";
var streamWriter = outputTopic.CreateStream();

streamWriter.Parameters
    .Buffer
    .AddTimestamp(DateTime.Now)
    .AddTag("VehicleId", this.VehicleId)
    .AddValue("Telemetry", telemetryPayload)
    .Write();

telemetry = quix.open_input_topic("vehicle-telemetry")
output_topic  = quix.open_output_topic("telemetry-enriched")
weather_data = weather_api.connect()

def handle_data(df: pd.DataFrame):
 df['weather'] = df.apply(lambda row: weather_data[row.region], axis = 1)

 output_stream.events.write(telemetry)

//receives params sends params

using var inputTopic = quix.OpenInputTopic("vehicle-telemetry");
using var outputTopic = quix.OpenOutputTopic("telemetry-enriched");
var weatherData = httpApi.GetWeatherData();

inputTopic.OnStreamReceived += (s, streamReader) =>
{
 var streamWriter = outputTopic.CreateStream();
    streamReader.Parameters.OnRead += (vehicleData) =>
    {
        var region = vehicleData.Timestamps[0].Parameters["region"].StringValue;
        var regionWeather = weatherData[region];

        streamWriter.Parameters
            .Buffer
            .AddTimestamp(DateTime.Now)
            .AddTag("VehicleId", vehicleData.VehicleId)
            .AddValue("weather", regionWeather)
            .AddValue("vehicle", vehicleData)
            .Write();
    };
};

enriched_telemetry = quix.open_input_topic("telemetry-enriched")
hazards = quix.open_output_topic("hazards")
weather_model = pickle.load(open("hazard_model.pkl"))

def handle_data(data, hazard_warnings_stream):
 for vehicle in enriched_telemetry:
  vehicle_danger = weather_model.detect_danger(
   vehicle.weather_data,
   vehicle.gForce,
   vehicle.speed)

  hazard_warnings_stream.events.write(vehicle, vehicle_danger)

factory.on_read_data += handle_data

//receives param, sends danger event

using var inputTopic = quix.OpenInputTopic("telemetry-enriched");
using var outputTopic = quix.OpenOutputTopic("hazards");
var weatherModel = System.IO.File.OpenRead("weatherModel.pkl");

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    var streamWriter = outputTopic.CreateStream();
    streamReader.Parameters.OnRead += (parameterData) =>
    {
        var weather = parameterData.Weather;
        var gForce = parameterData.Vehicle.GForce;
        var speed = parameterData.Vehicle.Speed;
        var vehicleDanger = weatherModel.DetectDanger(weather,
            gForce,
            speed);

        streamWriter.Events
            .AddTimestamp(DateTime.Now)
            .AddValue(parameterData.Vehicle.Id, vehicleDanger);
    };
};

hazard_warnings = quix.open_input_topic("hazards")
snowflake.connect()

def on_vehicle_hazard(hazard_data):
 if hazard_data.vehicle == this.vehicle:
  this.vehicle.display_hazard(hazard_data)

hazard_warnings.subscribe += on_vehicle_hazard

// receives an event

using var inputTopic = quix.OpenInputTopic("hazards");
using var snowflakeConnector = new SnowflakeConnector();

inputTopic.OnStreamReceived += (s, streamReader) =>
{
    streamReader.Events.OnRead += (eventData) =>
    {
        if (eventData.Id == this.Vehicle.Id)
        {
            this.Vehicle.DisplayHazard(eventData.VehicleDanger);
        }
    };
};