Skip to content

Weather#

Weather file fetching and caching.

BaseWeather #

Bases: BaseModel

A base class for fetching weather files.

Responsible for fetching weather files from a URL and extracting the .epw and .ddy files.

Also takes care of caching the weather files in a directory.

Source code in epinterface\weather.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class BaseWeather(BaseModel):
    """A base class for fetching weather files.

    Responsible for fetching weather files from a URL and extracting the .epw and .ddy files.

    Also takes care of caching the weather files in a directory.
    """

    Weather: WeatherUrl = Field(
        default=WeatherUrl(  # pyright: ignore [reportCallIssue]
            "https://climate.onebuilding.org/WMO_Region_4_North_and_Central_America/USA_United_States_of_America/MA_Massachusetts/USA_MA_Boston-Logan.Intl.AP.725090_TMYx.2009-2023.zip"
        )
    )

    async def fetch_weather(self, cache_dir: Path | str):
        """Fetch the weather file from the URL and extract the .epw and .ddy files.

        Args:
            cache_dir (Path | str): The directory to cache the weather files.

        Returns:
            epw_path (Path): The path to the .epw file.
            ddy_path (Path): The path to the .ddy file.
        """
        if isinstance(cache_dir, str):
            cache_dir = Path(cache_dir)

        if not self.Weather.path or not self.Weather.path.endswith(".zip"):
            raise NotAZipError()

        weather_path = Path(self.Weather.path).relative_to("/")
        weather_dir = cache_dir / weather_path.with_suffix("")
        epw_path = weather_dir / weather_path.with_suffix(".epw").name
        ddy_path = weather_dir / weather_path.with_suffix(".ddy").name
        weather_dir.mkdir(parents=True, exist_ok=True)
        if not epw_path.exists() or not ddy_path.exists():
            logger.info(f"Fetching weather file from {self.Weather}")
            # fetch the .zip file, unzip it, and extract the .epw and .ddy files
            client = httpx.AsyncClient()
            response = await client.get(str(self.Weather))
            with tempfile.TemporaryFile() as f:
                f.write(response.content)
                f.seek(0)
                with zipfile.ZipFile(f, "r") as z:
                    # z.extractall(weather_dir)
                    z.extract(epw_path.name, weather_dir)
                    z.extract(ddy_path.name, weather_dir)
            await client.aclose()
        else:
            logger.info(f"Using cached weather file from {weather_dir}")
        return epw_path, ddy_path

fetch_weather(cache_dir) async #

Fetch the weather file from the URL and extract the .epw and .ddy files.

Parameters:

Name Type Description Default
cache_dir Path | str

The directory to cache the weather files.

 required

Returns:

Name Type Description
epw_path Path

The path to the .epw file.
ddy_path Path

The path to the .ddy file.
Source code in epinterface\weather.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def fetch_weather(self, cache_dir: Path | str):
    """Fetch the weather file from the URL and extract the .epw and .ddy files.

    Args:
        cache_dir (Path | str): The directory to cache the weather files.

    Returns:
        epw_path (Path): The path to the .epw file.
        ddy_path (Path): The path to the .ddy file.
    """
    if isinstance(cache_dir, str):
        cache_dir = Path(cache_dir)

    if not self.Weather.path or not self.Weather.path.endswith(".zip"):
        raise NotAZipError()

    weather_path = Path(self.Weather.path).relative_to("/")
    weather_dir = cache_dir / weather_path.with_suffix("")
    epw_path = weather_dir / weather_path.with_suffix(".epw").name
    ddy_path = weather_dir / weather_path.with_suffix(".ddy").name
    weather_dir.mkdir(parents=True, exist_ok=True)
    if not epw_path.exists() or not ddy_path.exists():
        logger.info(f"Fetching weather file from {self.Weather}")
        # fetch the .zip file, unzip it, and extract the .epw and .ddy files
        client = httpx.AsyncClient()
        response = await client.get(str(self.Weather))
        with tempfile.TemporaryFile() as f:
            f.write(response.content)
            f.seek(0)
            with zipfile.ZipFile(f, "r") as z:
                # z.extractall(weather_dir)
                z.extract(epw_path.name, weather_dir)
                z.extract(ddy_path.name, weather_dir)
        await client.aclose()
    else:
        logger.info(f"Using cached weather file from {weather_dir}")
    return epw_path, ddy_path

NotAZipError #

Bases: ValueError

Raised when a URL does not end with a .zip extension.

Source code in epinterface\weather.py 
15
16
17
18
19
20
class NotAZipError(ValueError):
    """Raised when a URL does not end with a .zip extension."""

    def __init__(self) -> None:
        """Initialize the error."""
        super().__init__("The URL provided does not end with a .zip extension.")

__init__() #

Initialize the error.

Source code in epinterface\weather.py 
18
19
20
def __init__(self) -> None:
    """Initialize the error."""
    super().__init__("The URL provided does not end with a .zip extension.")

check_path_ends_with_zip(url) #

Check that the path of the URL ends with a .zip extension.

Parameters:

Name Type Description Default
url AnyUrl

The URL to check.

 required

Raises:

Type Description
NotAZipError

If the URL does not end with a .zip extension.

Returns:

Name Type Description
url AnyUrl

The URL.
Source code in epinterface\weather.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def check_path_ends_with_zip(url: AnyUrl):
    """Check that the path of the URL ends with a .zip extension.

    Args:
        url (AnyUrl): The URL to check.

    Raises:
        NotAZipError: If the URL does not end with a .zip extension.

    Returns:
        url (AnyUrl): The URL.
    """
    if not url.path:
        raise NotAZipError()
    if not url.path.endswith(".zip"):
        raise NotAZipError()
    return url