Open Footprint on Fabric Part 2 – Activity Data Models for OPC UA

In the last post, I have shown how to deploy Open Footprint schemas on Microsoft Fabric as OneLake Tables. I’ve got great feedback and working with the forum to get the schemas finalized for v4 and also adding them as templates to the tooling. Once the tables are there and the reference data loaded, as expected the immediate question was how to get the data in?

I’ve been working with Northern Lights CCS project in developing a digitalization platform. Northern Lights is the first ever cross-border, open-source CO2 transport and storage infrastructure network and offers companies across Europe the opportunity to store their CO2 safely and permanently underground. CCS projects consist of very complex value chains to capture, transport and store the CO2, which in most cases are run my multiples of different entities which needs to communicate with each other in a common language. There are numerous business functions that needs to be implemented to successfully plan and operate CCS projects. ?Calculating and reporting emissions is one of top workflows if not the top one. In medallion architecture jargon, OFP is a data format that is useful for gold layer for curated analytics data. The emissions workflows start with ingesting the activity data from the different equipment which will then be used to calculate and report the overall emissions footprint of the organization. To be able to ingest data in a standard and repeatable way across hundreds of industrial emitters, we need a transport data format and an accompanying data schema for bronze layer, which is practically a time series store.

As part of our effort with various CCS projects, with a group of technology and software providers we have founded a workgroup under OPC Foundation (Working Group Carbon Capture and Storage (CCS) (opcfoundation.org)) to build a companion spec for data exchange across the value chain, the companion spec is in draft state and we do welcome new members.

In this post, I will introduce a preliminary information model for ingesting activity data into an OFP store in Microsoft Fabric OneLake in a medallion architecture.

Let us start with the entity types for activity data in Open Footprint.

Starting with the reference data;

Emission Activity Type defines the type of emissions such as Methane Fugitive and the Category.

The categories are defined as a combination of the scope defined in the Activity Scope such as Scope 1 to 3 and the Categories Scope 3? / 1 -15 that are defined in the Activity Category tables.

Master data tables contain;

Emission Inventory has the inventory buckets in the organization per facility and equipment.

Emission Activity connects the activities to Scope type, Organizational Boundary, Inventory and Activity Types.

Emission Activity Parameter defines the unit of measure for the activity value.

Emission Parameter Type defines the names for the type of activities.?

Our main entity is Emission Activity Parameter Value where the main time series data for activity is stored, is has the Value, DateTime and connects the Activity to other master data.

Emission Inventory - master-data

Emission Activity Type - reference-data

Emission Activity Category - reference-data

Emission Scope Type - reference-data

Emission Activity - master-data

Emission Activity Parameter - master-data

Emission Activity Parameter Value - work-product-component

Emission Parameter Type - master-data

To be able to store the activity data information that is needed for the Open Footprint, we create a data model for the time series data to be stored in the Bronze Layer. The entity is called EmissionActivity and has the following attributes that could be defined in an OPC UA information model.

?

EmissionActivity

?

??????? ActivityName – String (e.g. Fuel Sold to Consumers)

??????? ActivityMeasurement

??????? DeviceId – String (physical and virtual measurement device tag name)

??????? MeasurementValue - Int

??????? DateTime – DateTime

??????? UnitofMeasurement – String (mmscf)

??????? ActivityLocation

??????? OrganizationId – String (EnergyCorpLNG)

??????? FacilityId – String (RgeProduction)

??????? AssetId – String (XYZPipeline)

??????? EquipmentId – String (Compressor)

??????? ActivityCategory

??????? Scope – Enum [Scope1, Scope2, Scope3]

??????? Category – Enum [Scope3Category1, …, Scope3Category15]

?The OPC UA Nodeset Defining the Data Model

?As the last step, I have defined a Nodeset file for the EmissionActivity Object Type for OPC UA. The model was created using the open-source Python based opcua-asyncua library. I have setup a simple opc ua server code to simulate an EmissionActivity object, however the proper way forward would be to create a custom data type which will be used in CCS Companion Specification. Furthermore, this is a very Open Footprint centric model, we’ll look into extending the model to support other use cases.

In the next post, I will build an OPC UA server generating sample data and populate the Fabric OneLake tables. Stay tuned and as always looking forward for feedback.

Appendix - opc ua server code for simulations

import logging
import asyncio

from asyncua import ua, Server
from asyncua.common.structures104 import new_struct, new_enum, new_struct_field

logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger('asyncua')

async def main():
    # setup our server
    server = Server()
    await server.init()
    server.set_endpoint('opc.tcp://0.0.0.0:4840/emissions/server/')

    # setup our own namespace, not really necessary but should as spec
    uri = 'https://emissions.freeopcua.github.io'
    idx = await server.register_namespace(uri)


    # First a folder to organise our nodes
    emissions = await server.nodes.objects.add_folder(idx, "EmissionActivities")
    emissionsfolder = await emissions.add_folder(idx, "RgeProductionMethaneVenting001")

    escope = await new_enum(server, idx, "Scope", ["Scope1","Scope2","Scope3",])
    ecategory = await new_enum(server, idx, "Category", ["Scope3Category1","Scope3Category2","Scope3Category3",    "Scope3Category4","Scope3Category5","Scope3Category6",
"Scope3Category7","Scope3Category8","Scope3Category9",     "Scope3Category10","Scope3Category11","Scope3Category12",
"Scope3Category13","Scope3Category14","Scope3Category15",])

    custom_objs = await server.load_data_type_definitions()
    activityname = await emissionsfolder.add_variable(idx, "ActivityName", "Fuel Sold to Consumers")

    activitymeasurement = await server.nodes.base_object_type.add_object_type(idx, "MeasurementDevice")
    await (await activitymeasurement.add_property(idx, "DeviceId", "MethaneFugitive")).set_modelling_rule(True)
    await (await activitymeasurement.add_variable(idx, "MeasurementValue", 1.0)).set_modelling_rule(True)
    await (await activitymeasurement.add_property(idx, "UnitofMeasurement", "mmscf")).set_modelling_rule(True)
    measurementdevice = await emissionsfolder.add_object(idx, "ActivityMeasurement", activitymeasurement)

    activitylocation = await server.nodes.base_object_type.add_object_type(idx, "ActivityLocation")
    await (await activitylocation.add_property(idx, "OrganizationId", "EnergyCorpLNG")).set_modelling_rule(True)
    await (await activitylocation.add_property(idx, "FacilityId", "RgeProduction")).set_modelling_rule(True)
    await (await activitylocation.add_property(idx, "AssetId", "XYZPipeline")).set_modelling_rule(True)
    await (await activitylocation.add_property(idx, "EquipmentId", "Compressor")).set_modelling_rule(True)        
    await emissionsfolder.add_object(idx, "ActivityLocation", activitylocation)

    activitycategory = await server.nodes.base_object_type.add_object_type(idx, "ActivityCategory")
    await (await activitycategory.add_variable(idx, "Scope", ua.Scope.Scope3,  datatype=escope.nodeid)).set_modelling_rule(True)
    await (await activitycategory.add_variable(idx, "Category", ua.Category.Scope3Category1,  datatype=ecategory.nodeid)).set_modelling_rule(True)
    await emissionsfolder.add_object(idx, "ActivityCategory", activitycategory)

    await server.export_xml([server.nodes.objects, server.nodes.root, activityname, measurementdevice, activitylocation, activitycategory, escope, ecategory], "EmissionNodeset.xml", export_values=True)

    async with server:
        while True:
            await asyncio.sleep(1)

if __name__ == '__main__':
    asyncio.run(main())        

?