isd.conf

# Beats -> Logstash -> Elasticsearch pipeline.s



input {
  beats {
    port => 5044
  }
}

# fixed width match for required fields 
filter {
	grok { match => [ "message", "^(?<total-variable-characters>.{4})(?<fixed-weather-station-identifier>.{6})(?<fixed-weather-station-ncei-wban-id>.{5})(?<geophysical-point-observation-date>.{8})(?<geophysical-point-observation-time>.{4})(?<geophysical-point-observation-data-source-flag>.{1})(?<geophysical-point-observation-latitude-coordinate>.{6})(?<geophysical-point-observation-longitude-coordinate>.{7})(?<geophysical-report-type-code>.{5})(?<geophysical-point-observation-elevation-dimension>.{5})(?<fixed-weather-station-call-letter-identifier>.{5})(?<meteorological-point-observation-quality-control-process-name>.{4})(?<wind-observation-direction-angle>.{3})(?<wind-observation-direction-quality-code>.{1})(?<wind-observation-type-code>.{1})(?<wind-observation-speed-rate>.{4})(?<wind-observation-speed-quality-code>.{1})(?<sky-condition-observation-ceiling-height-dimension>.{5})(?<sky-condtion-observation-ceiling-quality-code>.{1})(?<sky-condition-observation-ceiling-determination-code>.{1})(?<sky-condition-observation-cavok-code>.{1})(?<visibility-observation-distance-dimension>.{6})(?<visibility-observation-distance-quality-code>.{1})(?<visibility-observation-variability-code>.{1})(?<visibility-observation-quality-variability-code>.{1})(?<air-temperature-observation-air-temperature>.{5})(?<air-temperature-observation-air-temperature-quality-code>.{1})(?<air-temperature-observation-dew-point-temperature>.{5})(?<air-temperature-observation-dew-point-quality-code>.{1})(?<atmospheric-pressure-observation-sea-level-pressure>.{5})(?<atmospheric-pressure-observation-sea-level-pressure-quality-code>.{1})"] }
}

# 
filter {
	mutate { add_field => { "[@metadata][ts]" => "%{geophysical-point-observation-date}%{geophysical-point-observation-time}" }}
	}

filter 
	{
	date { match => [ "[@metadata][ts]", "YYYYMMddHHmm" ] }
	}

# type conversions
filter {
      mutate {
        convert => {
          "geophysical-point-observation-longitude-coordinate" => "float"
          "geophysical-point-observation-latitude-coordinate" => "float"
          "geophysical-point-observation-elevation-dimension" => "integer"
          "wind-observation-direction-angle" => "integer"
          "wind-observation-speed-rate" => "float"
          "sky-condition-observation-ceiling-height-dimension" => "integer"
          "visibility-observation-distance-dimension" => "integer"
          "air-temperature-observation-air-temperature" => "integer"
          "air-temperature-observation-dew-point-temperature" => "integer"
          "atmospheric-pressure-observation-sea-level-pressure" => "integer"
        }
      }
    }

# POS 28 geophysical-point-observation-data-source-flag
filter {
      translate {
        field => "[geophysical-point-observation-data-source-flag]"
        destination => "[m-geophysical-point-observation-data-source-translated]"
        dictionary => {
"1"=>"USAF SURFACE HOURLY"
"2"=>"NCEI SURFACE HOURLY"
"3"=>"USAF SURFACE HOURLY/NCEI SURFACE HOURLY"
"5"=>"NCEI SURFACE HOURLY"
"6"=>"ASOS/AWOS"
"7"=>"ASOS/AWOS"
"8"=>"MAPSO"
"A"=>"USAF SURFACE HOURLY/NCEI HOURLY PRECIPITATION"
"B"=>"NCEI SURFACE HOURLY/NCEI HOURLY PRECIPITATION"
"C"=>"USAF SURFACE HOURLY/NCEI SURFACE HOURLY/NCEI HOURLY PRECIPITATION"
"D"=>"USAF SURFACE HOURLY/NCEI HOURLY PRECIPITATION"
"G"=>"SAO surface airways observation=> pre-1949 (keyed data)"
"H"=>"SAO surface airways observation=> 1965-1981 format/period (keyed data)"
"J"=>"Cooperative Network observation"
"K"=>"Radiation Network observation"
"L"=>"Data from Climate Data Modernization Program (CDMP) data source"
"M"=>"Data from National Renewable Energy Laboratory (NREL) data source"
"N"=>"NCAR / NCEI cooperative effort (various national datasets)"
"O"=>"Summary observation created by NCEI"
"E"=>"NCEI SURFACE HOURLY/NCEI HOURLY PRECIPITATION"
"4"=>"USAF SURFACE HOURLY observation"
"I"=>"Climate Reference Network observation"
"9"=>"Missing"
        }
        fallback => "no match"
        } 
      }

