openrvdas.logger.transforms.count_transform

 1#!/usr/bin/env python3
 2
 3import logging
 4from typing import Union
 5
 6
 7from logger.utils.das_record import DASRecord  # noqa: E402
 8from logger.transforms.transform import Transform  # noqa: E402
 9
10
11################################################################################
12#
13class CountTransform(Transform):
14    """Return number of times the passed fields have been seen as a dict
15    (or DASRecord, depending on what was passed in) where the keys are
16    'field_name:count' and the values are the number of times the passed
17    in fields have been seen. E.g:
18    ```
19    counts = CountTransform()
20    counts.transform({'f1': 1, 'f2': 1.5}) -> {'f1:count':1, 'f2:count':1}
21    counts.transform({'f1': 1}) -> {'f1:count':2}
22    counts.transform({'f1': 1.1, 'f2': 1.4}) -> {'f1:count':3, 'f2:count':2}
23    ```
24    """
25
26    def __init__(self, **kwargs):
27        """
28        """
29        super().__init__(**kwargs)  # processes 'quiet' and type hints
30
31        self.counts = {}
32
33    ############################
34    def transform(self, record: Union[DASRecord, dict]) -> Union[DASRecord, dict]:
35        """Return counts of the previous times we've seen these field names."""
36
37        # See if it's something we can process, and if not, try digesting
38        if not self.can_process_record(record):  # BaseModule
39            return self.digest_record(record)  # BaseModule
40
41        if type(record) is DASRecord:
42            fields = record.fields
43        elif type(record) is dict:
44            fields = record
45        else:
46            logging.warning('Input to CountTransform must be either '
47                            'DASRecord or dict. Received type "%s"', type(record))
48            return None
49
50        new_counts = {}
51
52        for field, value in fields.items():
53            if field not in self.counts:
54                self.counts[field] = 1
55            else:
56                self.counts[field] += 1
57            new_counts[field + ':count'] = self.counts[field]
58
59        if type(record) is DASRecord:
60            if record.data_id:
61                data_id = record.data_id + '_counts' if record.data_id else 'counts'
62            return DASRecord(data_id=data_id,
63                             message_type=record.message_type,
64                             timestamp=record.timestamp,
65                             fields=new_counts)
66
67        return new_counts