如何在ApacheEdgent上进行筛选并显示筛选的值

Question

我使用Apache Edgent(Java框架(每3秒轮询一次树莓派上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);

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

公共类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;
}

}

Answer 1

您可以使用TStream.split((创建两个流：一个流的元组与筛选器谓词匹配，另一个流与筛选器谓词不匹配。然后，您可以对任意一个流执行任意操作。例如TStream.peek(t->System.out.println("excluded:"+t((或TStream.print(…(

Answer 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);