OPC-UA server

When someone told me to use one of the already existing OPC-UA servers and implement them so that the data gets directly sent into the cloud, I thought it sounded easy. Just configure them and boom, that should be it, right?

Well, there is not one single version out there that supports something like that. Why? The software is old, built on outdated standards dating back to the 90s. That's not necessarily bad, but it becomes problematic when there are no further improvements.

So, how do we achieve something like that today? With Windows clients that look and feel like they're straight from the 90s. At this point, do you really need to look into the code of the servers? Well, I don't.

Luckily, there is open-source software available, like Rust Crates or even Python alternatives, where you can see the code. These tools are developed by a small team of unpaid developers. Now, you might think they must be inferior and not provide meaningful usage, right? Well, buckle your seatbelt because they don't. Surprisingly, they are rather performant and easily customizable. And did I mention they are free?

So, does it mean I need a lot of time to do all this right? Well... the last time I did it, it took me just 4 hours to make it work, from variable to MQTT topic.

Code

Below you can see a short code snippet

 
    #!/usr/bin/env python
    # -*- coding: utf-8 -*-
 
 
 
    try:
        import json
        import asyncio
        from datetime import timezone
        import datetime
    except (ImportError, ModuleNotFoundError) as exception:
        raise exception
 
 
    class Event:
 
        def __init__(self, sync_rate_publish, sync_rate_subscribe):
            self.sync_rate_publish = int(sync_rate_publish)
            self.sync_rate_subscribe = int(sync_rate_subscribe)
 
        async def publish(self, objs):
            await asyncio.sleep(self.sync_rate_publish)
 
            batch = []
            for obj in objs:
                data = await self._publish(obj)
                if data.get("variables") is not None:
                    if len(data.get("variables")) != 0:
                        batch.append(data)
 
            if len(batch) != 0:
                return batch
            else:
                return "N.a."
 
        async def subscribe(self, data):
            await asyncio.sleep(self.sync_rate_subscribe)
            data = json.dumps(data.decode())
            data = data.replace("\'", '\"')
            data = data[1:]
            data = data[:-1]
            data_dict = json.loads(data)
            return data_dict['data']
 
        async def _publish(self, obj):
            async def get_browse_name(obj, standard):
                try:
                    return (await obj.read_browse_name()).Name
 
                except (Exception,):
                    return standard
 
            async def get_value(obj, standard):
                try:
                    data = []
                    variables = await obj.get_variables()
                    if len(variables) is not 0:
                        [data.append({str((await var.read_browse_name()).Name): await var.read_value()})
                         for var in variables if str((await var.read_browse_name()).Name) != "ServerStatus"]
                        return data
                    else:
                        return None
 
                except (Exception,):
                    return standard
 
            self.data = {
                "timestamp[utc]": self.__synchronous_utility_method(
                    function=lambda: datetime.datetime.now(timezone.utc).replace(tzinfo=timezone.utc).timestamp(),
                    standard="N.a."),
                "object": await get_browse_name(obj=obj, standard="N.a."),
                "variables": await get_value(obj=obj, standard="N.a.")
            }
 
            return self.data
 
        @staticmethod
        def __synchronous_utility_method(function, standard):
            try:
                return function()
            except (Exception,):
                return standard

Sounds interesting?

Get in touch with me and I will help you transform your IT & OT landscape, far more cost-effectively and efficiently than these servers ever could. Oh, and did I mention it's even safer? Well, it is.

Get in touch with me via

August 10, 2023
finished
OPC-UA server, open-source software, Rust Crates, Python, performance, customization, cost-effective