Skip to content

PB serialization for CS-Studio/Phoebus by integrating IoTDB with Archiver Appliance #642

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
glcj merged 2 commits into from
Apr 15, 2026
+698 −143
Merged

PB serialization for CS-Studio/Phoebus by integrating IoTDB with Archiver Appliance #642

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1b936e2
–feat: PB serialization for CS-Studio/Phoebus, integrating IoTDB with…
Lerb123 Apr 15, 2026
f9d83ed
–feat: PB serialization for CS-Studio/Phoebus, integrating IoTDB with…
Lerb123 Apr 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,12 @@
<configuration>
<mainClass>org.apache.plc4x.merlot.archiver.test.NewMain</mainClass>
</configuration>
</plugin>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.22.0</version>
</plugin>
</plugins>
</build>
<dependencies>
Expand Down Expand Up @@ -236,5 +241,12 @@
<artifactId>lz4</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
<type>jar</type>
</dependency>
</dependencies>
</project>
122 changes: 61 additions & 61 deletions ...lot.archiver/src/main/java/org/apache/plc4x/merlot/archiver/command/MerlotHtcCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,8 @@
*/
package org.apache.plc4x.merlot.archiver.command;


import java.time.Instant;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.karaf.shell.api.action.Action;
import org.apache.karaf.shell.api.action.Argument;
import org.apache.karaf.shell.api.action.Command;
Expand All @@ -30,52 +26,56 @@
import org.apache.karaf.shell.api.action.lifecycle.Service;
import org.apache.karaf.shell.support.table.ShellTable;
import org.apache.plc4x.merlot.archiver.api.MerlotHtc;
import org.epics.gpclient.PV;
import org.apache.plc4x.merlot.archiver.impl.MerlotHtcIoTDBImpl;
import org.apache.plc4x.merlot.archiver.impl.MerlotHtcRTImpl;
import org.epics.vtype.Scalar;
import org.epics.vtype.Time;
import org.epics.vtype.VType;
import org.osgi.framework.BundleContext;
import org.slf4j.LoggerFactory;

