如何过滤 Apache Edgent 并显示被过滤的值？答案

【问题标题】：How to filter on Apache Edgent and also show the values which were filtered?如何过滤 Apache Edgent 并显示被过滤的值？
【发布时间】：2018-12-03 18:39:21
【问题描述】：

我正在使用 Apache Edgent（Java 框架）每 3 秒从 Raspberry Pi 上的 HCSR04 超声波传感器轮询一次值。我使用过滤器无法获取 50 厘米到 80 厘米之间的值。

    UltrasonicStream sensor = new UltrasonicStream();
    DirectProvider dp = new DirectProvider();
    Topology topology = dp.newTopology();
    TStream<Double> tempReadings = topology.poll(sensor, 3, TimeUnit.SECONDS);
    TStream<Double> filteredReadings = tempReadings.filter(reading -> reading < 50 || reading > 80);
    System.out.println("filter added: tempReadings.filter(reading -> reading < 50 || reading > 80);");
    filteredReadings.print();
    dp.submit(topology);

我想在过滤值时显示一些消息。当值与我的过滤器不匹配时，我可以轮询它们，但是当它们匹配时我不返回，那没关系。但是，我只想表明使用 Apache Edgent 库过滤了一个值。我知道我可以在 public double get() 方法上做一些事情，但我想知道我是否可以用 Apache Edgent 的一些方法来做这个技巧。

公共类 UltrasonicStream 实现供应商{

private static final long serialVersionUID = -6511218542753341056L;

private static GpioPinDigitalOutput sensorTriggerPin;
private static GpioPinDigitalInput sensorEchoPin;
private static final GpioController gpio = GpioFactory.getInstance();
private double currentDistance = -1.0;

/**
 * The HCSR04 Ultrasonic sensor is connected on the physical pin 16 and 18 which
 * correspond to the GPIO 04 and 05 of the WiringPi library.
 */
public UltrasonicStream() {
    // Trigger pin as OUTPUT
    sensorTriggerPin = gpio.provisionDigitalOutputPin(RaspiPin.GPIO_04);
    // Echo pin as INPUT
    sensorEchoPin = gpio.provisionDigitalInputPin(RaspiPin.GPIO_05, PinPullResistance.PULL_DOWN);
}

/**
 * This is the override method of the Supplier interface from Apache Edgent
 */
@Override
public Double get() {
    try {
        System.out.print("Distance in centimeters: ");
        currentDistance = getDistance();
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return currentDistance;
}

/**
 * Retrieve the distance measured by the HCSR04 Ultrasonic sensor connected on a
 * Raspberry Pi 3+B
 * 
 * @return the distance in centimeters
 * @throws InterruptedException
 */
public double getDistance() throws InterruptedException {

    double distanceCM = -1;
    try {
        // Thread.sleep(2000);
        sensorTriggerPin.high(); // Make trigger pin HIGH
        Thread.sleep((long) 0.01);// Delay for 10 microseconds
        sensorTriggerPin.low(); // Make trigger pin LOW

        // Wait until the ECHO pin gets HIGH
        while (sensorEchoPin.isLow()) {

        }
        // Store the current time to calculate ECHO pin HIGH time.
        long startTime = System.nanoTime();
        // Wait until the ECHO pin gets LOW
        while (sensorEchoPin.isHigh()) {

        }
        // Store the echo pin HIGH end time to calculate ECHO pin HIGH time.
        long endTime = System.nanoTime();

        distanceCM = ((((endTime - startTime) / 1e3) / 2) / 29.1);
        // Printing out the distance in centimeters
        // System.out.println("Distance: " + distanceCM + " centimeters");

        return distanceCM;
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return distanceCM;
}

}

【问题讨论】：

标签： apache-edgent

【解决方案1】：

您可以使用 TStream.split() 创建两个流：一个其元组与您的过滤谓词匹配，另一个用于不匹配的那些。然后，您可以对任一流做任何您想做的事情。例如TStream.peek(t -> System.out.println("excluded: " + t)) 或 TStream.print(...)

【讨论】：

【解决方案2】：

我是这样实现的：

UltrasonicStream sensor = new UltrasonicStream();

DirectProvider dp = new DirectProvider();

Topology topology = dp.newTopology();

TStream<Double> tempReadings = topology.poll(sensor, 3, TimeUnit.SECONDS);

TStream<Double> filteredReadings = tempReadings.filter(reading -> {
    boolean threshold = reading < 20 || reading > 80;
    if (!threshold) {
        System.out.println(String.format("Threshold reached: %s cm", reading));
    }
    return threshold;
});
filteredReadings.print();

dp.submit(topology);

【讨论】：