wip updating tx module

Drafting a main script for examplifying main components. This include:
* gmail client
* iridium message parsing
* Binary payload parsing

wip tx processing
* Implemented fully functional binary payload decoder
* Added parsing exception with handling examples

Added various comments

Author: ladsmund

src/pypromice/tx/gmail_client.py

@@ -0,0 +1,122 @@
+import email.parser
+import imaplib
+import logging
+from configparser import ConfigParser
+from datetime import datetime
+from mailbox import Message
+from pathlib import Path
+from typing import Iterator
+
+__all__ = ["GmailClient"]
+
+logger = logging.getLogger(__name__)
+
+class GmailClient:
+
+    def __init__(self, server: str, port: int, account: str, password: str, last_uid: int = 1, chunk_size: int = 500):
+
+        self.last_uid = last_uid
+        self.chunk_size = chunk_size
+        self.parser = email.parser.Parser()
+        logger.info("Logging in to email server...")
+        self.mail_server = imaplib.IMAP4_SSL(server, port)
+        typ, accountDetails = self.mail_server.login(account, password)
+        if typ != "OK":
+            logger.error("Not able to sign in!")
+            raise Exception("Not able to sign in!")
+        logger.info("Logged in to email server")
+
+        result, data = self.mail_server.select(mailbox='"[Gmail]/All Mail"', readonly=True)
+        logger.info("mailbox contains %s messages" % data[0])
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.mail_server.logout()
+        return False
+
+
+    def fetch_mail(self, uid: int) -> Message | None:
+        result, data = self.mail_server.uid("fetch", str(uid), "(RFC822)")
+        if result != "OK":
+            logger.error(f"Error fetching mail with UID {uid}: {data}")
+            return None
+        if not data or len(data) < 2:
+            logger.error(f"No data returned for UID {uid}")
+            return None
+
+        mail_str = data[0][1].decode()
+        return self.parser.parsestr(mail_str)
+
+    def get_mail_ids(self, date: datetime|str) -> list[int]:
+        # Issue search command of the form "SEARCH UID 42:*"
+        if isinstance(date, str):
+            date = datetime.strptime(date, "%d-%b-%Y")
+
+        result, data = self.mail_server.uid("search", None, f"(ON {date.strftime('%d-%b-%Y')})")
+        if result != "OK":
+            logger.error(f"Error searching for mails on {date}: {data}")
+            return []
+        message_uids = list(map(int, data[0].split()))
+        logger.info(f"Found {len(message_uids)} messages on {date}")
+        return message_uids
+
+    def fetch_mails(self, uuids: list[int]) -> Iterator[Message]:
+        logger.info(f"Fetching {len(uuids)} mails")
+        batch_ids = ','.join(map(str, uuids))
+        result, data = self.mail_server.uid("fetch", batch_ids, "(RFC822)")
+        logger.info(f"Parsing string to Messages")
+        if result != "OK":
+            logger.info(f"Error fetching mails with UIDs {batch_ids}: {data}")
+            return
+        if not data or len(data) < 2:
+            logger.warning(f"No data returned for UIDs {batch_ids}")
+            return
+        for response in data:
+            if isinstance(response, tuple):
+                mail_str = response[1].decode()
+                message = self.parser.parsestr(mail_str)
+                yield message
+
+    def fetch_mails_chunked(self, uuids: list[int], chunk_size: int = 100) -> Iterator[Message]:
+        """
+        Fetch mails in chunks of size chunk_size.
+        """
+        for idx in range(0, len(uuids), chunk_size):
+            message_uid_chunk = uuids[idx:idx + chunk_size]
+            yield from self.fetch_mails(message_uid_chunk)
+
+    def fetch_new_mails(self, last_uid:int|None = None) -> Iterator[Message]:
+        if last_uid is None:
+            last_uid = self.last_uid
+
+        # Get the first 100 mails on 2025-01-01
+        result, data = self.mail_server.uid("search", None, f"(ON 01-Apr-2025)")
+        message_uuids = data[0].decode().split()
+        len(message_uuids)
+
+        # Issue search command of the form "SEARCH UID 42:*"
+        result, data = self.mail_server.uid("search", None, f"(UID {last_uid}:*)")
+        message_uids = list(map(int, data[0].split()))
+        logger.info(f"Found {len(message_uids)} new messages since UID {last_uid}")
+
+        for idx in range(0, len(message_uids), self.chunk_size):
+            message_uid_chunk = message_uids[idx:idx + self.chunk_size]
+            yield from self.fetch_mails(message_uid_chunk)
+
+
+    @classmethod
+    def from_config_files(cls, *init_file_path: Path) -> 'GmailClient':
+        """
+        Create a GmailClient instance from configuration files.
+
+        The latest path takes precedence over the previous ones.
+
+        """
+        configuration = ConfigParser()
+        for file_path in init_file_path:
+            configuration.read(file_path)
+
+        return cls(**configuration["aws"])
+

