pahoMqttExt

Hi guys,

we just started to use Desigo Optic recently. try to creat Mqtt connection on Optic. we have some troulbe on parsing predefined Mqtt data format, which was predefined in the gateway hardware. the parsing axon function could work propely in Skyspark 3.1. but it looks that the configuration method on Optic is different from skyspark. does anybody have detailed documentation for config the pahomqtt in Desigo Optic to share, that would be very helpful. as the folio version is 3.0.29 in Desigo Optic, for the Mqtt connection, is there any other option besides pahomqtt?

thank you in advance.

Hi William.

I can't tell from the note what the problem is. Have you configured on the connector what your axon parser function name is? And is that function working when you test it with a sample message?

This MQTT Connector is intended to be completely agnostic of what payload formats you're dealing with, hence I built it such that you can configure in your own parser function.

Please can you provide some more detail on what is failing.

Regards, John.

Hi John,

thank you for your reply. here is the data format example which was sent from our gateway:

Topic: /edge/mqttsn001/20249ED38CBF48B385B776906A52D55C/rtgQoS: 0 {"devs":[{"d":[{"dq":192,"m":"FanC","ts":1706596835,"v":1},{"dq":192,"m":"SN","ts":1706596835,"v":0},{"dq":192,"m":"FanMA","ts":1706596835,"v":0},{"dq":192,"m":"FanAml","ts":1706596835,"v":0},{"dq":192,"m":"FanS","ts":1706596835,"v":0},{"dq":192,"m":"T","ts":1706596835,"v":0.000},{"dq":192,"m":"H","ts":1706596835,"v":0.000},{"dq":192,"m":"Valve","ts":1706596835,"v":-198656.000},{"dq":192,"m":"TSP","ts":1706596835,"v":0.000}],"dev":"Device"}],"pKey":"mqttsn001","sn":"20249ED38CBF48B385B776906A52D55C","ts":1706596835,"ver":"2.0.0"}

and we had edited a function from an example in finstack's website, which could only parse one point's data in Desigo Optic, below is the code for function: (msg,params: null) => do

jsonData: ioReadJson(msg)
jds:jsonData->devs.get(0)
ename:jds->dev
jd:jds->d
tags:{}
tags = tags.set("curVal",jd.get(0)->v)
tags = tags.set("curStatus","ok")

end

as we try to pass an parameter to the function, but the function did not work, here is the rewrited function code: (msg,params: null,x) => do

jsonData: ioReadJson(msg)
jds:jsonData->devs.get(0)
ename:jds->dev
jd:jds->d
tags:{}
tags = tags.set("curVal",jd.get(x)->v)
tags = tags.set("curStatus","ok")

end

any fault in this function?

Regards, William

Hi William.

I'm not sure this Paho MQTT Connector is the right thing for your use case. I think you just need a subscriber to parse the messages and write to point history.

This Paho connector is quite old so I'd recommend moving to using the built in SkySpark support for MQTT.

Your message indicates you've got values for multiple points in the one message, and the fact that there is a timestamp means it's not really a curVal message.

I would suggest reading the SkySpark MQTT docs, set up a native SkySpark MQTT connector to your MQTT Broker, and paying particular attention to the Subscribe section, I think that would give you a much better means of processing these messages and writing directly to point history for all points in the message using the timestamp each has, along with its value.

Hi John,

i think you are right. what i need just a subscriber to parse the messages and write to point history.

we had already wroten an axon function in skyspark with its buildin MQTT. the function could work properly.

my problems is that we need to work in Desigo Optic, which is the Siemens product name of FinStack. till now, the latest Desigo Optic only support to folio 3.0.29, which still don't have built in MQTT. that's why we choose the pahomqtt, and this is also the suggestion from Siemens.

we don't know why our axon function could not work in Desigo Optic. does the pahomqtt have any restrict on axon function? here is the code:

(msg,params: null) => do

//读取topic的json数据，数据类型为dict

jsonData: ioReadJson(msg)
jds:jsonData->devs.get(0)
ename:jds->dev//设备名字
jd:jds->d
//tags:{}
//tags = tags.set("curVal",jd.get(0)->v)
//tags = tags.set("curStatus","ok")
fanc:read(point and navName == "FanC"and equipRef->navName==ename)
sn:read(point and navName == "SN"and equipRef->navName==ename)
fanma:read(point and navName == "FanMA"and equipRef->navName==ename)
fanaml:read(point and navName == "FanAml"and equipRef->navName==ename)
fans:read(point and navName == "FanS"and equipRef->navName==ename)
t:read(point and navName == "T"and equipRef->navName==ename)
h:read(point and navName == "H"and equipRef->navName==ename)
valve:read(point and navName == "Valve"and equipRef->navName==ename)
tsp:read(point and navName == "TSP"and equipRef->navName==ename)
dt:now()
 hisWrite({ts: dt, val:jd.get(0)->v},  fanc)
 hisWrite({ts: dt, val:jd.get(1)->v},  sn)
 hisWrite({ts: dt, val:jd.get(2)->v},  fanma)
 hisWrite({ts: dt, val:jd.get(3)->v},  fanaml)
 hisWrite({ts: dt, val:jd.get(4)->v},  fans)
 hisWrite({ts: dt, val:jd.get(5)->v},  t)
 hisWrite({ts: dt, val:jd.get(6)->v},  h)
 hisWrite({ts: dt, val:jd.get(7)->v},  valve)
 hisWrite({ts: dt, val:jd.get(8)->v},  tsp)

