openrvdas.logger.utils.sealog_event

  1#!/usr/bin/env python3
  2
  3import json
  4import pprint
  5from typing import Union, Optional
  6
  7from logger.utils.timestamp import time_str  # noqa:E402
  8from logger.utils.das_record import DASRecord  # noqa: E402
  9
 10
 11class SealogEvent:
 12    """SealogEvent is a structured representation of a Sealog event object.
 13    Class includes help methods like __str__, __eq__ and as_json
 14    """
 15
 16    ############################
 17    def __init__(self, json_str=None, event_value=None, event_author=None,
 18                 timestamp=None, event_free_text=None, event_options=None):
 19        """
 20        If a json string is passed, it is parsed into a dictionary and its
 21        values for timestamp, fields and metadata are copied in. Otherwise,
 22        the DASRecord object is initialized with the passed-in values for
 23        instrument, timestamp, fields (a dictionary of fieldname-value pairs)
 24        and metadata.
 25
 26        If timestamp is not specified, the instance will use the current time.
 27        """
 28        if json_str:
 29            parsed = json.loads(json_str)
 30            self.event_value = parsed.get('event_value')
 31            self.event_author = parsed.get('event_author')
 32            ts = parsed.get('timestamp')
 33            if isinstance(ts, int):
 34                self.timestamp = time_str(ts)
 35            else:
 36                self.timestamp = ts
 37            self.timestamp = parsed.get('timestamp')
 38            self.event_free_text = parsed.get('event_free_text', "")
 39            self.event_options = parsed.get('event_options', [])
 40        else:
 41            # self.source =
 42            self.event_value = event_value
 43            self.event_author = event_author
 44            self.timestamp = time_str(timestamp) if isinstance(timestamp, int) else timestamp
 45            self.event_free_text = event_free_text or ""
 46            self.event_options = event_options or []
 47
 48    ############################
 49    def as_json(self, pretty=False) -> str:
 50        """Return DASRecord as a JSON string."""
 51        json_dict = {
 52            'event_value': self.event_value,
 53            'event_author': self.event_author,
 54            'ts': self.timestamp,
 55            'event_free_text': self.event_free_text,
 56            'event_options': self.event_options
 57        }
 58
 59        json_dict = {k: v for k, v in json_dict.items() if v is not None}
 60
 61        if pretty:
 62            return json.dumps(json_dict, indent=4)
 63        else:
 64            return json.dumps(json_dict)
 65
 66    ############################
 67    def __str__(self):
 68        das_dict = {
 69            'event_value': self.event_value,
 70            'event_author': self.event_author,
 71            'ts': self.timestamp,
 72            'event_free_text': self.event_free_text,
 73            'event_options': self.event_options
 74        }
 75
 76        das_dict = {k: v for k, v in das_dict.items() if v is not None}
 77
 78        return pprint.pformat(das_dict)
 79
 80    ############################
 81    def __eq__(self, other) -> bool:
 82        return (other and
 83                self.event_value == other.event_value and
 84                self.event_author == other.event_author and
 85                self.timestamp == other.timestamp and
 86                self.event_free_text == other.event_free_text and
 87                self.event_options == other.event_options)
 88
 89
 90def to_event(record: Union[SealogEvent, DASRecord, str],
 91             configs: Optional[dict] = None) -> SealogEvent:
 92    """
 93    Uses an object of configs to translates a DASRecord object or json string to a SealogEvent
 94    object.
 95    """
 96
 97    if not configs:
 98        configs = {
 99            '_default': {
100                'event_free_text': "",
101                'field_map': None
102            }
103        }
104
105    if isinstance(record, SealogEvent):
106        return record
107
108    if isinstance(record, str):
109        event = SealogEvent(json_str=record)
110        return event
111
112    record = json.loads(record.as_json())
113
114    data_id = record.get("data_id", "_default")
115    config = configs.get(data_id) or configs.get('_default', {})
116
117    event_value = record['fields'].get("event_value") or config.get("event_value", 'FROM_OPENRVDAS')
118    event_author = record['fields'].get("event_author") or config.get("event_author")
119    event_free_text = record['fields'].get("event_free_text") or config.get("event_free_text", "")
120
121    ts_str = time_str(record["timestamp"]) if record.get("timestamp") else None
122
123    event_options = [
124        {"event_option_name": k, "event_option_value": v}
125        for k, v in config.get("event_options", {}).items()
126    ]
127
128    field_map = config.get("field_map", {})
129
130    record_fields = record.get("fields", {})
131
132    if not field_map:
133        event_options.extend([
134            {"event_option_name": k.lstrip("event_option_"), "event_option_value": v}
135            for k, v in record_fields.items()
136            if k.startswith("event_option_")
137        ])
138    else:
139        event_options.extend(
140            {"event_option_name": dst, "event_option_value": record_fields[f'event_option_{src}']}
141            for src, dst in field_map.items()
142            if f'event_option_{src}' in record_fields
143        )
144
145    event = SealogEvent(
146        event_value=event_value,
147        timestamp=ts_str,
148        event_author=event_author,
149        event_free_text=event_free_text,
150        event_options=event_options
151    )
152
153    return event