Getting Machine Readable News (MRN) with Data Library for Python

    	
            
import sys
import datetime
import time
import json
import base64
import zlib
import binascii
import lseg.data as ld
from lseg.data.delivery import omm_stream
 
# Config the encoding for the console
sys.stdin.reconfigure(encoding='utf-8')
sys.stdout.reconfigure(encoding='utf-8')

        
        Copy
    

    	
            ld.open_session()
        
        Copy
    

    	
            
username = 'MACHINE-ID'
password = 'PASSWORD'
app_key = 'APP-KEY'
try:
    # Open the data session
    session = ld.session.platform.Definition(
        app_key=app_key,
        grant=ld.session.platform.GrantPassword(
            username=username,
            password=password
        ),
        signon_control=True
    ).get_session()
    ld.session.set_default(session)
    session.open()
except Exception as ex:
    print(f'Error in open_session: {str(ex)}')
    sys.exit(1)

        
        Copy
    

    	
            
try:
    # Open the data session
    session = ld.session.platform.Definition(
        app_key = 'APP-KEY',
        deployed_platform_host = 'RTDS HOST',
        deployed_platform_username = 'DACS USERNAME'
    ).get_session()
    ld.session.set_default(session)
    session.open()
except Exception as ex:
    print(f'Error in open_session: {str(ex)}')
    sys.exit(1)

        
        Copy
    

    	
            
RIC_CODE = 'MRN_STORY'
DOMAIN = 'NewsTextAnalytics'
SERVICE = 'ELEKTRON_DD'
 
# Create an OMM stream and register event callbacks
stream = omm_stream.Definition(
    name=RIC_CODE,
    domain=DOMAIN,
    service=SERVICE).get_stream()

        
        Copy
    

    	
            
def display_event(eventType, event):
    """Retrieve data: Callback function to display data or status events. """
    current_time = datetime.datetime.now().time()
    print('----------------------------------------------------------')
    print(f'>>> {eventType} event received at {current_time}')
    print(json.dumps(event, indent=2))
 
    return
 
# Define the event callbacks
# Refresh - the first full image we get back from the server
stream.on_refresh(lambda event, item_stream: display_event('Refresh', event))
 
# Status - if data goes stale or item closes, we get a status message
stream.on_status(lambda event, item_stream: display_event('Status', event))
 
# Other errors
stream.on_error(lambda event, item_stream: display_event('Error', event))

        
        Copy
    

    	
            
# list to contain the news envelopes
_news_envelopes = []
 
def process_mrn_update(event):
    """Process MRN Update messages."""
    message_json = event
    fields_data = message_json['Fields']
 
    # declare variables
    tot_size = 0
    guid = None
 
    # Get data for all requried fields
    fragment = base64.b64decode(fields_data['FRAGMENT'])
    frag_num = int(fields_data['FRAG_NUM'])
    guid = fields_data['GUID']

        
        Copy
    

    	
            
def process_mrn_update(event):
    ...
    # Get data for all required fields
    if frag_num > 1:  # We are now processing more than one part of an envelope - retrieve the current details
        pass
    else:  # FRAG_NUM = 1 The first fragment
        tot_size = int(fields_data['TOT_SIZE'])
        print(f'FRAGMENT length = {len(fragment)}')
        # The fragment news is not completed, waiting and add this news data to envelop object.
        if tot_size != len(fragment):
            print(f'Add new fragments to news envelop for guid {guid}')
            _news_envelopes.append({  # the envelop object is a Python dictionary with GUID as a key and other fields are data
                'GUID': guid,
                'data': {
                    'FRAGMENT': fragment,
                    'MRN_SRC': mrn_src,
                    'FRAG_NUM': frag_num,
                    'tot_size': tot_size
                }
            })
            return None
 

        
        Copy
    

    	
            