@Service
@Command(scope = "plc4x", name = "htc", description = "Historical collector admin.")
public class MerlotHtcCommand implements Action {
public class MerlotHtcCommand implements Action {

@Reference
BundleContext bc;

@Reference
volatile List<MerlotHtc> htcs;
volatile List<MerlotHtc> htcs;

@Option(name = "-l", aliases = "--list", description = "List collector PVs.", required = false, multiValued = false)
String strHtc;
String strHtc;

@Option(name = "-p", aliases = "--pv", description = "Displays the stored values ​​of the process variable.", required = false, multiValued = false)
String strPV;

@Option(name = "-a", aliases = "--add", description = "The designated PV tags must be incorporated into the collector.", required = false, multiValued = false)
Boolean blnAdd;
Boolean blnAdd;

@Option(name = "-f", aliases = "--from", description = "Start date for data query.", required = false, multiValued = false)
String from;
String from;

@Option(name = "-t", aliases = "--to", description = "End date for data consultation.", required = false, multiValued = false)
String to;
String to;

@Option(name = "-r", aliases = "--remove", description = "The designated PV tags must be remove from the collector.", required = false, multiValued = false)
String strRemovePV;


String strRemovePV;

@Argument(index = 0, name = "htc", description = "The Collector to which the PV tag will be assigned.", required = false, multiValued = false)
String strMainHtc;
String strMainHtc;

@Argument(index = 1, name = "rate", description = "Maximum rate at which a tag change must be acquired.", required = false, multiValued = false)
String strMaxRate;
String strMaxRate;

@Argument(index = 2, name = "pvs", description = "List of PVS tags to htc.", required = false, multiValued = true)
List<String> strPVs;


List<String> strPVs;


private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(MerlotHtcCommand.class);
@Override
public Object execute() throws Exception {

if ((null == strHtc) && (null == strMainHtc) ){
ListCollectorsServices(null);
} else if ((null != strHtc) && (null == strMainHtc) && (null == strPV)) {
Expand All @@ -87,75 +87,75 @@ public Object execute() throws Exception {
} else if ((null != strRemovePV) && (null != strMainHtc)) {
removePV(strMainHtc, strRemovePV);
}


return null;
}
private void ListCollectorsServices(String id){

private void ListCollectorsServices(String id) {
if (null == id) {
ShellTable table = new ShellTable();
table.column("Uid");
table.column("Key");
htcs.forEach(htc -> {
table.addRow().addContent(htc.getID(),htc.getPVs().size());
table.addRow().addContent(htc.getID(), htc.getPVs().size());
});
table.print(System.out);

} else {
htcs.forEach(h -> {
if(h.getID().equalsIgnoreCase(id)){
if (h.getID().equalsIgnoreCase(id)) {
int count = 1;
ShellTable table = new ShellTable();
table.column("Item");
table.column("PV");
table.column("PV");
h.getPVs().forEach(pc -> {
table.addRow().addContent(count, pc);
});
table.print(System.out);
table.print(System.out);
}
});

}
}
private void ListHistoricalValues(String pv, String init, String end){

private void ListHistoricalValues(String pv, String init, String end) {
int[] counter = new int[1];
htcs.forEach(h -> {
if(h.getID().equalsIgnoreCase(strHtc)){
String strInit = (null == init)?Instant.MIN.toString():init;
String strEnd = (null == end)?Instant.MAX.toString():end;
if (h.getID().equalsIgnoreCase(strHtc)) {
String strInit = (null == init) ? Instant.MIN.toString() : init;
String strEnd = (null == end) ? Instant.MAX.toString() : end;

var pvs = h.getPVs(pv, strInit, strEnd);
ShellTable table = new ShellTable();
table.column("Date");
table.column("Value");
table.column("Value");
counter[0] = 1;
pvs.forEach((p) ->{
table.addRow().addContent(counter[0], p.toString());
pvs.forEach((p) -> {
table.addRow().addContent(counter[0], p.toString());
counter[0]++;
});
table.print(System.out);
table.print(System.out);
}
});
});
}

private void addPV(String pv, String maxRate, List<String> listPVs) {
htcs.forEach(h -> {
if(h.getID().equalsIgnoreCase(strMainHtc)){
listPVs.forEach(s ->{
h.addPV(s, Double.MAX_VALUE);
});
}
});
htcs.forEach(h -> {
if (h.getID().equalsIgnoreCase(strMainHtc)) {
listPVs.forEach(s -> {
h.addPV(s, Double.MAX_VALUE);
});
}
});
}

private void removePV(String htc, String strpv) {
htcs.forEach(h -> {
if(h.getID().equalsIgnoreCase(htc)){
h.removePV(strpv);
}
});
}


htcs.forEach(h -> {
if (h.getID().equalsIgnoreCase(htc)) {
h.removePV(strpv);
}
});
}

}
122 changes: 122 additions & 0 deletions ...rlot.archiver/src/main/java/org/apache/plc4x/merlot/archiver/core/MerlotIoTDBMapping.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.plc4x.merlot.archiver.core;

import com.google.protobuf.ByteString;
import java.util.function.BiFunction;
import org.apache.iotdb.pipe.api.type.Type;
import org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarByte;
import org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarDouble;
import org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarFloat;
import org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarInt;
import org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarString;

public enum MerlotIoTDBMapping {
INT32(Type.INT32, (val, meta)
-> ScalarInt.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal((Integer) val)
.setSeverity(meta.severity)
.setStatus(meta.status)
.build()),

INT64(Type.INT64, (val, meta) -> {
long numericValue = ((Number) val).longValue();

return org.apache.plc4x.merlot.api.PB.EPICSEvent.ScalarDouble.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal((double) numericValue)
.setSeverity(meta.severity)
.setStatus(meta.status)
.build();
}),

FLOAT(Type.FLOAT, (val, meta)
-> ScalarFloat.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal((Float) val)
.setSeverity(meta.severity)
.setStatus(meta.status)
.build()),
DOUBLE(Type.DOUBLE, (val, meta)
-> ScalarDouble.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal((Double) val)
.setSeverity(meta.severity)
.setStatus(meta.status)
.build()),
TEXT(Type.TEXT, (val, meta)
-> ScalarString.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal(val.toString())
.setSeverity(meta.severity)
.setStatus(meta.status)
.build()),
BOOLEAN(Type.BOOLEAN, (val, meta)
-> ScalarByte.newBuilder()
.setSecondsintoyear(meta.seconds)
.setNano(meta.nano)
.setVal(ByteString.copyFrom(new byte[]{(byte) ((boolean) val ? 1 : 0)}))
.setSeverity(meta.severity)
.setStatus(meta.status)
.build());


private final Type iotdbType;
private final BiFunction<Object, EpicsMetadata, Object> converter;

MerlotIoTDBMapping(Type iotdbType, BiFunction<Object, EpicsMetadata, Object> converter) {
this.iotdbType = iotdbType;
this.converter = converter;
}

public Object convert(Object value, EpicsMetadata meta) {
return this.converter.apply(value, meta);
}

public static MerlotIoTDBMapping fromIotdb(Type type) {
for (MerlotIoTDBMapping c : values()) {
if (c.iotdbType == type) {
return c;
}
}
throw new UnsupportedOperationException("IoTDB type not supported: " + type);
}

/**
* Container for additional fields defined in EPICSEvent.proto
*/
public static class EpicsMetadata {

public final int seconds;
public final int nano;
public final int severity;
public final int status;

public EpicsMetadata(int seconds, int nano, int severity, int status) {
this.seconds = seconds;
this.nano = nano;
this.severity = severity;
this.status = status;
}
}
}
Loading
Loading