Events Parser

Classes Event and Events work together to parse radio events and append to a csv file.

Usage Examples

Parsing the output of the get-events RPC into list of parsed events

>>> from pathlib import Path
>>> from tgtools.parsers.event import Event
>>> xml = ''' <rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"
                         xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0"
                         message-id="urn:uuid:fb6a10dc-1097-4e6d-8cb7-7524fc94d327">
                      <events xmlns="http://siklu.com/yang/tg/events">
                              <index>0</index>
                              <event>2024-10-13 23:31:03.784, 1388,SYSTEM, LOCAL,  RADIO,   cn1 link up; Local/Remote Sector: 3/1; Tx/Rx Beam: 32/32; Tx/Rx Az: 20/20 deg; Tx/Rx El: -10/-10 deg</event>
                      </events>
                      <events xmlns="http://siklu.com/yang/tg/events">
                              <index>1</index>
                              <event>2024-10-13 23:56:02.304, 1389,SYSTEM, LOCAL,  RADIO,   cn1 link down (was up for 00000:00:24:58); Cause: LinkNegotiationFailed; Local/Remote Sector: 3/1</event>
                      </events>
              </rpc-reply>
          '''
     >>> prefix_dict = {'route'='192.168.0.1', 'sampled':'2024-10-23 17:09:00'}
     >>> events = Events.from_xml(xml, prefix_dict)
     >>> print(events)
             Events:
                     {'route': '192.168.0.1', 'sampled': '2024-10-23 17:09:00', 'index': 0, 'timestamp': ' 2024-10-13 23:31:03.784', 'id': '1388', 'category': 'SYSTEM', 'device': 'LOCAL', 'type': 'RADIO', 'msg': 'cn1 link up; Local/Remote Sector: 3/1; Tx/Rx Beam: 32/32; Tx/Rx Az: 20/20 deg; Tx/Rx El: -10/-10 deg', 'power_event': '', 'link_name': 'cn1', 'link_event': 'link up', 'up_for': '', 'cause': ''}
                     {'route': '192.168.0.1', 'sampled': '2024-10-23 17:09:00', 'index': 1, 'timestamp': ' 2024-10-13 23:56:02.304', 'id': '1389', 'category': 'SYSTEM', 'device': 'LOCAL', 'type': 'RADIO', 'msg': 'cn2 link down (was up for 00000:00:24:58); Cause: LinkNegotiationFailed; Local/Remote Sector: 3/1', 'power_event': '', 'link_name': 'cn2', 'link_event': 'link down', 'up_for': '00000:00:24:58', 'cause': 'LinkNegotiationFailed'}
     >>> events.append_csv(Path('events.csv'))

Class Information:

class tgtools.parsers.events.events.Events

A processor/parser for the XML response of the get-events RPC, including:

separating to individual events, sorting chronologically, and parsing of each into Event,
appending to an existing csv file, ensuring no replication.

This class is an extension of list.

Instantiation is typically via the class method from_xml().

append_csv(path_file: Path, threading_lock: lock | None = None)

Save class instance (list of events) to a csv file. If the file does not exist, it is created. But if the file does already exist (includes previous events), then only the new (not yet saved) events are appended. This avoids duplication in the output csv file.

Parameters:

path_file (pathlib.Path) – Path to csv file.
threading_lock (threading.Lock | None) – Threading lock, if needing to ensure method is thread-safe (Default: None).

calc_xml_etree(xml: str) → ElementTree

Convert the xml output of the get-events RPC into an ElementTree (thereby validating xml syntax). This method is not typically called directly, but rather implicitly by calling from_xml().

Parameters:: xml (str) – The xml output of the get-events RPC.
Returns:: An etree representation of the xml input.
Return type:: lxml.ElementTree

classmethod from_csv(filepath: Path) → Self

Create a class instance by reading parsed events from csv file.

Parameters:: filepath (pathlib.Path) – Path to csv file
Returns:: Class instance.
Return type:: Self

classmethod from_xml(xml: str, prefix_dict: dict[str, str] | None = None) → Self

Create class instance by parsing the xml output of the get-events RPC.

Parameters:

xml (str) – The xml string output returned by the get-events RPC.
prefix_dict (dict | None) – an optional arbitrary dictionary, prepended to each parsed event (default: None).

Returns:

a class instance.

Return type:

Self

parse_xml_etree(xml_etree: ElementTree, prefix_dict: dict[str, str] | None = None)

Parse each element of ElementTree to Event, and append to class instance. Finally, sort chronologically. This method is not typically called directly, but rather implicitly by calling from_xml().

Parameters:

xml_etree (etree.ElementTree) – an ElementEtree representing the xml output of the get-events RPC.
prefix_dict (dict[str, str] | None) – An arbitrary dictionary, typically identifying the source radio and computer’s timestamp (default: None).

errors: list[Err]: List of errors

class tgtools.parsers.events.event.Event

Representation for a single event fetched from a TG radio, including parsing. This class is an extension of dict.

An instance can be initialised:

from a string (refer to from_xml()), or
from a dictionary (refer to from_dict()).

classmethod from_dict(source_dict: dict) → Self

Create a class instance from a dictionary.

Parameters:: source_dict (dict) – Source dictionary
Returns:: A class instance
Return type:: Self

classmethod from_xml(event_string: str, event_index: int | None = None, prefix_dict: dict[str, str] | None = None) → Self

Return a class instance from an event string, for example:

"2024-10-13 23:31:03.784, 1388,SYSTEM, LOCAL,  RADIO,   cn1 link up; Local/Remote Sector: 3/1; Tx/Rx Beam: 32/32; Tx/Rx Az: 20/20 deg; Tx/Rx El: -10/-10 deg"

The event string is parsed, and prepended by additional optional tokens:

prefix_dict: an arbitrary dictionary, typically identifying the source radio and computer’s timestamp. Note: the keys of this dictionary must not replicate any tokens in core_tokens.

event_index: an integer designating the event number as returned by the get-events RPC.

Parameters:

event_string (str) – an event string - example shown above.
event_index (int | None) – an optional integer designating the event number as returned by the get-events RPC (default: None).
prefix_dict (dict | None) – an optional arbitrary dictionary, prepending the parsed event (default None).

Returns:

a class instance.

Return type:

Self

parse_xml(event_string: str, event_index: int | None = None)

Tokenise a string representing a TG event. The parsing tokens used are listed in core_tokens. This method is not typically called directly, but rather implicitly by calling from_xml().

Parsing errors appended to errors.

Parameters:

event_string (str) – an event string.
event_index (int | None) – an optional integer designating the event number as returned by the get-events RPC (default: None).

core_tokens = ('index', 'timestamp', 'id', 'category', 'device', 'type', 'msg', 'power_event', 'link_name', 'link_event', 'up_for', 'cause'): The principal parsing tokens.

errors: list[Err]: List of parsing errors

property prefix_dict: An optional arbitrtary dictionary whose key/value pairs are prepend do the parsed event.