def process_mrn_update(event):
    ...
    # Get data for all required fields
    if frag_num > 1:  # We are now processing more than one part of an envelope - retrieve the current details
        guid_index = next((index for (index, d) in enumerate(_news_envelopes) if d['GUID'] == guid), None)
        envelop = _news_envelopes[guid_index]
        if envelop and envelop['data']['MRN_SRC'] == mrn_src and frag_num == envelop['data']['FRAG_NUM'] + 1:
            print(f'process multiple fragments for guid {envelop["GUID"]}')
 
            # Merge incoming data to existing news envelop and getting FRAGMENT and TOT_SIZE data to local variables
            fragment = envelop['data']['FRAGMENT'] = envelop['data']['FRAGMENT'] + fragment
            envelop['data']['FRAG_NUM'] = frag_num
            tot_size = envelop['data']['tot_size']
            print(f'TOT_SIZE = {tot_size}')
            print(f'Current FRAGMENT length = {len(fragment)}')
 
            # The multiple fragments news are not completed, waiting.
            if tot_size != len(fragment):
                return None
            # The multiple fragments news are completed, delete associate GUID envelop
            elif tot_size == len(fragment):
                del _news_envelopes[guid_index]
        else:
            print(f'Error: Cannot find fragment for GUID {guid} with matching FRAG_NUM or MRN_SRC {mrn_src}')
            return None
        else:  # FRAG_NUM = 1 The first fragment
            ...  

        
        Copy
    

    	
            
def process_mrn_update(ws, message_json):  # process incoming News Update messages
    ...
    # Get data for all required fields
    ...
    if frag_num > 1:  # We are now processing more than one part of an envelope - retrieve the current details
        ...
    else:  # FRAG_NUM = 1 The first fragment
        ...
    # News Fragment(s) completed, decompress and print data as JSON to console
    if tot_size == len(fragment):
        print(f'decompress News FRAGMENT(s) for GUID {guid}')
        decompressed_data = zlib.decompress(fragment, zlib.MAX_WBITS | 32)
        print(f'News = {json.loads(decompressed_data)}')

        
        Copy
    

    	
            
# Update - as and when field values change, we receive updates from the server and process the MRN data
stream.on_update(lambda event, item_stream: process_mrn_update(event))

        
        Copy
    

    	
            
# Send request to server and open stream
stream.open()
# We should receive the initial Refresh for the current field values
# followed by updates for the fields as and when they occur

        
        Copy
    

    	
            
>>> Refresh event received at 14:04:40.076276
{
  "ID": 5,
  "Type": "Refresh",
  "Domain": "NewsTextAnalytics",
  "Key": {
    "Service": "ELEKTRON_DD",
    "Name": "MRN_STORY"
  },
  "State": {
    "Stream": "Open",
    "Data": "Ok"
  },
  "Qos": {
    "Timeliness": "Realtime",
    "Rate": "JitConflated"
  },
  "PermData": "AwEBEAAc",
  "SeqNumber": 26414,
  "Fields": {
    "PROD_PERM": 10001,
    "ACTIV_DATE": "2025-01-04",
    "RECORDTYPE": 30,
    "RDN_EXCHD2": "MRN",
    "TIMACT_MS": 65276147,
    "GUID": null,
    "CONTEXT_ID": 3752,
    "DDS_DSO_ID": 4232,
    "SPS_SP_RIC": ".[SPSML2L1",
    "MRN_V_MAJ": "2",
    "MRN_TYPE": "STORY",
    "MDU_V_MIN": null,
    "MDU_DATE": null,
    "MRN_V_MIN": "10",
    "MRN_SRC": "HK1_PRD_A",
    "MDUTM_NS": null,
    "FRAG_NUM": 1,
    "TOT_SIZE": 0,
    "FRAGMENT": null
  }
}

        
        Copy
    

    	
            
