Commit a93ace6a authored by JOSSOUD Olivier's avatar JOSSOUD Olivier
Browse files

FTPSIRTA. MeteoFrance data.

parent 951a1524
Pipeline #57848 passed with stages
in 3 minutes and 12 seconds
import configobj
import datetime
import wimcollect.common.logger as logger
import wimcollect.common.ftp as ftp
import wimcollect.common.utils as utils
class SirtaMfCollector(utils.LogConfig):
"""Collector for Sirta's Meteo-France meteo data file"""
def __init__(self, config_parser: configobj.ConfigObj = None, log: logger = None):
self.object_id = "FTPSIRTA"
utils.LogConfig.__init__(self, self.object_id, config_parser, log)
self.distant_base_dir = self.config[self.object_id]["distant_base_dir"]
def download(self, day:
"""Download Sirta's Meteo-France meteo data file.
Date of the data which should be downloaded.
date_str = day.strftime("%Y%m%d")
self.logger.date_str = date_str
# Build source and destination file paths
source_filepath = self.distant_base_dir + "/" + day.strftime("%Y/%m/%d/") + \
"meteoairsol_1a_Lz1NairF1minPtuvPrain_v01_" + date_str + "_000000_1440.asc"
dest_filepath = utils.get_standard_filepath(self.local_base_dir, "SRT", "meteo", "sirtamf", date_str, "tsv")
# Download
success = ftp.download_file(source_filepath, dest_filepath, self.logger,
# Compress
if success:
utils.compress_file(dest_filepath, self.logger)
return True
import datetime
from unittest import TestCase
import wimcollect.ftpsirta as ftpsirta
class TestSirtaMfCollector(TestCase):
def test_download(self):
collector = ftpsirta.SirtaMfCollector()
self.assertTrue(, 1, 1)))
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment