Started building a file processor for cosmos data

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "oracledb",
     "pandas",
     "platformdirs",
+    "pydantic",
     "setuptools",
 ]
 name = "iot-swarm"

src/iotswarm/formatters.py

@@ -0,0 +1,42 @@
+from inspect import signature
+from typing import List
+from iotswarm.readers import Column, Metadata
+import pandas as pd
+
+def cr1000x(metadata, columns: List[Column], dataframe: pd.DataFrame):
+    """Processes data from a CR1000X device.
+
+    Args:
+        metadata: Metadata for the dataset.
+        columns: Columns in the dataset.
+        dataframe: DataFrame with the data.
+    """
+
+    payload = dict()
+    payload["head"] = {
+        "transaction": metadata.id,
+        "signature": metadata.signature,
+        "environment": {
+            "station_name": metadata.station_name,
+            "table_name": metadata.dataset,
+            "model": metadata.device_type,
+            "serial_number": metadata.serial_no,
+            "os_version": metadata.os_version,
+            "prog_name": metadata.program_name,
+        }
+    }
+
+    payload["data"] = []
+
+    for line in dataframe.iterrows():
+        row = {
+            "data": {
+                "time": line[1]["TIMESTAMP"],
+                "vals": {
+                    column: line[column] for column in dataframe.columns
+                }
+            }
+        }
+        payload["data"].append(row)
+
+    for col in columns

src/iotswarm/readers.py

@@ -0,0 +1,81 @@
+from io import StringIO
+from os import PathLike
+import os
+from pathlib import Path
+from typing import Dict, Tuple
+import pandas as pd
+from pydantic import BaseModel
+import csv
+from devices import CR1000XDevice
+
+class Metadata(BaseModel):
+    """Represents metadata for a dataset."""
+
+    id: str
+    station_name: str
+    device_type: str
+    logger_id: int
+    os_version: str
+    program_name: str
+    signature: int
+    dataset: str
+
+class Column(BaseModel):
+    """Represents a column in the dataset."""
+    name: str
+    type: str
+    unit: str
+    process: str
+    settable: bool = False
+
+
+class CosmosReader:
+    """Reads and processes data from a given file."""
+
+    def __init__(self, file: PathLike):
+        """Initializes the CosmosReader with the given file.
+
+        Args:
+            file: The path to the file to be read.
+        """
+        self.file = file
+
+    def read(self) -> Tuple[Metadata, Dict[str,Column], pd.DataFrame]:
+        """Reads the file and processes its content.
+
+        Returns:
+            A tuple containing metadata, columns, and a DataFrame with the data.
+        """
+        with open(self.file, "r") as f:
+            lines = f.readlines()
+
+        metadata, headers, units, processes = list(csv.reader(StringIO("".join(lines[0:4]))))
+        data_df = pd.read_csv(StringIO(''.join(lines[4:])), names=headers)
+
+        columns = {}
+        for h, u, p in zip(headers, units, processes):
+            columns[h] = Column(unit=u, process=p)
+
+        meta = Metadata(
+            id=metadata[0],
+            station_name=metadata[1],
+            device_type=metadata[2],
+            logger_id=metadata[3],
+            os_version=metadata[4],
+            program_name=metadata[5],
+            signature=metadata[6],
+            dataset=metadata[7]
+        )
+
+        return (meta, columns, data_df)
+
+
+if __name__ == "__main__":
+
+    precip_dir = Path(__file__).parent / "__assets__" / "sensor_data" / "PRECIP"
+    precip_files = [precip_dir / f for f in os.listdir(precip_dir) if f.endswith(".dat")]
+    metadata, columns, df = CosmosReader(precip_files[0])
+
+    device = CR1000XDevice(
+        device_id = 
+    )