src/pypromice/tx/iridium.py

@@ -0,0 +1,64 @@
+import re
+from datetime import datetime
+from email.message import Message
+
+import attrs
+
+
+@attrs.define
+class IridiumMessage:
+    """
+    Class to represent an Iridium message.
+    """
+
+    imei: str
+    momsn: int
+    mtmsn: int
+    time_of_session: datetime
+    payload_filename: str
+    payload_bytes: bytes | None
+
+    # This is the original subject line of the email
+    # NOTE:
+    # This is not a part of the Iridium message, but rather a part of the email a layer above
+    # It is stored here for payload decoding purposes
+    subject: str | None
+
+
+def parse_mail(email: Message) -> IridiumMessage:
+    assert email["From"] == "sbdservice@sbd.iridium.com"
+    subject = str(email["Subject"])
+
+    sbd_raw = None
+    sdb_dict = None
+    data_message: bytes | None = None
+    filename = None
+    for payload_message in email.get_payload():
+        filename: str = payload_message.get_filename()
+        if filename is None:
+            # There should only be a single payload without a filename
+            assert sbd_raw is None
+            sbd_raw = payload_message.get_payload()
+            sdb_dict = dict(
+                re.findall(r"^([^:\n]+):\s*([^\n\r]+)", sbd_raw, re.MULTILINE)
+            )
+        else:
+            assert data_message is None
+            data_message = payload_message.get_payload(decode=True)
+
+    assert sdb_dict is not None
+    assert data_message is not None
+    assert filename is not None
+
+    date_string = sdb_dict["Time of Session (UTC)"]
+    date_object = datetime.strptime(date_string, "%a %b %d %H:%M:%S %Y")
+
+    return IridiumMessage(
+        imei=filename.split("_")[0],
+        momsn=int(sdb_dict["MOMSN"]),
+        mtmsn=int(sdb_dict["MTMSN"]),
+        time_of_session=date_object,
+        payload_bytes=data_message,
+        payload_filename=filename,
+        subject=subject,
+    )

src/pypromice/tx/payload_decoder.py