#POS 46 geophysical-report-type-code]
filter {
      translate {
        field => "[geophysical-report-type-code]"
        destination => "[m-geophysical-report-type-code-translated]"
        dictionary => {
"AERO"=>"Aerological report"
"AUST"=>"Dataset from Australia"
"AUTO"=>"Report from an automatic station"
"BOGUS"=>"Bogus report"
"BRAZ"=>"Dataset from Brazil"
"COOPD"=>"US Cooperative Network summary of day report"
"COOPS"=>"US Cooperative Network soil temperature report"
"CRB"=>"Climate Reference Book data from CDMP"
"CRN05"=>"Climate Reference Network report, with 5-minute reporting interval"
"CRN15"=>"Climate Reference Network report, with 15-minute reporting interval"
"FM-12"=>"SYNOP Report of surface observation form a fixed land station"
"FM-13"=>"SHIP Report of surface observation from a sea station"
"FM-14"=>"SYNOP MOBIL Report of surface observation from a mobile land station"
"FM-15"=>"METAR Aviation routine weather report"
"FM-16"=>"SPECI Aviation selected special weather report"
"FM-18"=>"BUOY Report of a buoy observation"
"GREEN"=>"Dataset from Greenland"
"MESOH"=>"Hydrological observations from MESONET operated civilian or government agency" 
"MESOS"=>"MESONET operated civilian or government agency"
"MESOW"=>"Snow observations from MESONET operated civilian or government agency"
"MEXIC"=>"Dataset from Mexico"
"NSRDB"=>"National Solar Radiation Data Base"
"PCP15"=>"US 15-minute precipitation network report"
"PCP60"=>"US 60-minute precipitation network report"
"S-S-A"=>"Synoptic, airways, and auto merged report"
"SA-AU"=>"Airways and auto merged report"
"SAO"=>"Airways report (includes record specials)"
"SAOSP"=>"Airways special report (excluding record specials)"
"SHEF"=>"Standard Hydrologic Exchange Format"
"SMARS"=>"Supplementary airways station report"
"SOD"=>"Summary of day report from U.S. ASOS or AWOS station" 
"SOM"=>"Summary of month report from U.S. ASOS or AWOS station" 
"SURF"=>"Surface Radiation Network report"
"SY-AE"=>"Synoptic and aero merged report"
"SY-AU"=>"Synoptic and auto merged report"
"SY-MT"=>"Synoptic and METAR merged report"
"SY-SA"=>"Synoptic and airways merged report"
"WBO"=>"Weather Bureau Office"
"WNO"=>"Washington Naval Observatory"
"99999"=>"Missing"

              }
        fallback => "no match"
        } 
      }  

filter {
      translate {
        field => "[meteorological-point-observation-quality-control-process-name]"
        destination => "[m-meteorological-point-observation-quality-control-process-name-translated]"
        dictionary => {
"V01"=>"No A or M Quality Control applied" 
"V02"=>"Automated Quality Control"
"V03"=>"subjected to Quality Control"
           }
        fallback => "no match"
        } 
      }  

# POS 64 wind-observation-type-code      
filter {
      translate {
        field => "[wind-observation-direction-quality-code]"
        destination => "[m-wind-observation-direction-quality-code-translated]"
        dictionary => {
"0"=>"Passed gross limits check"
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source"
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
           }
        fallback => "no match"
        } 
      } 

# POS 65 wind-observation-type-code
        filter {
      translate {
        field => "[wind-observation-type-code]"
        destination => "[m-wind-observation-type-code-translated]"
        dictionary => {
"A"=>"Abridged Beaufort"
"B"=>"Beaufort"
"C"=>"Calm"
"H"=>"5-Minute Average Speed N - Normal"
"R"=>"60-Minute Average Speed Q - Squall"
"T"=>"180 Minute Average Speed V - Variable"
"9"=>"Missing"
           }
        fallback => "no match"
        } 
      }

