Connecting to Thermo Fisher Momentum#

This guide will walk you through how to connect Artificial to Thermo Fisher Momentum using the Momentum Resource Library.

Attention

Before you get started, make sure your local Adapter development environment is set up and ready to go.

Supported Momentum versions#

Thermo Fisher Momentum 7.3.1 with Unite 2.2.5

Note

No driver is needed for the Momentum Resource Library.

Install the Resource Library in your Adapter#

Run the following command in the terminal of your adapter repository to install Artificial’s Momentum Resource Library:

uv add artificial-momentum-resource-library==0.1.*

Your pyproject.toml file should now include artificial-momentum-resource-library in the project dependencies section:

[project]
dependencies = [
  ...
  "artificial-momentum-resource-library=0.1.*"
]

Update your Adapter Config#

config.yaml#
adapter:
  name: MomentumTutorialAdapter
  remoteConfig: False
  allowAnySequenceId: True # Useful when running in a local dev container
  plugin: # all resources this adapter can connect to
    host: "IP address" #  IP address of momentum PC
    port: 443 # port set on momentum webservice
    user: "xyz"
    password: "xyz"
  asset_sync:
    devices:
      "asset_name": { rid: "d1234567-34d0-4391-be64-7aef4e0b28be" } # asset_name: name of the asset from momentum inventory

  1. Fill in the correct URL host for the Momentum device. The device’s IPv4 URL can be obtained from Network & Internet Settings or by running ipconfig in a command window.

  2. Fill in the user and password, if needed.

  3. Under asset_sync, match at least one of your devices in the momentum inventory to a Digital Twin asset ID. You can fill out the rest of the devices later. Consult the corresponding step in Asset Sync Config for how to do this.

Use the Resource Library in your Adapter#

adapter/main/plugin.py file#
import asyncio

from artificial.adapter_common import ActionModulePlugin, action_modules
from artificial.adapter_common.plugin import PluginContext, plugin_config
from artificial.client.notificationapi.v1.notificationapi import Severity
from artificial.logging import get_logger
from artificial.momentum.actions import MomentumActions
from artificial.momentum.driver.momentum_api4 import MomentumAPI
from artificial.momentum.models import MomentumConfig, SyncConfig
from artificial.momentum.syncer import MomentumSyncer

logger = get_logger(__name__)


@action_modules(MomentumActions)
class AdapterPlugin(ActionModulePlugin):
    cfg = plugin_config(MomentumConfig)

    async def setup(self, pctx: PluginContext) -> None:
        api_base_url = f'https://{self.cfg.host}:{self.cfg.port}'
        logger.info(f'Connecting to Momentum API at: {api_base_url}')
        self._mom_api = MomentumAPI.create(api_base_url, self.cfg.user, self.cfg.password, verify_ssl=False)

        while True:
            try:
                await self._mom_api.ainit()
                break
            except Exception:
                await pctx.artificial_api.notification_api_v1.raise_notification(
                    action_id='',
                    lab_id=pctx.config.artificial.labId,
                    job_id='',
                    message='Failed to connect to Momentum API, retrying...',
                    severity=Severity.ERROR,
                )
                logger.exception('Failed to connect to Momentum API, retrying...')
                await asyncio.sleep(5.0)

        logger.info(f'Creating Momentum Module for instance at: {self.cfg.host}:{self.cfg.port!s}')

        sync_config: SyncConfig = pctx.raw_config.to_dataclass(SyncConfig, 'adapter.asset_sync')
        if len(sync_config.plate_types) == 0:
            logger.warning('empty plate type configuration, no plate templates will be synchronized')
        else:
            logger.info('Synchronizing Momentum inventory templates', names=list(sync_config.plate_types.keys()))
        syncer = pctx.asset_sync_manager_v1(f'ResourceSyncer-{pctx.config.adapter.name}')

        mom_syncer = MomentumSyncer(
            self._mom_api,
            pctx.artificial_api.notification_api_v1,
            pctx.config.artificial.labId,
            syncer,
            'MomentumSyncer',
            sync_config.plate_types,
            sync_config.plate_locations,
        )

        module = MomentumActions(self._mom_api, mom_syncer)

        await mom_syncer.link_devices(sync_config.devices)
        logger.info('Performing full sync after Adapter start')
        await mom_syncer.sync()
        await mom_syncer.set_health_monitors(pctx.lab.health, pctx.config.adapter.name)
        self.add_module(module)
        self.add_webroute(mom_syncer.routes())

    async def aclose(self) -> None:
        if self._mom_api:
            await self._mom_api.aclose()

Add the required Actors#

Add the following actors to the list of actors in adapter/main/__main__.py.

adapter/main/__main__.py#
actors = [
    ActorConfig(id='momentum', abilities={'momentum': 1, 'exclusive_locks_global': 1, 'exclusive_locks_momentum': 1}),
]

Publish and run a test Workflow#

This sample workflow will connect with Momentum, start the system, and then log a message in the UI upon successful completion.

from artificial.workflows.decorators import workflow
from artificial.workflows.runtime import show_info
from stubs.stubs_actions.artificial.momentum.actions import start_system

@workflow('Simple Momentum Connectivity Workflow', 'simple_momentum_connectivity_workflow')
async def simple_momentum_connectivity_workflow() -> None:
    await start_system(normal_mode=True)
    await show_info('Congratulations, you successfully ran your hardware!', 'Hardware Success')