@@ -0,0 +1,191 @@
+from datetime import datetime
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+__all__ = [
+    "decode_payload",
+    "DecodeError",
+]
+
+CR_BASIC_EPOCH_OFFSET = datetime(1990, 1, 1, 0, 0, 0, 0).timestamp()
+
+# Load payload formats from CSV file
+payload_formats_path = Path(__file__).parent.joinpath("payload_formats.csv")
+payload_formats = pd.read_csv(payload_formats_path, index_col=0)
+payload_formats = payload_formats[payload_formats["flags"].values != "don’t use"]
+
+
+class DecodeError(Exception):
+
+    def __init__(
+        self,
+        format_index: int | None = None,
+        format_name: str | None = None,
+        letter_index: int | None = None,
+        buffer_index: int | None = None,
+        message: str | None = None,
+    ):
+        self.format_index = format_index
+        self.format_name = format_name
+        self.letter_index = letter_index
+        self.buffer_index = buffer_index
+        self.message = message
+
+    def as_dict(self) -> dict:
+        return {
+            "format_index": self.format_index,
+            "format_name": self.format_name,
+            "letter_index": self.letter_index,
+            "buffer_index": self.buffer_index,
+            "message": self.message,
+        }
+
+    def __str__(self):
+        return repr(self)
+
+    def __repr__(self):
+        variables_str = ", ".join(f"{k}={v}" for k, v in self.as_dict().items())
+
+        return f"{self.__class__.__name__}({variables_str})"
+
+
+def parse_gfp2(buffer: bytes) -> float:
+    """Two-bit floating point decoder"""
+    if len(buffer) < 2:
+        raise ValueError("Buffer too short for gfp2 decoding")
+
+    msb = buffer[0]
+    lsb = buffer[1]
+    Csign = -2 * (msb & 128) / 128 + 1
+    CexpM = (msb & 64) / 64
+    CexpL = (msb & 32) / 32
+    Cexp = 2 * CexpM + CexpL - 3
+    Cuppmant = (
+        4096 * (msb & 16) / 16
+        + 2048 * (msb & 8) / 8
+        + 1024 * (msb & 4) / 4
+        + 512 * (msb & 2) / 2
+        + 256 * (msb & 1)
+    )
+    Cnum = Csign * (Cuppmant + lsb) * 10**Cexp
+    return round(Cnum, 3)
+
+
+def parse_gli4(buffer: bytes) -> int:
+    """Four-bit decoder
+
+    Parameters
+    ----------
+    buffer : bytes
+        List of four values
+
+    Returns
+    -------
+    float
+        Decoded value
+    """
+    if len(buffer) < 4:
+        raise ValueError("Buffer too short for gli4 decoding")
+
+    return int.from_bytes(buffer[:4], byteorder="big", signed=True)
+
+
+def decode_payload(payload: bytes) -> list:
+
+    # The original code also had a special case for imei number 300234064121930. This should be handled differently.
+    bidx = ord(payload[:1])
+
+    # The payload format is determined by the first byte of the payload
+    if bidx not in payload_formats.index:
+        raise DecodeError(format_index=bidx, message=f"Unknown payload format: {bidx}")
+
+    payload_format = payload_formats.loc[bidx]
+    _, bin_format, bin_name = payload_format.iloc[:3]
+    # Note: bin_val is just len(bin_format)
+    indx = 1  # The first byte is the payload format
+    dataline: list = []
+
+    # Iterate over each token in the bin_format and decode the payload
+    format_letter_index = None
+    try:
+        for format_letter_index, type_letter in enumerate(bin_format):
+
+            if type_letter == "f":
+                nan_values = (8191,)
+                inf_values = (8190,)
+                neg_inf_values = -8190, -8191
+
+                value = parse_gfp2(payload[indx:])
+                # Be careful with float and int comparison. Float doesn't guarantee exact values.
+                # Consider using binary patterns for the special values instead.
+                if value in nan_values:
+                    dataline.append(np.nan)
+                elif value in inf_values:
+                    dataline.append(np.inf)
+                elif value in neg_inf_values:
+                    dataline.append(-np.inf)
+                else:
+                    dataline.append(value)
+                indx += 2
+                continue
+
+            if type_letter == "l":
+                # Encoded as a 4 byte two complement integer
+
+                value = parse_gli4(payload[indx:])
+
+                # The value 2147450879 seems arbitrary, but 2147483648 is the maximum value for a 4-byte signed integer
+                nan_values = -2147483648, 2147450879
+                if value in nan_values:
+                    dataline.append(np.nan)
+                else:
+                    # TODO: Integers doesn't support NaN values and this will cause mixed types which should be avoided
+                    dataline.append(float(value))
+                indx += 4
+                continue
+
+            if type_letter == "t":
+                value = parse_gli4(payload[indx:])
+                time = datetime.fromtimestamp(CR_BASIC_EPOCH_OFFSET + value)
+                dataline.append(time)
+                indx += 4
+                continue
+
+            # GPS time or coordinate encoding
+            if type_letter in ("g", "n", "e"):
+                nan_values_fp2 = (8191,)
+                # Check if byte is a 2-bit NAN. This occurs when the GPS data is not
+                # available and the logger sends a 2-bytes NAN instead of a 4-bytes value
+                if np.isnan(parse_gfp2(payload[indx:])) in nan_values_fp2:
+                    # This is a 2-byte NAN
+                    # The GPS data is not available
+                    # We need to skip the next 2 bytes
+                    dataline.append(np.nan)
+                    indx += 2
+                    continue
+                else:
+                    # This is a 4-byte value
+                    value = parse_gli4(payload[indx:])
+
+                    if type_letter == "g":
+                        value /= 100.0
+                    else:  # type_letter in ('n', 'e'):
+                        value /= 100000.0
+
+                    dataline.append(value)
+                    indx += 4
+                continue
+
+            raise Exception(f"Unknown type letter: {type_letter}")
+
+
+        # TODO: Compute and validate checksum
+        if indx != len(payload):
+            raise Exception(f"Payload length mismatch: {indx} != {len(payload)}")
+
+    except Exception as e:
+        raise DecodeError(bidx, bin_name, format_letter_index, indx) from e
+
+    return dataline

src/pypromice/tx/payload_types.csv

@@ -1,6 +1,6 @@
 type_letter,bytes,process,nan_values,inf_values,-inf_values,notes
 f,2,1,8191,8190,"-8190,-8191",value encoded as 2 bytes base-10 floating point (GFP2)
-l,4,1,"-2147483648,2147450879",,,value encoded as 4 bytes two's complement integer (GLI4)
+l,4,1,"-2147483648,2147450879",,,value encoded as 4 bytes two's complement integer (GLI4) - note the mac nan value is not a max value
 t,4,%Y-%m-%d %H:%M:%S,,,,timestamp as seconds since 1990-01-01 00:00:00 +0000 encoded as GLI4
 g,4,100,,,,GPS time encoded as GLI4
 n,4,100000,,,,GPS latitude encoded as GLI4