# fixup latitude and longitude for the right scale factor 1000
#POS 34, 41
filter 
	{
	math 	{
	calculate =>[
		["fdiv", "[geophysical-point-observation-longitude-coordinate]", 1000, "[geophysical-point-observation-longitude-coordinate]" ]
	 ,["fdiv", "[geophysical-point-observation-latitude-coordinate]", 1000, "[geophysical-point-observation-latitude-coordinate]"]
		]
			}
	}
#convert wind speed to scale factor of 10 and to Miles PH from Meters per second MPS * 0.223694 and round to 2 decimel places
  filter 
  {
  if "[wind-observation-speed-rate]" != 999 {
  math  {
  calculate =>[
    ["multiply", "[wind-observation-speed-rate]", 0.223694, "[m-wind-observation-speed-rate-mph]" ],
    ["round", "[m-wind-observation-speed-rate-mph]", 2, "[m-wind-observation-speed-rate-mph]"]
    ]
      }
      }
  }
# create geopoint

filter{
if "[geophysical-point-observation-longitude-coordinate]" != "+999999" {
mutate {
  add_field => [ "[location][lat]", "%{geophysical-point-observation-latitude-coordinate}" ]
  add_field => [ "[location][lon]", "%{geophysical-point-observation-longitude-coordinate}" ]
  }
  }
}

filter {
      mutate {
        convert => {
          "[location][lat]" => "float"
          "[location][lon]" => "float"
        }
      }

}

# POS 76 sky-condtion-observation-ceiling-quality-code
filter {
      translate {
        field => "[sky-condtion-observation-ceiling-quality-code]"
        destination => "[m-sky-condtion-observation-ceiling-quality-code-translated]"
        dictionary => {
"0"=>"Passed gross limits check"
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source "
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
           }
        fallback => "no match"
        } 
      }
# POS 77 sky-condition-observation-ceiling-determination-code
filter {
      translate {
        field => "[sky-condition-observation-ceiling-determination-code]"
        destination => "[m-sky-condition-observation-ceiling-determination-code-translated]"
        dictionary => {
"A"=>"Aircraft"
"B"=>"Balloon"
"C"=>"Statistically derived"
"D"=>"Persistent cirriform ceiling (pre-1950 data)"
"E"=>"Estimated"
"M"=>"Measured"
"P"=>"Precipitation ceiling (pre-1950 data)"
"R"=>"Radar"
"S"=>"ASOS augmented"
"U"=>"Unknown ceiling (pre-1950 data)"
"V"=>"Variable ceiling (pre-1950 data)"
"W"=>"Obscured"
"9"=>"Missing"
           }
        fallback => "no match"
        } 
      }

# POS 78 sky-condition-observation-cavok-code
filter {
      translate {
        field => "[sky-condition-observation-cavok-code]"
        destination => "[m-sky-condition-observation-cavok-code-translated]"
        dictionary => {
"N"=>"No"
"Y"=>"Yes"
"9"=>"Missing"
          }
        fallback => "no match"
        } 
      }     


# POS 84 visibility-observation-distance-dimension convert from meters to miles adn round to two decimel places
filter 
{
  if "[visibility-observation-quality-variability-code]" != "9"
    {
    math  
        {
        calculate =>[
        ["multiply", "[visibility-observation-distance-dimension]", 0.000621371, "[m-visibility-observation-distance-dimension-miles]" ],
        ["round", "[m-visibility-observation-distance-dimension-miles]", 2, "[m-visibility-observation-distance-dimension-miles]"]
        ]
    
        }
        
    }
}

# POS  85 visibility-observation-distance-quality-code
filter {
      translate {
        field => "[visibility-observation-distance-quality-code]"
        destination => "[m-visibility-observation-distance-quality-code-translated]"
        dictionary => {
"0"=>"Passed gross limits check"
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source "
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
           }
        fallback => "no match"
        } 
      } 

# POS 92 air-temperature-observation-air-temperature convert to F and scale factor of 10 and round to 1 place
filter 
{
  if "[air-temperature-observation-air-temperature]" != 9999
    {
    math  
        {
        calculate =>[
        ["multiply", "[air-temperature-observation-air-temperature]", 0.18, "[m-air-temperature-observation-air-temperature-farenheit]" ],
        ["add", "[m-air-temperature-observation-air-temperature-farenheit]", 32, "[m-air-temperature-observation-air-temperature-farenheit]" ],
        ["round", "[m-air-temperature-observation-air-temperature-farenheit]", 1, "[m-air-temperature-observation-air-temperature-farenheit]"]
        ]
        }
    }
}