end

could you help us check this code? thanks.

William

Hi William.

I understand. It's a bit unfortunate to be forced to be stuck on a version of SkySpark that's now nearly 3 years old!

Anyway, given that, I'd suggest you move to using the pahoMqttSub extension instead. You don't need the connector, you just need an MQTT Subscription to give you the messages when they arrive and you just handle the content, as laid out in your code.

As for the validity of the code, there seems to be no space before the and in each of the read calls to get the points. And you're using a variable called ename that hasn't been declared or set.

The function (even it corrected) would still be very prone to errors as it relies on unchanging navNames and the order of values in the MQTT message. I'll write something that could work, but it might be a couple of days before I get back to you.

Regards, John.

Hi John,

thank you for your advice. i will try pahoMqttSub then. and I appreciate your kind help very much.

Regards,

William

Hi William.

As a rough guide for what I think the code in a subscriber func should do:

jsonData: ioReadJson(msg)
metrics:jsonData->devs.get(0)->d
metrics.each(metric => do
  pt: read(point and someTag == metric->m, false)
  if (pt != null) do
    ts: m->ts.fromJavaMillis(pt->tz)
    hisWrite({ts:ts, val:m->v}, pt)    
  end
end)

You just need to decide on the name of some tag to mark the 9 points with a value that equals the value of m in each metric.

Hi John,

i just came back from vocation. thank you for your kind support. i will try this func soon, and let you know.

Hi John,

i had tried your function. with some small change, i already could read and save the data from MQTT borker by the function. thank you for your help.

one more question, i am tring to write data back(publish) to local point by MQTT. i think that would need to edit an function for pahoMqttConstructor, is that right? it looks like that the function, pahoMqttConstructJson(pt, val, level), would return an fixed JSON format. can you provide an example function which could edit JSON data format as the gateway published. or i have any misunderstand on this point?

Hi William,

Yes, if you wish to make points writable, and control the format of the message sent to the Broker when you use the pointWrite function you need a custom constructor, you can use your own Axon function. The contract for a payload constructor (in Fantom syntax) is:

Str myAxonConstructor(ConnPoint pt, Dict val, Dict? params := null)

The signature is shown in Fantom syntax to highlight the parameter types passed to the function. The function must return a Str. The pt param can be accessed as a Dict to extract any Point tags needed. The val Dict will at a minimum contain writeLevel. If the val being set on the point is not null, it will contain writeVal.

{writeVal:true, writeLevel:8}

An example of a custom constructor configuration in cpMqttConstructor tag is:

myCsvConstructor {colOrder:["curVal", "curLevel"]}

Hi John,

i still need your help. i already wrote an axon funtion to construct data format as predefined. the function could work correctly. below is the code: (pt, val) => do

jsonData: ioReadJson(`io/jsonDataFormat1.json`)
jds:jsonData->data
tn: round(now().toJavaMillis()/1000)
if (val != null) do
  jsonData = jsonData.set("ts",tn)
  jds = jds.set("m", pt->navName.toStr)
  jds = jds.set("v", val)
  jsonData = jsonData.set("data", jds)
  ioWriteJson(jsonData,`io/testjson.json`)
  return jsonData
end
else return "NULL"

end

but after i set it up on the pahoMqttConstructor tag, while i try to sent the data through pahoMqtt conn, i got the error message as below: axon::EvalErr: Func failed: trap(Obj val,Str name); args: (ConnPoint,Str)

sys::ArgErr: Invalid arg 'connExt::ConnPoint' to 'core::trap' [proj_demo::jsonTest:7]
axon::FantomFn.callx (FantomFn.fan:133)
axon::Fn.callLazy (Fn.fan:75)
axon::Call.eval (Call.fan:39)
axon::Fn.callLazy (Fn.fan:75)
fan.sys.List.map (List.java:844)
axon::Fn.callLazy (Fn.fan:75)
axon::Call.eval (Call.fan:39)
axon::Fn.callLazy (Fn.fan:75)
fan.sys.List.map (List.java:844)
axon::Fn.callLazy (Fn.fan:75)
axon::Call.eval (Call.fan:39)
axon::Assign.doEval (Operators.fan:307)
axon::BinaryOp.eval (Operators.fan:65)
axon::Block.eval (Block.fan:33)
fan.sys.List.each (List.java:604)
axon::Block.eval (Block.fan:33)
axon::If.eval (Operators.fan:268)
axon::Block.eval (Block.fan:33)
fan.sys.List.each (List.java:604)
axon::Block.eval (Block.fan:33)
axon::Fn.doCall (Fn.fan:105)
axon::AxonContext.callInNewFrame (AxonContext.fan:153)
axon::AxonContext.callInNewFrame (AxonContext.fan)
axon::Fn.callx (Fn.fan:98)
axon::Fn.call (Fn.fan:71)
14 More...

any suggestion on this error message, and how to fix it?

William

hi William, I just used Desigo Optic version 5.4.191 and encountered the same problem. The pahomqtt library requires Axon 3.1, but Optic only has version 3.0.29. How did you use it?

Hi Kevin,

you could download v1.0.14 version of pahomqtt pod, which could work on Optic.