sift_py.data_import.ch10

  1import json
  2from typing import Any, Dict, Optional
  3
  4import requests
  5
  6from sift_py.data_import.config import CsvConfig
  7from sift_py.data_import.csv import CsvUploadService
  8from sift_py.data_import.status import DataImportService
  9from sift_py.data_import.time_format import TimeFormatType
 10
 11
 12class BaseCh10File:
 13    """
 14    Base class for uploading IRIG Chapter 10/Chapter 11 files.
 15
 16    Implement a concrete version of this class that parses a ch10 stream and returns
 17    a csv row of data on each iteration.
 18
 19    Set `gzip` to `True` if sending a compressed stream.
 20
 21    Example:
 22    ```python
 23
 24    class Ch10(BaseCh10File):
 25
 26        def __init__(self, path):
 27            self.file = open(path, "rb")
 28            self.initialize_csv_data_columns = None
 29
 30        def initialize_csv_data_columns(self):
 31            self.csv_config_data_columns = self.process_ch10_computer_f1_packet()
 32
 33        def process_ch10_computer_f1_packet(self) -> Dict[int, dict]:
 34            # Processes the first Computer F1 packet
 35            # and returns the measurements as a dict.
 36            ...
 37
 38        def process_ch10_pcm_packet(self) -> str:
 39            # Processed the data packets and returns
 40            # a CSV row.
 41            ...
 42
 43        def __next__(self) -> str:
 44            # On all iterations, return data for the CSV file.
 45            if end_of_file:
 46                raise StopIteration()
 47            else:
 48                return self.process_ch10_data_packet()
 49    ```
 50    """
 51
 52    csv_config_data_columns: Dict[int, dict]
 53    gzip: bool = False
 54
 55    def initialize_csv_data_columns(self) -> None:
 56        """
 57        Must populate the `csv_config_data_columns` attribute
 58        that is the data_columns entry in the CsvConfig.
 59
 60        See the Sift data_import module or API docs for the schema.
 61        """
 62        raise NotImplementedError
 63
 64    def __iter__(self):
 65        return self
 66
 67    def __next__(self) -> str:
 68        raise NotImplementedError
 69
 70
 71class Ch10UploadService(CsvUploadService):
 72    """Service to upload ch10 files."""
 73
 74    def upload(  # type: ignore
 75        self,
 76        ch10_file: BaseCh10File,
 77        asset_name: str,
 78        time_format: TimeFormatType = TimeFormatType.ABSOLUTE_UNIX_NANOSECONDS,
 79        run_name: Optional[str] = None,
 80        run_id: Optional[str] = None,
 81    ) -> DataImportService:
 82        """
 83        Uploads the ch10 file to the specified asset.
 84
 85        Override `time_format` to specify the time data format. Default is `TimeFormatType.ABSOLUTE_UNIX_NANOSECONDS`.
 86        Override `run_name` to specify the name of the run to create for this data. Default is None.
 87        Override `run_id` to specify the id of the run to add this data to. Default is None.
 88        """
 89        ch10_file.initialize_csv_data_columns()
 90
 91        assert getattr(ch10_file, "csv_config_data_columns"), (
 92            "`csv_config_data_columns` was not set correctly on the first iteration"
 93        )
 94
 95        config_info: Dict[str, Any] = {
 96            "asset_name": asset_name,
 97            "first_data_row": 2,
 98            "time_column": {
 99                "format": time_format,
100                "column_number": 1,
101            },
102            "data_columns": ch10_file.csv_config_data_columns,
103        }
104        if run_name:
105            config_info["run_name"] = run_name
106
107        if run_id:
108            config_info["run_id"] = run_name
109
110        csv_config = CsvConfig(config_info)
111
112        response = requests.post(
113            url=self._upload_uri,
114            headers={
115                "Authorization": f"Bearer {self._apikey}",
116                "Content-Encoding": "application/octet-stream",
117            },
118            data=json.dumps({"csv_config": csv_config.to_dict()}),
119        )
120
121        if response.status_code != 200:
122            raise Exception(
123                f"Config file upload request failed with status code {response.status_code}. {response.text}"
124            )
125
126        try:
127            upload_info = response.json()
128        except (json.decoder.JSONDecodeError, KeyError) as e:
129            raise Exception(f"Invalid response: {response.text}.\n{e}")
130
131        try:
132            upload_url: str = upload_info["uploadUrl"]
133            data_import_id: str = upload_info["dataImportId"]
134        except KeyError as e:
135            raise Exception(
136                f"Response missing required keys: {e}. This is unexpected. Please reach out to the Sift team about this error."
137            )
138
139        headers = {
140            "Authorization": f"Bearer {self._apikey}",
141        }
142
143        if ch10_file.gzip:
144            headers["Content-Encoding"] = "gzip"
145
146        response = requests.post(
147            url=upload_url,
148            headers=headers,
149            data=ch10_file,
150        )
151
152        if response.status_code != 200:
153            raise Exception(
154                f"Data file upload request failed with status code {response.status_code}. {response.text}"
155            )
156
157        return DataImportService(self._rest_conf, data_import_id)