>>> Update event received at 14:05:40.076276
{
  "ID": 5,
  "Type": "Update",
  "Domain": "NewsTextAnalytics",
  "UpdateType": "Unspecified",
  "DoNotConflate": true,
  "DoNotRipple": true,
  "DoNotCache": true,
  "Key": {
    "Service": "ELEKTRON_DD",
    "Name": "MRN_STORY"
  },
  "PermData": "AwEBEBU8",
  "SeqNumber": 54958,
  "Fields": {
    "TIMACT_MS": 26035856,
    "ACTIV_DATE": "2025-01-06",
    "MRN_TYPE": "STORY",
    "MRN_V_MAJ": "2",
    "MRN_V_MIN": "10",
    "TOT_SIZE": 716,
    "FRAG_NUM": 1,
    "GUID": "Pt3BVKYG__2501062wgGRnlW/ulmGvzHnd35kqZ6qc0/chruUU4ph2",
    "MRN_SRC": "HK1_PRD_A",
    "FRAGMENT": "H4sIAAAAAAAC/8VUUW/TMBB+51ec8gRS0ybp2iG/tdlWCmsa2oxpY2hyE6cxuHbmOB0B8d85J+0mISbxxov93dm++8732T8dKsw8c4gjYzOcfvpwM3N6Dq0zzmTKKod8dqKYxNML9HYgjJwvPWejsgYPeeP3VAajAXiBP4TJAs7XycA7RTxbJHfSdY+YQCyoMRVM4kkIU64wQcUExEVT8ZQKAknBXlqEUiskUwGvIKcpF9xQwzIwhVb1tsCZAbvmMlOP8DouaMVgPn8DhqWFVEJtmz6EqpaG6ZJq00CqWcaNjaZKJiFXGqgQkKpdSSWmxoDUANUMpDJgz/CUl9RwuQUu23RHRhnXLDWieeJy4NGHeQ6NqtsoVb3ZcdMeR5fGGJhyh/GUfDpHJVhSlorMoKD7LjkVmtGssZjnHGs+XKPKcR9W8peyNKuM5qmNXvWgFMzeB/Ld84zBo0YiWHMXL+04YLF2lwElGRSWYak5MjGqrbUyGN5mtMZiGT6342USORfoqkAy2ycFG+yQpBuBFgbeYYdtNX04Jf6QjEb98TA4SuZQIcogr1EEE8G0cd07iQrMua5MiDeC3Uf1BV4wcj3f9caJdwx0OgpucWeBtya4ZLgrvpwkyZqA77qY7r/pEElxS/r4zO7vg5Hne+PgcTtbSXE9qMVutv/xTmbD0beH2/FD6g3SQtdXVydlEdjTEhthH+Uyx2eJb1BQua3p1tbI7O3skBqaSVOiK0Cb7w6GY9h3MygF5XbfQQsa/dGaxJfJ2jrrzRqrqfHJO5jHkLqy7cIVVO9X1Hj7F0zICXoWxP941s5BmOB8Rd62Y9iOi3aMcFzZ/8LzPP8ZP8Ew8v6wrBEFZDJbzTs0nS8vOhSeJTcHtFx04PI86kC0mnVgFV3bn8nQb2zNHmr7fznE7zm13iLG32rYc/aoSVT8P0jo16vf3flwkBsFAAA="
  }
}
FRAGMENT length = 716
decompress News FRAGMENT(s) for GUID Pt3BVKYG__2501062wgGRnlW/ulmGvzHnd35kqZ6qc0/chruUU4ph2
News = {'altId': 'nPt3BVKYG', 'audiences': ['NP:PBF', 'NP:PBFCN'], 'body': '06Jan25/ 0213 AM EST/0713 GMT\n--0713 GMT: Platts APAC Biodiesel Physical: The APAC Biodiesel Physical process is facilitated through the eWindow (Phase II) technology. Counterparty credit is open for all companies that are not participating in the process directly through eWindow. If you are submitting your information through an editor and have not already notified Platts of any counterparty credit restrictions, please provide written notification at least one hour prior to the start of the MOC process if any counterparty credit filters need to be enabled or modified. 7:13:55.632 GMT\n--Platts Biofuel Alert--\n', 'firstCreated': '2025-01-06T07:13:55.752Z', 'headline': 'PLATTS: 1-- 7: Platts APAC Biodiesel Physical: The APAC Biodiesel Physical process is facilitated through the eWindow (Phase II) techno', 'id': 'Pt3BVKYG__2501062wgGRnlW/ulmGvzHnd35kqZ6qc0/chruUU4ph2', 'instancesOf': [], 'language': 'en', 'messageType': 2, 'mimeType': 'text/plain', 'provider': 'NS:PLTS', 'pubStatus': 'stat:usable', 'subjects': ['A:4', 'M:1QD', 'M:2CT', 'U:8', 'U:C', 'U:M', 'U:N', 'R:PBF0001', 'R:PBF001', 'R:PBFCN0001', 'R:PBFCN001', 'N2:AGRI', 'N2:BIOF', 'N2:CDTY', 'N2:COM', 'N2:LEN', 'N2:NRG', 'N2:RNW'], 'takeSequence': 1, 'urgency': 3, 'versionCreated': '2025-01-06T07:13:55.752Z'}

        
        Copy
    

    	
            
# Close the stream
stream.close()
# Close the session
ld.close_session()

        
        Copy