Thermo Fisher Momentum Adapter Page#

This page has all the information specific to the Thermo Fisher Momentum adapter.

Supported Momentum versions#

Thermo Fisher Momentum 7.3

Momentum Driver Installer#

No driver is needed for the Momentum adapter.

Momentum Adapter pyproject.toml Dependencies#

Run uv add artificial-momentum-resource-library=0.1.* to add the required packages to the [project] section.

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

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

Momentum Adapter config.yaml File#

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 IPv4 URL can be obtained on the device 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 Hardware-Specific Adapter Setup for how to do this.

  4. Update the artificial section to the correct host and lab. Consult the corresponding step in Hardware-Specific Adapter Setup for how to do this.

Momentum Adapter adapter/main/plugin.py File#

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()

Momentum Adapter adapter/main/__main__.py File#

Copy the following list of actors into the file in place of the example actors.

actors = [
    ActorConfig(id='momentum', abilities={'momentum': 1, 'exclusive_locks_global': 1, 'exclusive_locks_momentum': 1}),
]

Momentum Simple Connectivity Workflow#

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

After copying this code into a file, remember to replace the lab_guid in the workflow heading with the one from your Lab (most easily found by opening the Lab in Artificial and looking in the browser address bar). Consult Adding and Running a Simple Connectivity Test Workflow for more info.

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

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