Boltwood and safety file writer for weather conditions for astronomy

boltwood_safety_writer.py

import re        
from enum import Enum
from datetime import datetime

class WindCondition(Enum):
    WIND_UNKNOWN = 0
    WIND_CALM = 1
    WIND_WINDY = 2
    WIND_VERY_WINDY = 3
    
class RainCondition(Enum):
    RAIN_UNKNKOWN = 0
    RAIN_DRY = 1
    RAIN_WET = 2
    RAIN_RAIN = 3
    
class DayCondition(Enum):
    DAY_UNKNOWN = 0
    DAY_DARK = 1
    DAY_LIGHT = 2
    DAY_VERY_LIGHT = 3
    
class CloudCondition(Enum):
    CLOUD_UNKNOWN = 0
    CLOUD_CLEAR = 1
    CLOUD_CLOUDY = 2
    CLOUD_VERY_CLOUDY = 3
    
def get_condition(value, mapping, enum, reversed=False):
    if value is None or np.isnan(value):
        return enum(0)
    mapping_array = np.array(mapping)
    assert len(mapping) == len(enum) - 2
    if reversed:
        index = (value <= mapping_array).sum() + 1
    else:
        index = (value >= mapping_array).sum() + 1
    condition = enum(index)
    return condition
    


class BoltwoodFile:
    def __init__(self, filename):
        self.filename = filename
        self.data = {}
        self.wind_condition_limits = [5, 10]
        self.rain_condition_limits = [1e-6, 0.01]
        self.day_condition_limits = [12, 18]
        self.cloud_condition_limits = [15, 60]
        
    def update_condition_limits(self, name, limits):
        assert(len(limits) == 2)
        if name == 'wind':
            self.wind_condition_limits = limits
        if name == 'rain':
            self.rain_condition_limits = limits
        if name == 'day':
            self.day_condition_limits = limits
        if name == 'cloud':
            self.cloud_condition_limits = limits

        
    def parse_data(self, temperature, dewpoint, humidity, 
                   wind_speed=0, 
                   sky_temp=0, 
                   sqm_value=None, 
                   rain_rate=None, 
                   cloud_cover=None,
                  seconds_stale=0):
        self.data['date_string'] = datetime.now().strftime('%Y-%m-%d')
        self.data['time_string'] = datetime.now().strftime('%H:%M:%S.%m')
        self.data['temp_scale'] = 'F'
        self.data['wind_scale'] = 'M'
        self.data['sky_temp'] = sky_temp
        self.data['amb_temp'] = temperature
        self.data['sens_temp'] = temperature
        self.data['wind_speed'] = wind_speed
        self.data['humidity'] = int(humidity)
        self.data['dewpoint'] = dewpoint
        self.data['dew_heater_pct'] = '000'
        self.data['rain_flag'] = int(rain_rate >= 0.001)
        self.data['wet_flag'] = int(rain_rate > 0.0)
        self.data['elapsed_time'] = '{:05d}'.format(seconds_stale)
        elapsed_days = (datetime.now() - datetime(1900, 1, 1)).days + 2
        elapsed_frac_days = (datetime.now() - datetime(1900, 1, 1)).seconds / 3600 / 24
        
        self.data['elapsed_days'] = '{:06d}.{:3d}'.format(elapsed_days, int(elapsed_frac_days * 1e5))
        
        self.data['cloud_condition_flag'] = get_condition(cloud_cover, self.cloud_condition_limits, CloudCondition).value
        self.data['wind_condition_flag'] = get_condition(wind_speed, self.wind_condition_limits, WindCondition).value
        self.data['rain_condition_flag'] = get_condition(rain_rate, self.rain_condition_limits, RainCondition).value
        self.data['dark_condition_flag'] = get_condition(sqm_value, self.day_condition_limits, DayCondition, reversed=True).value
        self.data['close_flag'] = 0
        self.data['alert_flag'] = 0
        
    def write_data(self):
        self.line_format = '''{date_string:10} {time_string:11} {temp_scale} {wind_scale} {sky_temp:6.1f} \
{amb_temp:6.1f} {sens_temp:6.1f} {wind_speed:6.1f} {humidity:3d} {dewpoint:6.1f} \
{dew_heater_pct:3} {rain_flag} {wet_flag} {elapsed_time} {elapsed_days} \
{cloud_condition_flag} {wind_condition_flag} {rain_condition_flag} {dark_condition_flag} {close_flag} {alert_flag}'''
        with open(self.filename, 'w') as f:
            f.write(self.line_format.format(**self.data))
        

class SafetyMonitor:
    def __init__(self, filename):
        self.filename = filename
        self.is_safe = False
        self.safe_limit = {}
        self.safe_limit['wind'] = WindCondition.WIND_VERY_WINDY.value
        self.safe_limit['cloud'] = CloudCondition.CLOUD_VERY_CLOUDY.value
        self.safe_limit['rain'] = RainCondition.RAIN_WET.value
        self.safe_limit['dark'] = DayCondition.DAY_VERY_LIGHT.value
        
    def update_safe_limit(self, name, limit):
        self.safe_limit[name] = limit
        
    def update_from_boltwood_file(self, boltwood_file):
        
        with open(boltwood_file, 'r') as f:
            bf_line = f.readline()
        names = [re.sub('{|}', '', re.sub(':.*', '', s)) for s in bf.line_format.split()]
        data = dict(zip(names, bf_line.split()))
        self.is_safe = True
        for flag in ['wind', 'cloud', 'rain', 'dark']:
            if flag in self.safe_limit and f'{flag}_condition_flag' in data:
                if int(data.get(f'{flag}_condition_flag', 0)) >= self.safe_limit.get(flag, 0):
                    self.is_safe = False
        output = 'not_safe'
        if self.is_safe:
            output = 'is_safe'
        with open(self.filename, 'w') as f:
            f.write(output)