Skip to content

Simple Time Synchronization

In this recipe, we will demonstrate how to implement a simple time synchronization mechanism using oneM2M. This is useful for applications that require consistent time across devices, but do not need high precision and want to avoid the complexity of more sophisticated protocols or extra hardware.

Background

oneM2M supports various time synchronization mechanisms, including the use of the resource. However, for many applications, a simple approach using a RETRIEVE request to retrieve the current time from a CSE is sufficient.

The <CSEBase> resource in oneM2M contains a currentTime (short name ctm) attribute that can be used to retrieve the current time from the CSE. This attribute is updated by the CSE and can be accessed by applications.

Step-by-Step

The first step is to register an <AE> resource with the CSE. This can be omitted if the application is already registered.

The next step is to retrieve the current time from the CSE. The code below demonstrates two methods for doing this:

  • Retrieving the whole <CSEBase> resource and extracting the currentTime attribute. This is useful if you want to access other attributes of the <CSEBase resource as well.
  • Directly retrieving the currentTime attribute using a partial retrieval request. This is more efficient if you only need the current time.

In both cases, the date and time are available in the ctm attribute of the <CSEBase> resource. The format of the time is ISO 8601, which is a standard format for representing date and time.

Source Code

The source code for this recipe is structured into several files, each serving a specific purpose. The main script retrieves the current time from the CSE and prints it. The other files contain setup variables and functions for connecting with the CSE.

Main script
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# Import the setup variables
from setup import *                                     

# Import AE functions
from ae import register_AE, unregister_AE

# Import CSE functions
from cse import retrieve_CSE, retrieve_CSE_ctm_only

try:
    # Register an AE
    register_AE(application_name)

    # Retrieve the <CSEBase> resource and print the CSE's timestamp
    resource = retrieve_CSE(application_name)
    print(f'ISO 8601 timestamp from full <CSEBase> resource: {resource["m2m:cb"]["ctm"]}')

    # Retrieve the <CSEBase> resource, but only the "ctm" attribute, and print the
    # CSE's timestamp
    # Partial retrieve is useful when you only need one or a few attributes from a resource
    resource = retrieve_CSE_ctm_only(application_name)
    print(f'ISO 8601 timestamp from partial <CSEBase> resource: {resource["m2m:cb"]["ctm"]}')

except Exception as e:
    print(e)

finally:
    # Unregister the AE
    try:
        unregister_AE(application_name)
    except:
        pass
Setup variables and support functions
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import random, string

# Setup variables
cse_url = 'http://localhost:8080/~/id-in/cse-in'    # The url of the CSE
application_name = 'CMyApplication'                 # The name of the application entity
application_path = f'{cse_url}/{application_name}'  # The path of the application entity


def randomID() -> str:
    """ Generate an ID. Prevent certain patterns in the ID.

        Return:
            String with a random ID
    """
    return ''.join(random.choices(string.ascii_uppercase + string.digits + string.ascii_lowercase, k=10))
Functions for working with the CSE
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from setup import cse_url, randomID
import requests


def retrieve_CSE(originator:str) -> dict:
    """ Retrieve the full CSEBase resource.

        Args:
            originator: The originator of the request.

        Returns:
            A dictionary containing the full CSEBase resource.
    """
    # Set the oneM2M headers for retrieving the <AE> resource
    headers = {
        'Content-Type': 'application/json',         # Encoding
        'X-M2M-Origin': originator,                 # unique application entity identifier
        'X-M2M-RI': randomID(),                     # unique request identifier
        'X-M2M-RVI': '4' 
    }

    response = requests.get(cse_url, headers=headers)

    # Check the response
    match response.status_code:
        case 200:
            print('Full <CSEBase> retrieved successfully')
        case _:
            raise RuntimeError(f'Error retrieving CSE: {response.status_code}')
    return response.json()


def retrieve_CSE_ctm_only(originator:str) -> dict:
    """ Retrieve only the ctm attribute from the CSEBase resource (partial retrieve).

        Args:
            originator: The originator of the request.

        Returns:
            A dictionary containing the CSEBase resource with only the ctm attribute.
    """
    # Set the oneM2M headers for retrieving the <AE> resource
    headers = {
        'Content-Type': 'application/json',         # Encoding
        'X-M2M-Origin': originator,                 # unique application entity identifier
        'X-M2M-RI': randomID(),                     # unique request identifier
        'X-M2M-RVI': '4' 
    }

    # Retrieve the <CSEBase> resource with only the ctm attribute
    # The 'atrl' parameter is used to specify the attribute(s) to be retrieved
    response = requests.get(f'{cse_url}?atrl=ctm', headers=headers)

    # Check the response
    match response.status_code:
        case 200:
            print('Partial <CSEBase< retrieved successfully')
        case _:
            raise RuntimeError(f'Error retrieving CSE: {response.status_code}')
    return response.json()
Functions for working with <AE> resources
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from setup import cse_url, randomID
import requests

def register_AE(originator:str) -> None:
    """ Register an Application Entity

        Args:
            originator: The originator of the request
    """

    # Set the oneM2M headers for creating the <AE> resource
    headers = {
        'Content-Type': 'application/json;ty=2',    # Encoding and type of the resource to be created
        'X-M2M-Origin': originator,                 # unique application entity identifier
        'X-M2M-RI': randomID(),                     # unique request identifier
        'X-M2M-RVI': '4' 
    }

    # Define the <AE> resource
    body = {
        'm2m:ae': {
            'rn': 'CMyApplication',                 # Resource Name
            'api': 'Nmy-application.example.com',   # Application ID
            'rr': True,                             # Request Reachability
            'srv': ['4']                            # Supported Release Versions
        }
    }

    # Perform the http request to create the <AE> resource
    response = requests.post(cse_url, headers=headers, json=body)

    # Check the response
    match response.status_code:
        case 201:
            print('AE registered successfully')
        case _:
            raise RuntimeError(f'Error registering AE: {response.status_code}')


# Unregister AE
def unregister_AE(application_name:str) -> None:
    """ Unregister an Application Entity

        Args:
            originator: The originator of the request
    """

    # Set the oneM2M headers for deleting the <AE> resource
    headers = {
        'X-M2M-Origin': application_name,           # unique application entity identifier
        'X-M2M-RI': randomID(),                     # unique request identifier
        'X-M2M-RVI': '4' 
    }

    # Perform the http request to delete the <AE> resource
    response = requests.delete(f'{cse_url}/{application_name}', headers=headers)

    # Check the response
    match response.status_code:
        case 200:
                print('AE unregistered successfully')
        case 404:
            pass
        case _:
            raise RuntimeError(f'Error unregistering AE: {response.status_code}')

Example Output

When you run the main script, it will output the current time retrieved from the CSE. The output will look similar to this:

Example Console Output
$ python main.py
AE registered successfully
Full <CSEBase> retrieved successfully
ISO 8601 timestamp from full <CSEBase> resource: 20250608T182113,363146
Partial <CSEBase< retrieved successfully
ISO 8601 timestamp from partial <CSEBase> resource: 20250608T182113,374562
AE unregistered successfully

Summary

In this recipe, we demonstrated how to implement a simple time synchronization mechanism using oneM2M. It is a straightforward approach that allows applications to retrieve the current time from the CSE without the need for complex protocols or additional hardware.

by Andreas Kraft, 2025-06-08