SpreadsheetSummaryFactory.py

import os

from obspy import UTCDateTime
import openpyxl
from typing import List

from .Absolute import Absolute
from . import Angle
from .Reading import Reading


def parse_relative_time(base_date: str, time: str) -> UTCDateTime:
    """Parse a relative date.

    Arguments
    ---------
    base_date: date when time occurs (YYYYMMDD)
    time: time on base_date (HHMMSS) or (HHMM)
    """
    try:
        time = "{0:04d}".format(time)
        return UTCDateTime(f"{base_date}T{time}")
    except Exception as e:
        print(f"error parsing relative date '{base_date}T{time}': {e}")
        return None


def convert_baseline(
    baseline: float = 0,
) -> float:
    """Convert basline from minutes to seconds"""
    try:
        return baseline / 60
    except:
        pass


class SpreadsheetSummaryFactory(object):
    """Read absolutes from summary spreadsheets"""

    def __init__(self, base_directory: str):
        self.base_directory = base_directory

    def get_readings(
        self, observatory: str, starttime: UTCDateTime, endtime: UTCDateTime
    ) -> List[Reading]:
        """Gathers readings from factory's base directory

        Attributes
        ----------
        observatory: 3-letter observatory code
        starttime: beginning date of readings
        endtime: end date of readings
        """
        readings = []
        start_filename = f"{observatory}{starttime.datetime:%Y%j%H%M}.xlsm"
        end_filename = f"{observatory}{endtime.datetime:%Y%j%H%M}.xlsm"
        for year in range(starttime.year, endtime.year + 1):
            observatory_directory = os.path.join(
                self.base_directory, observatory, f"{year}"
            )
            for (dirpath, _, filenames) in os.walk(observatory_directory):
                filenames.sort()
                for filename in filenames:
                    if start_filename <= filename < end_filename:
                        rs = self.parse_spreadsheet(
                            os.path.join(dirpath, filename),
                        )
                        for r in rs:
                            readings.append(r)
        return readings

    def parse_spreadsheet(self, path: str) -> List[Reading]:
        sheet = openpyxl.load_workbook(path, data_only=True)["Sheet1"]
        readings = self._parse_readings(sheet, path)
        return readings

    def _parse_metadata(self, sheet: openpyxl.worksheet) -> dict:
        """gather metadata from spreadsheet

        Attributes
        ----------
        sheet: excel sheet containing residual summary values
        observatory: 3-letter observatory code
        """
        date = sheet["I1"].value
        date = f"{date.year}{date.month:02}{date.day:02}"
        return {
            "station": sheet["D49"].value[0:3],
            "pier_correction": sheet["C5"].value,
            "instrument": sheet["B3"].value,
            "date": date,
            "observer": sheet["I10"].value,
        }

    def _parse_readings(self, sheet: openpyxl.worksheet, path: str) -> List[Reading]:
        """get list of readings from spreadsheet

        Attributes
        ----------
        sheet: excel sheet containing residual summary values
        path: spreadsheet's filepath

        Outputs
        -------
        List of readings from spreadsheet.
        Readings are returned regardless of if they are valid.
        """
        metadata = self._parse_metadata(sheet)
        date = sheet["I1"].value
        base_date = f"{date.year}{date.month:02}{date.day:02}"
        readings = []
        # define the number of sets in a spreadsheet:
        sets_len = len(
            [
                sheet.cell(row=i, column=3).value
                for i in range(10, 14)
                if sheet.cell(row=i, column=3).value is not None
            ]
        )

        for d_n in range(10, 10 + sets_len):
            h_n = d_n + 14
            v_n = d_n + 28
            absolutes = [
                Absolute(
                    element="D",
                    absolute=Angle.from_dms(
                        degrees=sheet[f"C{d_n}"].value, minutes=sheet[f"D{d_n}"].value
                    ),
                    baseline=convert_baseline(sheet[f"H{d_n}"].value),
                    starttime=parse_relative_time(
                        base_date, time=(sheet[f"B{v_n}"].value)
                    ),
                    endtime=parse_relative_time(
                        base_date, time=(sheet[f"B{d_n}"].value)
                    ),
                    valid=not bool(sheet[f"J{d_n}"].value),
                ),
                Absolute(
                    element="H",
                    absolute=sheet[f"D{h_n}"].value,
                    baseline=sheet[f"H{h_n}"].value,
                    starttime=parse_relative_time(
                        base_date, time=(sheet[f"B{v_n}"].value)
                    ),
                    endtime=parse_relative_time(
                        base_date, time=(sheet[f"B{h_n}"].value)
                    ),
                    valid=not bool(sheet[f"J{h_n}"].value),
                ),
                Absolute(
                    element="Z",
                    absolute=sheet[f"D{v_n}"].value,
                    baseline=sheet[f"H{v_n}"].value,
                    starttime=parse_relative_time(
                        base_date, time=(sheet[f"B{v_n}"].value)
                    ),
                    endtime=parse_relative_time(
                        base_date, time=(sheet[f"B{v_n}"].value)
                    ),
                    valid=not bool(sheet[f"J{v_n}"].value),
                ),
            ]
            readings.append(
                Reading(
                    metadata=metadata,
                    absolutes=absolutes,
                    pier_correction=metadata["pier_correction"],
                ),
            )
        return readings