pingu_98/cara
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'feature/climate_function' of https://gitlab.cern.ch/cara/cara into feature/climate_function

Browse source
This commit is contained in:
Luis Aleixo 2021-08-26 14:09:16 +02:00
commit 10ef9b4149
5 changed files with 81 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cara/apps/calculator/model_generator.py
View file

@ -313,7 +313,7 @@ class FormData:
else:
return models.MultipleVentilation((ventilation, infiltration_ventilation))
def nearest_weather_station(self) -> cara.data.weather.WxStationType:
def nearest_weather_station(self) -> cara.data.weather.WxStationRecordType:
wx_station = cara.data.weather.nearest_wx_station(
longitude=self.location_longitude, latitude=self.location_latitude
)

1
cara/data/.gitattributes vendored
View file

@ -1 +0,0 @@
cara_weather_stations.txt filter=lfs diff=lfs merge=lfs -text

3
cara/data/hadisd_station_fullinfo_v311_202001p.txt Normal file
View file

@ -0,0 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:4843d34b6e4c26d4382860e011451d5f32157b9a3660830f8d2894a11d022298
size 772370

83
cara/data/weather.py
View file

@ -1,53 +1,29 @@
import functools
import json
from pathlib import Path
import typing
import numpy as np
from cara import models
import json
import urllib.request
from pathlib import Path
from scipy.spatial import cKDTree
import os
import typing
from cara import models
weather_debug = False
DATA_LOCATION = Path(__file__).absolute().parent
def location_to_weather_stn(location_loc):
# expects a tuple (lat, long)
# returns: weather station ID, weather station name, weather station lat, long
search_coords = location_loc.split(',')
lat = []
long = []
station_array = []
fixed_delimits = [0, 12, 13, 44, 51, 60, 69, 90, 91]
station_file = DATA_LOCATION / 'hadisd_station_fullinfo_v311_202001p.txt'
for line in station_file.open('rt'):
start_end_positions = zip(fixed_delimits[:-1], fixed_delimits[1:])
split_vals = [line[start:end] for start, end in start_end_positions]
station_location = [split_vals[0],
split_vals[2], split_vals[3], split_vals[4]]
station_array.append(station_location)
lat.append(split_vals[3])
long.append(split_vals[4])
tree = cKDTree(np.c_[lat, long])
dd, ii = tree.query(search_coords, k=[1])
return (station_array[ii[0]][0], station_array[ii[0]][1], station_array[ii[0]][2], station_array[ii[0]][3])
WxStationType = MonthType = str
WxStationIdType = str
MonthType = str
# HourlyTempType - 24 temperatures, one for each hour of the day (the average for the given month).
HourlyTempType = typing.List[float]
@functools.lru_cache()
def wx_data() -> typing.Dict[WxStationType, typing.Dict[MonthType, HourlyTempType]]:
def wx_data() -> typing.Dict[WxStationIdType, typing.Dict[MonthType, HourlyTempType]]:
"""
Load the weather data.
Load the weather data (temperature in kelvin).
The data is structured by station location, and for each station location, by month.
@ -61,11 +37,18 @@ def wx_data() -> typing.Dict[WxStationType, typing.Dict[MonthType, HourlyTempTyp
return data
StationRecordType = typing.Tuple[WxStationType, str, float, float]
WxStationRecordType = typing.Tuple[WxStationIdType, str, float, float]
@functools.lru_cache()
def wx_station_data() -> typing.Dict[WxStationType, StationRecordType]:
def wx_station_data() -> typing.Dict[WxStationIdType, WxStationRecordType]:
"""
Return a dictionary of ``station-id: station records``, where station records
are of the form ``(station-id, station-name, station-latitude, station-longitude)``.
The stations returned are guaranteed to have valid weather data.
"""
weather_data = wx_data()
station_data = {}
fixed_delimits = [0, 12, 13, 44, 51, 60, 69, 90, 91]
@ -74,8 +57,9 @@ def wx_station_data() -> typing.Dict[WxStationType, StationRecordType]:
for line in station_file.open('rt'):
start_end_positions = zip(fixed_delimits[:-1], fixed_delimits[1:])
split_vals = [line[start:end] for start, end in start_end_positions]
station_location = (split_vals[0],
split_vals[2], split_vals[3], split_vals[4])
station_location = (
split_vals[0], split_vals[2], float(split_vals[3]), float(split_vals[4]),
)
# We only consider stations with weather data, don't include the rest.
if split_vals[0] in weather_data:
station_data[split_vals[0]] = station_location
@ -84,6 +68,7 @@ def wx_station_data() -> typing.Dict[WxStationType, StationRecordType]:
@functools.lru_cache()
def _wx_station_kdtree() -> cKDTree:
"""Build a kd-tree of wx station longitude & latitudes (note the coordinate order)"""
station_data = wx_station_data().values()
coords = np.array([(stn_record[3], stn_record[2]) for stn_record in station_data])
return cKDTree(coords)
@ -118,24 +103,12 @@ def hourly_to_piecewise(hourly_data: HourlyTempType) -> models.PiecewiseConstant
return pc
def nearest_wx_station(*, longitude: float, latitude: float) -> StationRecordType:
def nearest_wx_station(*, longitude: float, latitude: float) -> WxStationRecordType:
"""
Given a latitude & longitude, return the nearest station with valid weather data.
"""
ktree = _wx_station_kdtree()
station_data = list(wx_station_data().values())
dd, ii = ktree.query((longitude, latitude), k=[1])
return station_data[ii[0]]
def location_celcius_per_hour(location: object) -> typing.Dict[str, typing.List[float]]:
# expects a tuple (lat, long)
# returns a json format set of weather data
w_station = location_to_weather_stn(location)
with (DATA_LOCATION / 'global_weather_set.json').open("r") as json_file:
weather_dict = json.load(json_file)
Location_hourly_temperatures_celsius_per_hour = weather_dict[w_station[0]]
if weather_debug:
print(location)
print("weather station name: ", w_station[1])
print("weather station ref: ", w_station[0])
print("weather station location: ", w_station[2], " ", w_station[3])
print(Location_hourly_temperatures_celsius_per_hour)
return Location_hourly_temperatures_celsius_per_hour

75
cara/tests/apps/calculator/test_model_generator.py
View file

@ -1,14 +1,14 @@
import dataclasses
import typing
import numpy as np
import numpy.testing as npt
import pytest
from cara.apps.calculator import model_generator
from cara.apps.calculator.model_generator import _hours2timestring
from cara.apps.calculator.model_generator import minutes_since_midnight
from cara import models
from cara import data
import numpy as np
import numpy.testing as npt
def test_model_from_dict(baseline_form_data):
@ -33,13 +33,6 @@ def test_blend_expiration():
def test_ventilation_slidingwindow(baseline_form: model_generator.FormData):
room = models.Room(75)
window = models.SlidingWindow(
active=models.PeriodicInterval(period=120, duration=10),
inside_temp=models.PiecewiseConstant((0, 24), (293,)),
outside_temp=data.GenevaTemperatures['Dec'],
window_height=1.6, opening_length=0.6,
)
baseline_form.ventilation_type = 'natural_ventilation'
baseline_form.windows_duration = 10
baseline_form.windows_frequency = 120
@ -49,19 +42,28 @@ def test_ventilation_slidingwindow(baseline_form: model_generator.FormData):
baseline_form.window_height = 1.6
baseline_form.opening_distance = 0.6
ts = np.linspace(8, 16, 100)
np.testing.assert_allclose([window.air_exchange(room, t)+0.25 for t in ts],
[baseline_form.ventilation().air_exchange(room, t) for t in ts])
baseline_vent = baseline_form.ventilation()
assert isinstance(baseline_vent, models.MultipleVentilation)
baseline_window = baseline_vent.ventilations[0]
assert isinstance(baseline_window, models.SlidingWindow)
window = models.SlidingWindow(
active=models.PeriodicInterval(period=120, duration=10),
inside_temp=models.PiecewiseConstant((0, 24), (293,)),
outside_temp=baseline_window.outside_temp,
window_height=1.6, opening_length=0.6,
)
ach = models.AirChange(
active=models.PeriodicInterval(period=120, duration=120),
air_exch=0.25,
)
ventilation = models.MultipleVentilation((window, ach))
assert ventilation == baseline_vent
def test_ventilation_hingedwindow(baseline_form: model_generator.FormData):
room = models.Room(75)
window = models.HingedWindow(
active=models.PeriodicInterval(period=120, duration=10),
inside_temp=models.PiecewiseConstant((0, 24), (293,)),
outside_temp=data.GenevaTemperatures['Dec'],
window_height=1.6, window_width=1., opening_length=0.6,
)
baseline_form.ventilation_type = 'natural_ventilation'
baseline_form.windows_duration = 10
baseline_form.windows_frequency = 120
@ -72,9 +74,24 @@ def test_ventilation_hingedwindow(baseline_form: model_generator.FormData):
baseline_form.window_width = 1.
baseline_form.opening_distance = 0.6
ts = np.linspace(8, 16, 100)
np.testing.assert_allclose([window.air_exchange(room, t)+0.25 for t in ts],
[baseline_form.ventilation().air_exchange(room, t) for t in ts])
baseline_vent = baseline_form.ventilation()
assert isinstance(baseline_vent, models.MultipleVentilation)
baseline_window = baseline_vent.ventilations[0]
assert isinstance(baseline_window, models.HingedWindow)
window = models.HingedWindow(
active=models.PeriodicInterval(period=120, duration=10),
inside_temp=models.PiecewiseConstant((0, 24), (293,)),
outside_temp=baseline_window.outside_temp,
window_height=1.6, window_width=1., opening_length=0.6,
)
ach = models.AirChange(
active=models.PeriodicInterval(period=120, duration=120),
air_exch=0.25,
)
ventilation = models.MultipleVentilation((window, ach))
assert ventilation == baseline_vent
def test_ventilation_mechanical(baseline_form: model_generator.FormData):
@ -116,21 +133,27 @@ def test_ventilation_window_hepa(baseline_form: model_generator.FormData):
baseline_form.window_height = 1.6
baseline_form.opening_distance = 0.6
baseline_form.hepa_option = True
baseline_vent = baseline_form.ventilation()
assert isinstance(baseline_vent, models.MultipleVentilation)
baseline_window = baseline_vent.ventilations[0]
assert isinstance(baseline_window, models.SlidingWindow)
# Now build the equivalent ventilation instance directly, and compare.
room = models.Room(75)
window = models.SlidingWindow(
active=models.PeriodicInterval(period=120, duration=10),
inside_temp=models.PiecewiseConstant((0, 24), (293,)),
outside_temp=baseline_vent.ventilations[0].outside_temp,
outside_temp=baseline_window.outside_temp,
window_height=1.6, opening_length=0.6,
)
hepa = models.HEPAFilter(
active=models.PeriodicInterval(period=120, duration=120),
q_air_mech=250.,
)
ach = models.AirChange(active=models.PeriodicInterval(period=120, duration=120), air_exch=0.25)
ach = models.AirChange(
active=models.PeriodicInterval(period=120, duration=120),
air_exch=0.25,
)
ventilation = models.MultipleVentilation((window, hepa, ach))
assert ventilation == baseline_vent