# POS  93 air-temperature-observation-air-temperature-quality-code
filter {
      translate {
        field => "[air-temperature-observation-air-temperature-quality-code]"
        destination => "[m-air-temperature-observation-air-temperature-quality-code-translated]"
        dictionary => {
"0"=>"Passed gross limits check"
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source"
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
"A"=>"Data value flagged as suspect, but accepted as a good value"
"C"=>"Temperature and dew point received from Automated Weather Observing System (AWOS) are reported in whole degrees Celsius. Automated QC flags these values, but they are accepted as valid."
"I"=>"Data value not originally in data, but inserted by validator"
"M"=>"Manual changes made to value based on information provided by NWS or FAA"
"P"=>"Data value not originally flagged as suspect, but replaced by validator"
"R"=>"Data value replaced with value computed by NCEI software"
"U"=>"Data value replaced with edited value"
           }
        fallback => "no match"
        } 
      } 

# POS 98 air-temperature-observation-dew-point-temperature convert to F and scale factor of 10 and round to 1 place
filter 
{
  if "[air-temperature-observation-dew-point-temperature]" != 9999
    {
    math  
        {
        calculate =>[
        ["multiply", "[air-temperature-observation-dew-point-temperature]", 0.18, "[m-air-temperature-observation-dew-point-temperature-farenheit]" ],
        ["add", "[m-air-temperature-observation-dew-point-temperature-farenheit]", 32, "[m-air-temperature-observation-dew-point-temperature-farenheit]" ],
        ["round", "[m-air-temperature-observation-dew-point-temperature-farenheit]", 1, "[m-air-temperature-observation-dew-point-temperature-farenheit]"]
        ]
        }
    }
}

# POS  99 air-temperature-observation-dew-point-quality-code
filter {
      translate {
        field => "[air-temperature-observation-dew-point-quality-code]"
        destination => "[m-air-temperature-observation-dew-point-quality-code-translated]"
        dictionary => {
"0"=>"Passed gross limits check"
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source"
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
"A"=>"Data value flagged as suspect, but accepted as a good value"
"C"=>"Temperature and dew point received from Automated Weather Observing System (AWOS) are reported in whole degrees Celsius. Automated QC flags these values, but they are accepted as valid."
"I"=>"Data value not originally in data, but inserted by validator"
"M"=>"Manual changes made to value based on information provided by NWS or FAA"
"P"=>"Data value not originally flagged as suspect, but replaced by validator"
"R"=>"Data value replaced with value computed by NCEI software"
"U"=>"Data value replaced with edited value"
           }
        fallback => "no match"
        } 
        }

# POS 104 atmospheric-pressure-observation-sea-level-pressure scale factor of 10 
filter 
{
  if "[atmospheric-pressure-observation-sea-level-pressure]" != 99999
    {
    math  
        {
        calculate =>[
        ["multiply", "[air-temperature-observation-dew-point-temperature]", 0.1, "[m-air-temperature-observation-dew-point-temperature-farenheit]" ]
        ]
        }
    }
}

# POS  105 atmospheric-pressure-observation-sea-level-pressure-quality-code
filter {
      translate {
        field => "[atmospheric-pressure-observation-sea-level-pressure-quality-code]"
        destination => "[m-atmospheric-pressure-observation-sea-level-pressure-quality-code-translated]"
        dictionary => {
"1"=>"Passed all quality control checks"
"2"=>"Suspect"
"3"=>"Erroneous"
"4"=>"Passed gross limits check, data originate from an NCEI data source"
"5"=>"Passed all quality control checks, data originate from an NCEI data source "
"6"=>"Suspect, data originate from an NCEI data source"
"7"=>"Erroneous, data originate from an NCEI data source"
"9"=>"Passed gross limits check if element is present"
           }
        fallback => "no match"
        } 
      } 
output {
#output to elasticsearch or elastic cloud 
 elasticsearch {
    #Add your Elasticsearch hosts
    hosts => ["your elasticsearch endpoint"]
    index => "weather"
    #Add Elasticsearch credentials
    user => "elastic"
    password => "yourpassword"
  }
}

#output to stdout for debugging

#output { 
 # stdout { codec => rubydebug }
#}