"""
pyrad.io.write_data
====================
Functions for writing pyrad output data
.. autosummary::
:toctree: generated/
write_to_s3
write_csv_to_mysql
write_vol_csv
write_vol_kml
write_centroids
write_proc_periods
write_fixed_angle
write_ts_lightning
send_msg
write_alarm_msg
write_last_state
write_smn
write_timeseries_point
write_trt_info
write_trt_cell_data
write_trt_thundertracking_data
write_trt_cell_scores
write_trt_cell_lightning
write_trt_rpc
write_vpr_theo_params
write_vpr_info
write_rhi_profile
write_field_coverage
write_cdf
write_histogram
write_quantiles
write_multiple_points
write_multiple_points_grid
write_ts_polar_data
write_ts_grid_data
write_ts_ml
write_ts_stats
write_ts_cum
write_monitoring_ts
write_excess_gates
write_intercomp_scores_ts
write_colocated_gates
write_colocated_data
write_sun_hits
write_sun_retrieval
"""
from __future__ import print_function
import glob
import csv
import os
import re
from importlib.util import find_spec
from ..util import warn
import smtplib
from email.message import EmailMessage
import datetime
import numpy as np
import pandas as pd
from pyart.config import get_fillvalue, get_field_colormap, get_field_limits
from pyart.core import antenna_to_cartesian, cartesian_to_geographic_aeqd
from cmweather.cm import cmap_d
from .io_aux import generate_field_name_str
from .flock_utils import lock_file, unlock_file
from ..util.date_utils import detect_datetime_columns
from .read_data_other import read_intercomp_scores_ts
try:
import simplekml
_SIMPLEKML_AVAILABLE = True
except ImportError:
_SIMPLEKML_AVAILABLE = False
try:
import boto3
_BOTO3_AVAILABLE = True
except ImportError:
warn(
"boto3 is not installed, no copy to S3 bucket will be performed!",
use_debug=False,
)
_BOTO3_AVAILABLE = False
_SQLALCHEMY_AVAILABLE = find_spec("sqlalchemy") is not None
_PYMYSQL_AVAILABLE = find_spec("pymysql") is not None
_CRYPTO_AVAILABLE = find_spec("cryptography") is not None
_MYSQL_AVAILABLE = _SQLALCHEMY_AVAILABLE and _PYMYSQL_AVAILABLE and _CRYPTO_AVAILABLE
if not _MYSQL_AVAILABLE:
warn(
"sqlalchemy, pymysql and/or cryptography not installed, "
"no MySQL writing will be performed!"
)
else:
pass
def write_to_s3(
fname,
basepath,
s3endpoint,
s3bucket,
s3key,
s3secret,
s3path="",
s3accesspolicy=None,
s3splitext=False,
s3verify=True,
s3certificates="",
):
"""
Copies a locally stored product to a S3 bucket
Parameters
----------
fname : str
filename of the product on the local storage
basepath : str
basepath of pyrad products as provided with datapath key in
the main conf file
s3endpoint : str
s3 endpoint URL
for example
fr-par-1.linodeobjects.com/ or
https://eu-central-1.linodeobjects.com/ or
https://s3.us-east-1.amazonaws.com/
https:// prefix is optional!
or
s3bucket : str
name of the s3 bucket to use to access the data
the URL that will be polled is
https://s3bucket.s3endpoint
s3key: str
AWS API access key for the specified bucket
s3secret: str
AWS API secret for the corresponding s3key
s3path : str
name of the path where to store the data on the s3 bucket
the files will have the following url
https://s3bucket.s3endpoint/s3path/filename
Optional: default is empty string (place them at the root of the bucket)
s3accesspolicy : str
S3 access policy can be either private or public-read
Default is to use nothing which will inherit the policy of the bucket
s3splitext: bool
If True, the extension will be added before the filename in the s3 name
For example folderA/folderB/folderC/out.csv will become folderA/folderB/folderC/csv/out.csv
on the S3.
s3verify: bool
If False, ssl verification will be disabled when writing data to S3. Beware this can
be dangerous
s3certificates: str
Path of the certificates used for ssl verification. Only required if
you use custom certificates.
"""
if not _BOTO3_AVAILABLE:
warn("boto3 not installed, aborting writing to s3")
return
if not s3endpoint.startswith("https://"): # add prefix
s3endpoint = f"https://{s3endpoint}"
endpoint_raw = s3endpoint.replace("https://", "")
if s3path:
if not s3path.endswith("/"):
s3path += "/"
else:
s3path = ""
if s3splitext:
ext = os.path.splitext(fname)[1][1:]
fname_bis = os.path.join(os.path.dirname(fname), ext, os.path.basename(fname))
else:
fname_bis = fname
s3fname = s3path + fname_bis.replace(basepath, "")
if s3fname.startswith("/"): # Remove leading /
s3fname = s3fname[1:]
s3_client_config = {
"aws_access_key_id": s3key,
"endpoint_url": s3endpoint,
"aws_secret_access_key": s3secret,
}
if len(s3certificates):
s3verify = s3certificates
s3_client = boto3.client("s3", **s3_client_config, verify=s3verify)
if s3accesspolicy:
ExtraArgs = {"ACL": s3accesspolicy}
else:
ExtraArgs = None
s3copypath = f"https://{s3bucket}.{endpoint_raw}/{s3fname}"
s3_client.upload_file(fname, s3bucket, s3fname, ExtraArgs=ExtraArgs)
print(f"----- copying {fname} to {s3copypath}")
def write_csv_to_mysql(
hostname: str,
port: int,
database: str,
user: str,
password: str,
table_name: str,
csv_file: str,
chunksize: int = 1000,
drop_table: bool = False,
):
"""
Write a CSV file into a MySQL database table.
Behavior
--------
- The CSV is read into a pandas DataFrame.
- Column names are sanitized for MySQL compatibility.
- All likely datetime columns are auto-detected and converted.
- If at least one datetime column is found, the first detected datetime
column is used as the "date column". Only rows whose date value is not
already present in the database are inserted.
- If no datetime column is found, only rows that are not exact duplicates
of existing rows (across all columns) are inserted.
- The table is created automatically if it does not already exist.
Parameters
----------
hostname : str
MySQL server hostname.
port : int
MySQL server port.
database : str
Name of the target database.
user : str
MySQL username.
password : str
MySQL password.
table_name : str
Name of the target table.
csv_file : str
Path to the CSV file to import.
chunksize : int, optional
Number of rows inserted per batch. Default is 1000.
drop_table: bool, optional
Drop the table and recreate it if the format is not compatible with the CSV file. Default is False.
"""
if not _MYSQL_AVAILABLE:
warn(
"pymysql and/or sqlalchemy and/or cryptography not available, aborting writing to database"
)
return
import re
import hashlib
import pandas as pd
from sqlalchemy import create_engine, text
from sqlalchemy.types import Integer, Float, Boolean, DateTime, String
# -----------------------------
# 1) Read CSV with inference
# -----------------------------
df = pd.read_csv(csv_file, comment="#")
df = df.replace({-9999: None})
df = df.where(pd.notna(df), None) # converts NaN/NaT to None
def sanitize_columns(cols):
out = []
seen = set()
for c in cols:
c2 = str(c).strip()
c2 = re.sub(r"\s+", "_", c2)
c2 = re.sub(r"[^0-9a-zA-Z_]", "_", c2)
if re.match(r"^\d", c2):
c2 = f"c_{c2}"
base = c2
i = 1
while c2 in seen:
c2 = f"{base}_{i}"
i += 1
seen.add(c2)
out.append(c2)
return out
df.columns = sanitize_columns(df.columns)
# -----------------------------
# 2) Auto-detect datetime columns
# -----------------------------
datetime_cols = detect_datetime_columns(df)
date_col = datetime_cols[0] if datetime_cols else None
# -----------------------------
# 3) Map pandas dtypes → MySQL
# -----------------------------
def infer_sqlalchemy_type(series):
if pd.api.types.is_integer_dtype(series):
return Integer()
elif pd.api.types.is_float_dtype(series):
return Float()
elif pd.api.types.is_bool_dtype(series):
return Boolean()
elif pd.api.types.is_datetime64_any_dtype(series):
return DateTime()
else:
non_null = series.dropna()
if len(non_null) == 0:
return String(255)
max_len = int(non_null.astype(str).str.len().max())
return String(max(10, min(max_len, 1000)))
dtype_mapping = {col: infer_sqlalchemy_type(df[col]) for col in df.columns}
# -----------------------------
# 4) Connect to MySQL
# -----------------------------
engine = create_engine(
f"mysql+pymysql://{user}:{password}@{hostname}:{port}/{database}",
echo=False,
future=True,
)
# -----------------------------
# 5) Create table if not exists.
# If drop_table=True, only drop/recreate when schema differs.
# -----------------------------
def get_existing_columns(conn, table_name):
result = conn.execute(
text(
"""
SELECT COLUMN_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = :database
AND TABLE_NAME = :table_name
ORDER BY ORDINAL_POSITION
"""
),
{"database": database, "table_name": table_name},
)
return [row[0] for row in result.fetchall()]
with engine.begin() as conn:
existing_columns = get_existing_columns(conn, table_name)
csv_columns = df.columns.tolist()
if existing_columns:
same_format = existing_columns == csv_columns
if drop_table and not same_format:
print(
f"Table {database}.{table_name} exists but has a different format. "
"Dropping and recreating it."
)
print(f"Existing columns: {existing_columns}")
print(f"CSV columns: {csv_columns}")
conn.execute(text(f"DROP TABLE IF EXISTS `{table_name}`"))
elif not same_format:
raise ValueError(
f"Table {database}.{table_name} has a different format than the CSV file.\n"
f"Existing columns: {existing_columns}\n"
f"CSV columns: {csv_columns}\n"
"Set drop_table=True to allow automatic recreation."
)
df.head(0).to_sql(
table_name,
engine,
if_exists="append",
index=False,
dtype=dtype_mapping,
)
# -----------------------------
# 6) Remove duplicates inside input dataframe itself
# -----------------------------
if date_col is not None:
df = df.drop_duplicates(subset=[date_col], keep="last")
else:
df = df.drop_duplicates(keep="last")
# -----------------------------
# 7) Filter rows already present in DB
# -----------------------------
with engine.begin() as conn:
if date_col is not None:
# Fetch existing date values from DB
existing_dates_df = pd.read_sql(
text(f"SELECT `{date_col}` FROM `{table_name}`"),
conn,
)
if not existing_dates_df.empty:
existing_dates = pd.to_datetime(
existing_dates_df[date_col], errors="coerce"
)
df = df[~df[date_col].isin(existing_dates)]
else:
# No date column found:
# compare whole rows against existing DB content
existing_df = pd.read_sql(
text(f"SELECT * FROM `{table_name}`"),
conn,
)
if not existing_df.empty:
# Align columns just in case
existing_df = existing_df[df.columns.tolist()]
def normalize_value(v):
if pd.isna(v) or v is None:
return ""
if isinstance(v, pd.Timestamp):
return v.isoformat()
return str(v)
def row_hash(row):
joined = "\x1f".join(normalize_value(v) for v in row)
return hashlib.sha256(joined.encode("utf-8")).hexdigest()
existing_hashes = set(existing_df.apply(row_hash, axis=1).tolist())
new_hashes = df.apply(row_hash, axis=1)
df = df.loc[~new_hashes.isin(existing_hashes)].copy()
# Nothing left to insert
if df.empty:
print(
f"No new rows to insert into {database}.{table_name} from file {csv_file}"
)
return
# Convert pandas NaT/NaN again after filtering
df = df.astype(object)
df = df.where(pd.notna(df), None)
# -----------------------------
# 8) Insert remaining rows
# -----------------------------
columns = ", ".join(f"`{c}`" for c in df.columns)
placeholders = ", ".join(f":{c}" for c in df.columns)
insert_stmt = text(
f"INSERT INTO `{table_name}` ({columns}) VALUES ({placeholders})"
)
with engine.begin() as conn:
for start in range(0, len(df), chunksize):
chunk = df.iloc[start : start + chunksize]
conn.execute(insert_stmt, chunk.to_dict(orient="records"))
if date_col is not None:
print(
f"File {csv_file} successfully written to MySQL database {database} "
f"in table {table_name}. Uniqueness was checked using date column '{date_col}'."
)
else:
print(
f"File {csv_file} successfully written to MySQL database {database} "
f"in table {table_name}. No date column was found, so full-row duplicates were skipped."
)
[docs]
def write_vol_csv(fname, radar, field_name, ignore_masked=False):
"""
Creates a kml file with the radar volume data
Parameters
----------
fname : str
kml file name
radar : radar object
radar object containing the data
field_name : str
field name to convert
ignore_masked : bool
if True the masked values will be ignored
Returns
-------
fname : str
the name of the file containing the content
"""
df = pd.DataFrame()
df["lon"] = radar.gate_longitude["data"][0, :]
df["lat"] = radar.gate_latitude["data"][0, :]
df["alt"] = radar.gate_altitude["data"][0, :]
if radar.azimuth["data"].size == df.shape[0]:
df["azi"] = radar.azimuth["data"]
df["ele"] = radar.elevation["data"]
df["rng"] = radar.range["data"]
df[field_name] = radar.fields[field_name]["data"][0, :]
if ignore_masked:
df.dropna(inplace=True)
df.to_csv(fname, index=False)
return fname
[docs]
def write_vol_kml(
fname, radar, field_name, ignore_masked=False, rng_res_km=0.24, azi_res=0.5
):
"""
Creates a kml file with the radar volume data
Parameters
----------
fname : str
kml file name
radar : radar object
radar object containing the data
field_name : str
field name to convert
ignore_masked : bool
if True the masked values will be ignored
rng_res_km : float
range resolution in km
azi_res : float
azimuth resolution (deg)
Returns
-------
fname : str
the name of the file containing the content
"""
if not _SIMPLEKML_AVAILABLE:
warn("simplekml module required to write kml files")
return None
df = pd.DataFrame()
df["lon"] = radar.gate_longitude["data"][0, :]
df["lat"] = radar.gate_latitude["data"][0, :]
df["alt"] = radar.gate_altitude["data"][0, :]
if radar.azimuth["data"].size == df.shape[0]:
df["azi"] = radar.azimuth["data"]
df["ele"] = radar.elevation["data"]
df["rng"] = radar.range["data"]
df[field_name] = radar.fields[field_name]["data"][0, :]
if ignore_masked:
df.dropna(inplace=True)
x1, y1, z1 = antenna_to_cartesian(
df["rng"].values / 1000.0 - rng_res_km / 2.0,
df["azi"].values - azi_res / 2,
df["ele"].values,
)
x2, y2, z2 = antenna_to_cartesian(
df["rng"].values / 1000.0 - rng_res_km / 2.0,
df["azi"].values + azi_res / 2,
df["ele"].values,
)
x3, y3, z3 = antenna_to_cartesian(
df["rng"].values / 1000.0 + rng_res_km / 2.0,
df["azi"].values + azi_res / 2,
df["ele"].values,
)
x4, y4, z4 = antenna_to_cartesian(
df["rng"].values / 1000.0 + rng_res_km / 2.0,
df["azi"].values - azi_res / 2,
df["ele"].values,
)
lon1, lat1 = cartesian_to_geographic_aeqd(
x1, y1, radar.longitude["data"][0], radar.latitude["data"][0]
)
lon2, lat2 = cartesian_to_geographic_aeqd(
x2, y2, radar.longitude["data"][0], radar.latitude["data"][0]
)
lon3, lat3 = cartesian_to_geographic_aeqd(
x3, y3, radar.longitude["data"][0], radar.latitude["data"][0]
)
lon4, lat4 = cartesian_to_geographic_aeqd(
x4, y4, radar.longitude["data"][0], radar.latitude["data"][0]
)
df["lon1"] = lon1
df["lon2"] = lon2
df["lon3"] = lon3
df["lon4"] = lon4
df["lat1"] = lat1
df["lat2"] = lat2
df["lat3"] = lat3
df["lat4"] = lat4
df["alt1"] = z1
df["alt2"] = z2
df["alt3"] = z3
df["alt4"] = z4
kml = simplekml.Kml()
df.apply(
lambda X: kml.newpolygon(
name=f'azi={X["azi"]}, ele={X["ele"]}, rng={X["rng"]}',
outerboundaryis=[
(X["lon1"], X["lat1"], X["alt1"]),
(X["lon2"], X["lat2"], X["alt2"]),
(X["lon3"], X["lat3"], X["alt3"]),
(X["lon4"], X["lat4"], X["alt4"]),
(X["lon1"], X["lat1"], X["alt1"]),
],
altitudemode="absolute",
),
axis=1,
)
# color value
cmap = cmap_d[get_field_colormap(field_name).split("_")[1]]
vmin, vmax = get_field_limits(field_name)
col_number = (df[field_name].values - vmin) / (vmax - vmin)
for ind, pol in enumerate(kml.features):
r, g, b, a = cmap(col_number[ind], bytes=True)
pol.style.polystyle.color = simplekml.Color.rgb(r, g, b)
kml.save(fname)
return fname
[docs]
def write_centroids(fname, centroids_dict, var_names):
"""
writes the centroids of a semi-supervised hydrometeor classification
Parameters
----------
fname : str
file name
centroids_dict : dict
dictionary containing the centroids
var_names : tuple
List of variables
Returns
-------
fname : str
the name of the file containing the content
"""
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Centroids file\n")
csvfile.write("# Header lines with comments are preceded by '#'\n")
csvfile.write("#\n")
field_names = np.append("hydro_type", var_names)
writer = csv.DictWriter(csvfile, field_names)
writer.writeheader()
for hydro_type in centroids_dict.keys():
dict_row = {"hydro_type": hydro_type}
for ivar, var_name in enumerate(var_names):
dict_row.update({var_name: centroids_dict[hydro_type][ivar]})
writer.writerow(dict_row)
csvfile.close()
return fname
[docs]
def write_proc_periods(start_times, end_times, fname):
"""
writes an output file containing start and stop times of periods to
process
Parameters
----------
start_times, end_times : datetime object
The starting and ending times of the periods
fname : str
The name of the file where to write
Returns
-------
fname : str
the name of the file containing the content
"""
with open(fname, "w", newline="") as csvfile:
field_names = ["start_time", "end_time"]
writer = csv.DictWriter(csvfile, field_names)
writer.writeheader()
for start_time, end_time in zip(start_times, end_times):
dict_row = {
"start_time": start_time.strftime("%Y%m%d%H%M%S"),
"end_time": end_time.strftime("%Y%m%d%H%M%S"),
}
writer.writerow(dict_row)
csvfile.close()
return fname
[docs]
def write_fixed_angle(time_data, fixed_angle, rad_lat, rad_lon, rad_alt, fname):
"""
writes an output file with the fixed angle data
Parameters
----------
time_data : datetime object
The scan time
fixed_angle : float
The first fixed angle in the scan
rad_lat, rad_lon, rad_alt : float
Latitude, longitude [deg] and altitude [m MSL] of the radar
fname : str
The name of the file where to write
Returns
-------
fname : str
the name of the file containing the content
"""
filelist = glob.glob(fname)
if not filelist:
with open(fname, "w", newline="") as csvfile:
fieldnames = ["date_time", "rad_lat", "rad_lon", "rad_alt", "fixed_angle"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
writer.writerow(
{
"date_time": time_data.strftime("%Y%m%d%H%M%S"),
"rad_lat": rad_lat,
"rad_lon": rad_lon,
"rad_alt": rad_alt,
"fixed_angle": fixed_angle,
}
)
csvfile.close()
else:
with open(fname, "a", newline="") as csvfile:
fieldnames = ["date_time", "rad_lat", "rad_lon", "rad_alt", "fixed_angle"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writerow(
{
"date_time": time_data.strftime("%Y%m%d%H%M%S"),
"rad_lat": rad_lat,
"rad_lon": rad_lon,
"rad_alt": rad_alt,
"fixed_angle": fixed_angle,
}
)
csvfile.close()
[docs]
def write_ts_lightning(
flashnr,
time_data,
time_in_flash,
lat,
lon,
alt,
dBm,
vals_list,
fname,
pol_vals_labels,
):
"""
writes the LMA sources data and the value of the colocated polarimetric
variables
Parameters
----------
flashnr : int
flash number
time_data : datetime object
flash source time
time_in_flash : float
seconds since start of flash
lat, lon, alt : float
latitude, longitude [deg] and altitude [m MSL] of the flash source
dBm : float
flash power
vals_list : list of arrays
List containing the data for each polarimetric variable
fname : str
the name of the file containing the content
pol_values_labels : list of strings
List containing strings identifying each polarimetric variable
Returns
-------
fname : str
the name of the file containing the content
"""
with open(fname, "w", newline="") as csvfile:
vals_list_aux = []
for j, label in enumerate(pol_vals_labels):
vals_list_aux.append(vals_list[j].filled(fill_value=get_fillvalue()))
csvfile.write("# Weather radar timeseries data file\n")
csvfile.write("# Project: MALSplus\n")
if time_data.size > 0:
csvfile.write(
"# Start : %s UTC\n" % time_data[0].strftime("%Y-%m-%d %H:%M:%S")
)
csvfile.write(
"# End : %s UTC\n" % time_data[-1].strftime("%Y-%m-%d %H:%M:%S")
)
csvfile.write("# Header lines with comments are preceded by '#'\n")
csvfile.write("#\n")
field_names = [
"flashnr",
"time_data",
"time_in_flash",
"lat",
"lon",
"alt",
"dBm",
]
field_names.extend(pol_vals_labels)
writer = csv.DictWriter(csvfile, field_names)
writer.writeheader()
if flashnr.size == 0.0:
warn("No data to write in file " + fname)
csvfile.close()
return fname
for i, flash in enumerate(flashnr):
dict_row = {
"flashnr": int(flash),
"time_data": time_data[i].strftime("%Y-%m-%d %H:%M:%S.%f"),
"time_in_flash": time_in_flash[i],
"lat": lat[i],
"lon": lon[i],
"alt": alt[i],
"dBm": dBm[i],
}
for j, label in enumerate(pol_vals_labels):
dict_row.update({label: vals_list_aux[j][i]})
writer.writerow(dict_row)
csvfile.close()
return fname
[docs]
def send_msg(sender, receiver_list, subject, fname):
"""
sends the content of a text file by email
Parameters
----------
sender : str
the email address of the sender
receiver_list : list of string
list with the email addresses of the receiver
subject : str
the subject of the email
fname : str
name of the file containing the content of the email message
Returns
-------
fname : str
the name of the file containing the content
"""
# Create the container email message.
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = sender
msg["To"] = ", ".join(receiver_list)
# Open the plain text file whose name is in fname for reading.
with open(fname) as fp:
# Create a text/plain message
msg.set_content(fp.read())
# Send the message via our own SMTP server.
with smtplib.SMTP("localhost") as s:
s.send_message(msg)
return fname
[docs]
def write_alarm_msg(
radar_name,
param_name_unit,
date_last,
target,
tol_abs,
np_trend,
value_trend,
tol_trend,
nevents,
np_last,
value_last,
fname,
):
"""
writes an alarm file
Parameters
----------
radar_name : str
Name of the radar being controlled
param_name_unit : str
Parameter and units
date_last : datetime object
date of the current event
target, tol_abs : float
Target value and tolerance
np_trend : int
Total number of points in trend
value_trend, tol_trend : float
Trend value and tolerance
nevents: int
Number of events in trend
np_last : int
Number of points in the current event
value_last : float
Value of the current event
fname : str
Name of file where to store the alarm information
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as txtfile:
txtfile.write("Weather radar polarimetric parameters monitoring alarm\n")
if radar_name is not None:
txtfile.write("Radar name: " + radar_name + "\n")
txtfile.write("Parameter [Unit]: " + param_name_unit + "\n")
txtfile.write("Date : " + date_last.strftime("%Y%m%d") + "\n")
txtfile.write("Target value: " + str(target) + " +/- " + str(tol_abs) + "\n")
if np_trend > 0:
txtfile.write(
"Number of points in trend: "
+ str(np_trend)
+ " in "
+ str(nevents)
+ " events\n"
)
txtfile.write(
"Trend value: "
+ str(value_trend)
+ " (tolerance: +/- "
+ str(tol_trend)
+ ")\n"
)
else:
txtfile.write("Number of points in trend: NA\n")
txtfile.write("Trend value: NA\n")
txtfile.write("Number of points: " + str(np_last) + "\n")
txtfile.write("Value: " + str(value_last) + "\n")
txtfile.close()
return fname
[docs]
def write_last_state(datetime_last, fname):
"""
writes SwissMetNet data in format datetime,avg_value, std_value
Parameters
----------
datetime_last : datetime object
date and time of the last state
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
try:
with open(fname, "w", newline="") as txtfile:
txtfile.write(datetime_last.strftime("%Y%m%d%H%M%S"))
txtfile.close()
return fname
except EnvironmentError:
warn("Unable to write on file " + fname)
return None
[docs]
def write_smn(datetime_vec, value_avg_vec, value_std_vec, fname):
"""
writes SwissMetNet data in format datetime,avg_value, std_value
Parameters
----------
datetime_vec : datetime array
array containing the measurement time
value_avg_vec : float array
array containing the average value
value_std_vec : float array
array containing the standard deviation
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
fieldnames = ["datetime", "avg", "std"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, value_avg in enumerate(value_avg_vec):
writer.writerow(
{
"datetime": datetime_vec[i].strftime("%Y%m%d%H%M%S"),
"avg": value_avg,
"std": value_std_vec[i],
}
)
csvfile.close()
return fname
def write_timeseries_point(
fname,
data,
dstype,
text,
timeformat=None,
timeinfo=None,
):
"""
Write one timesample of a time series to a CSV file using pandas.
Parameters
----------
data : dict
Dictionary containing:
time : datetime object
label[] : list of column names
value[] : list/array of values
unit : data unit string
dstype : str
Data type name
text : list[str]
Description text written as metadata comments
timeformat : str, optional
Datetime format string
timeinfo : optional
Unused (kept for compatibility)
"""
print("----- Write timeseries", fname)
datatime = data["time"]
if timeformat is None:
tformat = "%Y-%m-%d %H:%M:%S"
else:
tformat = timeformat
time_str = datatime.strftime(tformat)
# Create dataframe row
row = {"Date": time_str}
for label, value in zip(data["label"], data["value"]):
row[label] = round(float(value), 4)
df = pd.DataFrame([row])
# File does not exist -> create with metadata + header
if not os.path.isfile(fname):
with open(fname, "w", encoding="utf-8") as f:
f.write("# Weather radar timeseries data file\n")
f.write("# Project: MALSplus\n")
f.write(f"# Data/Unit : {dstype} [{data['unit']}]\n")
f.write(f"# Start : {datatime.strftime('%Y-%m-%d %H:%M:%S')} UTC\n")
for line in text:
f.write(f"# {line}\n")
# Write dataframe with clean column names (no #)
df.to_csv(
fname,
mode="a",
index=False,
float_format="%.4f",
)
else:
# Append without header
df.to_csv(
fname,
mode="a",
index=False,
header=False,
float_format="%.4f",
)
return fname
[docs]
def write_trt_info(ids, max_rank, nscans, time_start, time_end, fname):
"""
writes TRT info of the thundertracking
Parameters
----------
ids, max_rank, nscans, time_start, time_end: array
the cell parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
fieldnames = ["id", "max_rank", "nscans_Xband", "time_start", "time_end"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, id_cell in enumerate(ids):
writer.writerow(
{
"id": id_cell,
"max_rank": max_rank[i],
"nscans_Xband": nscans[i],
"time_start": time_start[i].strftime("%Y%m%d%H%M"),
"time_end": time_end[i].strftime("%Y%m%d%H%M"),
}
)
csvfile.close()
return fname
[docs]
def write_trt_cell_data(
traj_ID,
yyyymmddHHMM,
lon,
lat,
ell_L,
ell_S,
ell_or,
area,
vel_x,
vel_y,
det,
RANKr,
CG_n,
CG_p,
CG,
CG_percent_p,
ET45,
ET45m,
ET15,
ET15m,
VIL,
maxH,
maxHm,
POH,
RANK,
Dvel_x,
Dvel_y,
cell_contour,
fname,
):
"""
writes TRT cell data
Parameters
----------
traj_ID, yyyymmddHHMM, lon, lat, ell_L, ell_S, ell_or, area,
vel_x, vel_y, det, RANKr, CG_n, CG_p, CG, CG_percent_p, ET45,
ET45m, ET15, ET15m, VIL, maxH, maxHm, POH, RANK, Dvel_x,
Dvel_y, cell_contour:
the cell parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
try:
with open(fname, "w", newline="") as csvfile:
fieldnames = [
"traj_ID",
"yyyymmddHHMM",
"lon",
"lat",
"ell_L",
"ell_S",
"ell_or",
"area",
"vel_x",
"vel_y",
"det",
"RANKr",
"CG-",
"CG+",
"CG",
"%CG+",
"ET45",
"ET45m",
"ET15",
"ET15m",
"VIL",
"maxH",
"maxHm",
"POH",
"RANK",
"Dvel_x",
"Dvel_y",
"cell_contour_lon-lat",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, traj_ID_el in enumerate(traj_ID):
cell_contour_aux = cell_contour[i]
npoints_contour = len(cell_contour_aux["lon"])
cell_contour_arr = np.empty(2 * npoints_contour, dtype=float)
cell_contour_arr[0:-1:2] = cell_contour_aux["lon"]
cell_contour_arr[1::2] = cell_contour_aux["lat"]
cell_contour_str = str(cell_contour_arr[0])
for j in range(1, 2 * npoints_contour):
cell_contour_str += " " + str(cell_contour_arr[j])
writer.writerow(
{
"traj_ID": traj_ID_el,
"yyyymmddHHMM": yyyymmddHHMM[i].strftime("%Y%m%d%H%M"),
"lon": lon[i],
"lat": lat[i],
"ell_L": ell_L[i],
"ell_S": ell_S[i],
"ell_or": ell_or[i],
"area": area[i],
"vel_x": vel_x[i],
"vel_y": vel_y[i],
"det": det[i],
"RANKr": RANKr[i],
"CG-": CG_n[i],
"CG+": CG_p[i],
"CG": CG[i],
"%CG+": CG_percent_p[i],
"ET45": ET45[i],
"ET45m": ET45m[i],
"ET15": ET15[i],
"ET15m": ET15m[i],
"VIL": VIL[i],
"maxH": maxH[i],
"maxHm": maxHm[i],
"POH": POH[i],
"RANK": RANK[i],
"Dvel_x": Dvel_x[i],
"Dvel_y": Dvel_y[i],
"cell_contour_lon-lat": cell_contour_str,
}
)
csvfile.close()
return fname
except EnvironmentError:
warn("Unable to write on file " + fname)
return None
[docs]
def write_trt_thundertracking_data(
traj_ID,
scan_ordered_time,
scan_time,
azi,
rng,
yyyymmddHHMM,
lon,
lat,
ell_L,
ell_S,
ell_or,
area,
vel_x,
vel_y,
det,
RANKr,
CG_n,
CG_p,
CG,
CG_percent_p,
ET45,
ET45m,
ET15,
ET15m,
VIL,
maxH,
maxHm,
POH,
RANK,
Dvel_x,
Dvel_y,
cell_contour,
fname,
):
"""
writes TRT cell data of the thundertracking scan
Parameters
----------
traj_ID, scan_ordered_time, scan_time, azi, rng, yyyymmddHHMM, lon, lat,
ell_L, ell_S, ell_or, area, vel_x, vel_y, det, RANKr, CG_n, CG_p, CG,
CG_percent_p, ET45, ET45m, ET15, ET15m, VIL, maxH, maxHm, POH, RANK,
Dvel_x, Dvel_y, cell_contour:
the cell parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
try:
with open(fname, "w", newline="") as csvfile:
fieldnames = [
"traj_ID",
"scan_ordered_time",
"scan_time",
"azi",
"rng",
"yyyymmddHHMM",
"lon",
"lat",
"ell_L",
"ell_S",
"ell_or",
"area",
"vel_x",
"vel_y",
"det",
"RANKr",
"CG-",
"CG+",
"CG",
"%CG+",
"ET45",
"ET45m",
"ET15",
"ET15m",
"VIL",
"maxH",
"maxHm",
"POH",
"RANK",
"Dvel_x",
"Dvel_y",
"cell_contour_lon-lat",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, traj_ID_el in enumerate(traj_ID):
cell_contour_aux = cell_contour[i]
npoints_contour = len(cell_contour_aux["lon"])
cell_contour_arr = np.empty(2 * npoints_contour, dtype=float)
cell_contour_arr[0:-1:2] = cell_contour_aux["lon"]
cell_contour_arr[1::2] = cell_contour_aux["lat"]
cell_contour_str = str(cell_contour_arr[0])
for j in range(1, 2 * npoints_contour):
cell_contour_str += " " + str(cell_contour_arr[j])
if np.ma.is_masked(scan_time[i]):
scan_time_str = get_fillvalue()
else:
scan_time_str = scan_time[i].strftime("%Y%m%d%H%M%S")
writer.writerow(
{
"traj_ID": traj_ID_el,
"scan_ordered_time": scan_ordered_time[i].strftime(
"%Y%m%d%H%M%S.%f"
),
"scan_time": scan_time_str,
"azi": azi[i],
"rng": rng[i],
"yyyymmddHHMM": yyyymmddHHMM[i].strftime("%Y%m%d%H%M"),
"lon": lon[i],
"lat": lat[i],
"ell_L": ell_L[i],
"ell_S": ell_S[i],
"ell_or": ell_or[i],
"area": area[i],
"vel_x": vel_x[i],
"vel_y": vel_y[i],
"det": det[i],
"RANKr": RANKr[i],
"CG-": CG_n[i],
"CG+": CG_p[i],
"CG": CG[i],
"%CG+": CG_percent_p[i],
"ET45": ET45[i],
"ET45m": ET45m[i],
"ET15": ET15[i],
"ET15m": ET15m[i],
"VIL": VIL[i],
"maxH": maxH[i],
"maxHm": maxHm[i],
"POH": POH[i],
"RANK": RANK[i],
"Dvel_x": Dvel_x[i],
"Dvel_y": Dvel_y[i],
"cell_contour_lon-lat": cell_contour_str,
}
)
csvfile.close()
return fname
except EnvironmentError:
warn("Unable to write on file " + fname)
return None
[docs]
def write_trt_cell_scores(
traj_ID,
flash_density_max_time,
flash_density_max_rank,
nflashes_max_list,
area_flash_max_list,
flash_density_max,
rank_max_time,
rank_max,
fname,
):
"""
writes TRT cells scores
Parameters
----------
traj_ID : array of ints
The ID of the cells
flash_density_max_time : array of date times
The time at which the maximum flash density was reached for each cell
flash_density_max_rank : array of floats
The rank when the maximum flash density was reached for each cell
nflashes_max_list : array of ints
the number of flashes when the max flash density was reached
area_flash_max_list : array of floats
The area when the max flash density was reached
flash_density_max : array of floats
The maximum flash density for each cell
rank_max_time : array of datetime
the time at wich the maximum rank of each cell was reached
rank_max : array of float
the rank when the maximum rank of each cell was reached
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
fieldnames = [
"traj ID",
"max flash density time",
"max flash density rank",
"max flash density flashes",
"max flash density area",
"max flash density",
"max rank time",
"max rank",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, traj_ID_el in enumerate(traj_ID):
writer.writerow(
{
"traj ID": traj_ID_el,
"max flash density time": flash_density_max_time[i].strftime(
"%Y-%m-%d %H:%M:%S"
),
"max flash density rank": flash_density_max_rank[i],
"max flash density flashes": nflashes_max_list[i],
"max flash density area": area_flash_max_list[i],
"max flash density": flash_density_max[i],
"max rank time": rank_max_time[i].strftime("%Y-%m-%d %H:%M:%S"),
"max rank": rank_max[i],
}
)
csvfile.close()
return fname
[docs]
def write_trt_cell_lightning(
cell_ID,
cell_time,
lon,
lat,
area,
rank,
nflash,
flash_density,
fname,
timeformat="%Y%m%d%H%M",
):
"""
writes the lightning data for each TRT cell
Parameters
----------
cell_ID : array of ints
the cell ID
cell_time : array of datetime
the time step
lon, lat : array of floats
the latitude and longitude of the center of the cell
area : array of floats
the area of the cell
rank : array of floats
the rank of the cell
nflash : array of ints
the number of flashes/sources within the cell
flash_density : array of floats
the flash/source density
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
nflash = nflash.filled(fill_value=get_fillvalue())
flash_density = flash_density.filled(fill_value=get_fillvalue())
with open(fname, "w", newline="") as csvfile:
fieldnames = [
"traj_ID",
"yyyymmddHHMM",
"lon",
"lat",
"area",
"RANKr",
"nflashes",
"flash_dens",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, traj_ID_el in enumerate(cell_ID):
writer.writerow(
{
"traj_ID": traj_ID_el,
"yyyymmddHHMM": cell_time[i].strftime(timeformat),
"lon": lon[i],
"lat": lat[i],
"area": area[i],
"RANKr": rank[i],
"nflashes": nflash[i],
"flash_dens": flash_density[i],
}
)
csvfile.close()
return fname
[docs]
def write_trt_rpc(
cell_ID,
cell_time,
lon,
lat,
area,
rank,
hmin,
hmax,
freq,
fname,
timeformat="%Y%m%d%H%M",
):
"""
writes the rimed particles column data for a TRT cell
Parameters
----------
cell_ID : array of ints
the cell ID
cell_time : array of datetime
the time step
lon, lat : array of floats
the latitude and longitude of the center of the cell
area : array of floats
the area of the cell
rank : array of floats
the rank of the cell
hmin, hmax : array of floats
Minimum and maximum altitude of the rimed particle column
freq : array of floats
Frequency of the species constituting the rime particle column within
the limits of it
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
hmin = hmin.filled(fill_value=get_fillvalue())
hmax = hmax.filled(fill_value=get_fillvalue())
freq = freq.filled(fill_value=get_fillvalue())
with open(fname, "w", newline="") as csvfile:
fieldnames = [
"traj_ID",
"yyyymmddHHMM",
"lon",
"lat",
"area",
"RANKr",
"hmin",
"hmax",
"freq",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, traj_ID_el in enumerate(cell_ID):
writer.writerow(
{
"traj_ID": traj_ID_el,
"yyyymmddHHMM": cell_time[i].strftime(timeformat),
"lon": lon[i],
"lat": lat[i],
"area": area[i],
"RANKr": rank[i],
"hmin": hmin[i],
"hmax": hmax[i],
"freq": freq[i],
}
)
csvfile.close()
return fname
[docs]
def write_vpr_theo_params(
vpr_theo_dict, fname, labels=("ml_top", "ml_bottom", "val_ml_peak", "val_dr")
):
"""
writes the parameters defining a theoretical VPR profile into a csv file
Parameters
----------
vpr_theo_dict : dict
dictionary containing the parameters of the VPR profile
fname : str
file name where to store the data
labels : list of str
new of parameters to write in file
Returns
-------
fname : str
the name of the file where data has been written
"""
with open(fname, "w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, labels)
writer.writeheader()
data_dict = dict((k, vpr_theo_dict[k]) for k in labels)
writer.writerow(data_dict)
csvfile.close()
return fname
[docs]
def write_vpr_info(vpr_info_dict, fname):
"""
Writes a dictionary containing relevant parameters of the VPR profile
retrieval into a text file
Parameters
----------
vpr_info_dict : dict
dictionary containing the parameters of the VPR profile
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has been written
"""
with open(fname, "w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, vpr_info_dict.keys())
writer.writeheader()
writer.writerow(vpr_info_dict)
csvfile.close()
return fname
[docs]
def write_rhi_profile(
hvec, data, nvalid_vec, labels, fname, datatype=None, timeinfo=None, sector=None
):
"""
writes the values of an RHI profile in a text file
Parameters
----------
hvec : float array
array containing the alitude in m MSL
data : list of float array
the quantities at each altitude
nvalid_vec : int array or None
number of valid data points used to compute the quantiles
labels : list of strings
label specifying the quantitites in data
fname : str
file name where to store the data
datatype : str
the data type
timeinfo : datetime object
time of the rhi profile
sector : dict
dictionary specying the sector limits
Returns
-------
fname : str
the name of the file where data has been written
"""
data_aux = list()
for data_points in data:
data_aux.append(data_points.filled(fill_value=get_fillvalue()))
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Height Profile\n")
csvfile.write("# ==============\n")
csvfile.write("#\n")
if datatype is None:
csvfile.write("# Datatype (Unit) : Not specified\n")
else:
csvfile.write("# Datatype (Unit) : " + datatype + "\n")
csvfile.write("# Fill value : " + str(get_fillvalue()) + "\n")
if timeinfo is None:
csvfile.write("# Time : Not specified\n")
else:
csvfile.write(
"# Time : " + timeinfo.strftime("%Y-%m-%d %H:%M:%S") + "\n"
)
if sector is None:
csvfile.write("# Sector specification: None\n")
else:
csvfile.write("# Sector specification:\n")
csvfile.write("# Azimuth : " + str(sector["az"]) + " deg\n")
csvfile.write("# Range start : " + str(sector["rmin"]) + " m\n")
csvfile.write("# Range stop : " + str(sector["rmax"]) + " m\n")
fieldnames = ["Altitude [m MSL]"]
fieldnames.extend(labels)
if nvalid_vec is not None:
fieldnames.append("N valid")
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for j, height in enumerate(hvec):
data_dict = {fieldnames[0]: height}
for i in range(len(labels)):
data_dict.update({fieldnames[i + 1]: data_aux[i][j]})
if nvalid_vec is not None:
data_dict.update({fieldnames[-1]: nvalid_vec[j]})
writer.writerow(data_dict)
csvfile.close()
return fname
[docs]
def write_field_coverage(
quantiles,
values,
ele_start,
ele_stop,
azi_start,
azi_stop,
threshold,
nvalid_min,
datatype,
timeinfo,
fname,
):
"""
writes the quantiles of the coverage on a particular sector
Parameters
----------
quantiles : datetime array
array containing the quantiles computed
values : float array
quantile value
ele_start, ele_stop, azi_start, azi_stop : float
The limits of the sector
threshold : float
The minimum value to consider the data valid
nvalid_min : int
the minimum number of points to consider that there are values in a
ray
datatype : str
data type and units
timeinfo : datetime object
the time stamp of the data
fname : str
name of the file where to write the data
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Quantiles of the field coverage\n")
csvfile.write("# Datatype (Unit) : " + datatype + "\n")
csvfile.write("# Time : " + timeinfo.strftime("%Y-%m-%d %H:%M:%S") + "\n")
csvfile.write("# Sector specification:\n")
csvfile.write("# Azimuth start : " + str(azi_start) + " deg\n")
csvfile.write("# Azimuth stop : " + str(azi_stop) + " deg\n")
csvfile.write("# Elevation start : " + str(ele_start) + " deg\n")
csvfile.write("# Elevation stop : " + str(ele_stop) + " deg\n")
csvfile.write("# Threshold : " + str(threshold) + "\n")
csvfile.write(
"# Minimum number of valid gates per ray : " + str(nvalid_min) + "\n"
)
fieldnames = ["Quantile [%]", "Rain extension [m]"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, quant in enumerate(quantiles):
writer.writerow({"Quantile [%]": quant, "Rain extension [m]": values[i]})
csvfile.close()
return fname
[docs]
def write_cdf(
quantiles,
values,
ntot,
nnan,
nclut,
nblocked,
nprec_filter,
noutliers,
ncdf,
fname,
use_nans=False,
nan_value=0.0,
filterprec=[],
vismin=None,
sector=None,
datatype=None,
timeinfo=None,
):
"""
writes a cumulative distribution function
Parameters
----------
quantiles : datetime array
array containing the measurement time
values : float array
array containing the average value
fname : float array
array containing the standard deviation
sector : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
hydrotype_list = ["NC", "DS", "CR", "LR", "GR", "RN", "VI", "WS", "MH", "IH/HDG"]
with open(fname, "w", newline="") as txtfile:
txtfile.write("Statistical analysis\n")
txtfile.write("====================\n\n")
if datatype is None:
txtfile.write("Datatype (Unit) : Not specified\n")
else:
txtfile.write("Datatype (Unit) : " + datatype + "\n")
if timeinfo is None:
txtfile.write("Time : Not specified\n")
else:
txtfile.write("Time : " + timeinfo.strftime("%Y-%m-%d %H:%M:%S") + "\n")
if sector is None:
txtfile.write("Sector specification: None\n")
else:
txtfile.write("Sector specification:\n")
if sector["rmin"] is None:
txtfile.write(" Range start : Not specified\n")
else:
txtfile.write(" Range start : " + str(sector["rmin"]) + " m\n")
if sector["rmax"] is None:
txtfile.write(" Range stop : Not specified\n")
else:
txtfile.write(" Range stop : " + str(sector["rmax"]) + " m\n")
if sector["azmin"] is None:
txtfile.write(" Azimuth start : Not specified\n")
else:
txtfile.write(" Azimuth start : " + str(sector["azmin"]) + " deg\n")
if sector["azmax"] is None:
txtfile.write(" Azimuth stop : Not specified\n")
else:
txtfile.write(" Azimuth stop : " + str(sector["azmax"]) + " deg\n")
if sector["elmin"] is None:
txtfile.write(" Elevation start : Not specified\n")
else:
txtfile.write(" Elevation start : " + str(sector["elmin"]) + " deg\n")
if sector["elmax"] is None:
txtfile.write(" Elevation stop : Not specified\n")
else:
txtfile.write(" Elevation stop : " + str(sector["elmax"]) + " deg\n")
if sector["hmin"] is None:
txtfile.write(" Height start : Not specified\n")
else:
txtfile.write(" Height start : " + str(sector["hmin"]) + " m\n")
if sector["hmax"] is None:
txtfile.write(" Height stop : Not specified\n")
else:
txtfile.write(" Height stop : " + str(sector["hmax"]) + " m\n")
txtfile.write("")
txtfile.write("Total number of gates in sector : " + str(ntot) + "\n")
txtfile.write("Number of gates with no value (NaNs) : " + str(nnan) + "\n")
if use_nans:
txtfile.write(" NaNs are set to : " + str(nan_value) + "\n")
else:
txtfile.write(" NaNs are ignored!\n")
if nclut == -1:
txtfile.write("Clutter contaminated gates : " + "Not checked\n")
else:
txtfile.write("Clutter contaminated gates : " + str(nclut) + "\n")
if nblocked == -1:
txtfile.write("Blocked gates : " + "Not checked\n")
else:
txtfile.write(
"Blocked gates (vismin = "
+ str(int(vismin))
+ ") : "
+ str(nblocked)
+ "\n"
)
if nprec_filter == -1:
txtfile.write("Filtered precipitation gates : None\n")
else:
txtfile.write(
"Filtered precipitation gates : " + str(nprec_filter) + "\n"
)
txtfile.write(" precipitation types filtered: ")
for ind_hydro in filterprec:
txtfile.write(hydrotype_list[ind_hydro] + " ")
txtfile.write("\n")
txtfile.write("Number of outliers : " + str(noutliers) + "\n")
txtfile.write(
"Number of gates used for histogram : "
+ str(ncdf)
+ " ("
+ str(int(ncdf * 100.0 / ntot))
+ "%)\n\n"
)
txtfile.write("Quantiles\n")
txtfile.write("=========\n")
for i, value in enumerate(values):
txtfile.write(
" Quantile_"
+ str(int(quantiles[i]))
+ " = "
+ "{:5.1f}".format(value)
+ "\n"
)
txtfile.close()
return fname
[docs]
def write_histogram(bin_edges, values, fname, datatype="undefined", step=0):
"""
writes a histogram
Parameters
----------
bin_edges : float array
array containing the histogram bin edges
values : int array
array containing the number of points in each bin
fname : str
file name
datatype :str
The data type
step : str
The bin step
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
datatype_str = "undefined"
if datatype != "undefined":
datatype_str = generate_field_name_str(datatype)
csvfile.write(
"# Weather radar data histogram file\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description:\n"
+ "# Histogram of weather radar data.\n"
+ "# Data: "
+ datatype_str
+ "\n"
+ "# Step: "
+ str(step)
+ "\n"
"#\n"
)
fieldnames = ["bin_edge_left", "bin_edge_right", "value"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, val in enumerate(values):
writer.writerow(
{
"bin_edge_left": bin_edges[i],
"bin_edge_right": bin_edges[i + 1],
"value": val,
}
)
csvfile.close()
return fname
[docs]
def write_quantiles(quantiles, values, fname, datatype="undefined"):
"""
writes quantiles
Parameters
----------
quantiles : float array
array containing the quantiles to write
values : float array
array containing the value of each quantile
fname : str
file name
datatype :str
The data type
Returns
-------
fname : str
the name of the file where data has written
"""
values_aux = values.filled(fill_value=get_fillvalue())
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Weather radar data histogram file\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description:\n"
+ "# Histogram of weather radar data.\n"
+ "# Data: "
+ generate_field_name_str(datatype)
+ "\n"
+ "#\n"
)
fieldnames = ["quantile", "value"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, quant in enumerate(quantiles):
writer.writerow({"quantile": quant, "value": values_aux[i]})
csvfile.close()
return fname
[docs]
def write_multiple_points(dataset, fname):
"""
write radar data obtained at multiple points
Parameters
----------
dataset : dict
dictionary containing the data
fname : str
file name
Returns
-------
fname : str
the name of the file where data has written
"""
val = dataset["value"].filled(fill_value=get_fillvalue())
lon_ref = dataset["used_point_coordinates_WGS84_lon"].filled(
fill_value=get_fillvalue()
)
lat_ref = dataset["used_point_coordinates_WGS84_lat"].filled(
fill_value=get_fillvalue()
)
alt_ref = dataset["used_point_coordinates_WGS84_alt"].filled(
fill_value=get_fillvalue()
)
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Weather radar data at multiple points file\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description:\n"
+ "# weather radar data at points of interest.\n"
+ "# Data: "
+ generate_field_name_str(dataset["datatype"])
+ "\n"
+ "#\n"
)
fieldnames = [
"point_ID",
"time",
"azi",
"ele",
"rng",
"lon",
"lat",
"alt",
"lon_ref",
"lat_ref",
"alt_ref",
"val",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i in range(val.size):
time_aux = dataset["time"][i]
if not np.ma.is_masked(dataset["time"][i]):
time_aux = time_aux.strftime("%Y%m%d%H%M%S")
else:
time_aux = "NA"
writer.writerow(
{
"point_ID": dataset["point_id"][i],
"time": time_aux,
"azi": dataset["antenna_coordinates_az"][i],
"ele": dataset["antenna_coordinates_el"][i],
"rng": dataset["antenna_coordinates_r"][i],
"lon": dataset["point_coordinates_WGS84_lon"][i],
"lat": dataset["point_coordinates_WGS84_lat"][i],
"alt": dataset["point_coordinates_WGS84_alt"][i],
"lon_ref": lon_ref[i],
"lat_ref": lat_ref[i],
"alt_ref": alt_ref[i],
"val": val[i],
}
)
csvfile.close()
return fname
[docs]
def write_multiple_points_grid(dataset, fname):
"""
write data obtained at multiple points from a grid
Parameters
----------
dataset : dict
dictionary containing the data
fname : str
file name
Returns
-------
fname : str
the name of the file where data has written
"""
val = dataset["value"].filled(fill_value=get_fillvalue())
lon_ref = dataset["used_point_coordinates_WGS84_lon"].filled(
fill_value=get_fillvalue()
)
lat_ref = dataset["used_point_coordinates_WGS84_lat"].filled(
fill_value=get_fillvalue()
)
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Gridded data at multiple points file\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description:\n"
+ "# Gridded data at points of interest.\n"
+ "# Data: "
+ generate_field_name_str(dataset["datatype"])
+ "\n"
+ "#\n"
)
fieldnames = [
"point_ID",
"time",
"ix",
"iy",
"lon",
"lat",
"lon_ref",
"lat_ref",
"val",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i in range(val.size):
time_aux = dataset["time"][i]
if not np.ma.is_masked(dataset["time"][i]):
time_aux = time_aux.strftime("%Y%m%d%H%M%S")
else:
time_aux = "NA"
writer.writerow(
{
"point_ID": dataset["point_id"][i],
"time": time_aux,
"ix": dataset["grid_point_ix"][i],
"iy": dataset["grid_point_iy"][i],
"lon": dataset["point_coordinates_WGS84_lon"][i],
"lat": dataset["point_coordinates_WGS84_lat"][i],
"lon_ref": lon_ref[i],
"lat_ref": lat_ref[i],
"val": val[i],
}
)
csvfile.close()
return fname
[docs]
def write_ts_polar_data(dataset, fname):
"""
Write time series data to CSV with a comment header.
Single-radar mode
-----------------
dataset is a dict for one radar.
Output columns: ["date", "az", "el", "r", "value"]
Multi-radar mode
---------------
dataset is a dict with keys RADAR001, RADAR002, ...
Output columns: ["date", "lon", "lat", "alt", "value_RADAR001", "value_RADAR002", ...]
Values are stored in wide format (one row per timestamp).
Parameters
----------
dataset : dict
Single-radar dataset dict OR multi-radar dict keyed by RADAR###
fname : str
File path to write/append to
Returns
-------
fname : str
Output filename
"""
def _is_multiradar(ds):
return isinstance(ds, dict) and any(
isinstance(k, str) and re.match(r"^RADAR\d{3}$", k) for k in ds.keys()
)
def _to_dt_series(t):
if isinstance(t, (list, tuple, np.ndarray, pd.Series)):
return pd.to_datetime(pd.Series(t))
return pd.to_datetime(pd.Series([t]))
def _to_1d_values(v, n, fillvalue=np.nan):
# Convert scalars/lists/masked -> 1D float array length n
if np.ma.isMaskedArray(v):
v = v.filled(fillvalue)
if isinstance(v, (list, tuple, np.ndarray, pd.Series)):
arr = np.asarray(v)
else:
arr = np.asarray([v])
if arr.size == 1 and n > 1:
arr = np.repeat(arr, n)
# If it's an object array with masked scalars, convert those to NaN
if arr.dtype == object:
arr = np.array(
[fillvalue if x is np.ma.masked else x for x in arr], dtype=float
)
return arr.reshape(-1)
file_exists = os.path.exists(fname)
file_empty = file_exists and (os.path.getsize(fname) == 0)
multiradar = _is_multiradar(dataset)
# ----------------------------
# Build dataframe + header
# ----------------------------
if not multiradar:
# ---- single radar (legacy) ----
t = _to_dt_series(dataset["time"])
n = len(t)
used = dataset["used_antenna_coordinates_az_el_r"]
az = np.asarray(used[0]).reshape(-1)
el = np.asarray(used[1]).reshape(-1)
rr = np.asarray(used[2]).reshape(-1)
if az.size == 1 and n > 1:
az = np.repeat(az, n)
if el.size == 1 and n > 1:
el = np.repeat(el, n)
if rr.size == 1 and n > 1:
rr = np.repeat(rr, n)
val = _to_1d_values(dataset["value"], n, get_fillvalue())
df = pd.DataFrame(
{
"date": t.dt.strftime("%Y%m%d%H%M%S"),
"az": az[:n],
"el": el[:n],
"r": rr[:n],
"value": val[:n],
}
)
start_time = t.iloc[0].to_pydatetime()
header = (
"# Weather radar timeseries data file\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of a weather radar data over a fixed location.\n"
f"# Location [lon, lat, alt]: {dataset.get('point_coordinates_WGS84_lon_lat_alt')}\n"
f"# Nominal antenna coordinates used [az, el, r]: {dataset.get('antenna_coordinates_az_el_r')}\n"
f"# Data: {generate_field_name_str(dataset.get('datatype'))}\n"
f"# Fill Value: {get_fillvalue()}\n"
f"# Start: {start_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
"#\n"
)
col_order = ["date", "az", "el", "r", "value"]
else:
# ---- multi radar (wide format, outer-joined timestamps) ----
radar_keys = sorted(
[
k
for k in dataset.keys()
if isinstance(k, str) and re.match(r"^RADAR\d{3}$", k)
]
)
if not radar_keys:
raise ValueError("Multi-radar dataset detected but no RADAR### keys found")
ds0 = dataset[radar_keys[0]]
fillvalue = get_fillvalue()
radar_frames = []
for rk in radar_keys:
ds = dataset[rk]
t = _to_dt_series(ds["time"])
n = len(t)
val = _to_1d_values(ds["value"], n, fillvalue=fillvalue)
if len(val) < n:
val = np.pad(
val,
(0, n - len(val)),
mode="constant",
constant_values=fillvalue,
)
radar_df = pd.DataFrame(
{
"time": t,
f"value_{rk.lower()}": val[:n],
}
)
radar_frames.append(radar_df)
# Outer join all radar time axes
df = radar_frames[0]
for radar_df in radar_frames[1:]:
df = pd.merge(df, radar_df, on="time", how="outer")
df = df.sort_values("time").reset_index(drop=True)
# Fill missing radar values introduced by the outer join
value_cols = [f"value_{rk.lower()}" for rk in radar_keys]
df[value_cols] = df[value_cols].fillna(fillvalue)
# Coordinates: keep lon/lat/alt columns from reference radar
lon0, lat0, alt0 = ds0["point_coordinates_WGS84_lon_lat_alt"]
df["lon"] = float(lon0)
df["lat"] = float(lat0)
df["alt"] = float(alt0)
start_time = df["time"].iloc[0].to_pydatetime()
df["date"] = df["time"].dt.strftime("%Y%m%d%H%M%S")
# Header listing coords/antenna per radar
coord_lines = []
ant_lines = []
for rk in radar_keys:
ds = dataset[rk]
coord_lines.append(
f"# {rk} Location [lon, lat, alt]: "
f"{ds.get('point_coordinates_WGS84_lon_lat_alt')}"
)
ant_lines.append(
f"# {rk} Nominal antenna coordinates used [az, el, r]: "
f"{ds.get('antenna_coordinates_az_el_r')}"
)
header = (
"# Weather radar timeseries data file (multi-radar)\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of weather radar data over a fixed location "
"(multiple radars).\n"
+ "\n".join(coord_lines)
+ "\n"
+ "\n".join(ant_lines)
+ "\n"
f"# Data: {generate_field_name_str(ds0.get('datatype'))}\n"
f"# Fill Value: {fillvalue}\n"
f"# Start: {start_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
"#\n"
)
col_order = ["date", "lon", "lat", "alt"] + value_cols
df = df[col_order]
write_header = (not file_exists) or file_empty
mode = "a" if (file_exists and not write_header) else "w"
with open(fname, mode, encoding="utf-8", newline="") as f:
if write_header:
f.write(header)
df.to_csv(f, index=False)
else:
df.to_csv(f, index=False, header=False)
return fname
[docs]
def write_ts_grid_data(dataset, fname):
"""
Write gridded-data time series to CSV with a comment header.
Single-radar mode
-----------------
dataset is a dict for one series point.
Output columns: ["date", "lon", "lat", "alt", "value"]
Multi-radar mode
---------------
dataset is a dict with keys RADAR001, RADAR002, ...
Output columns: ["date", "lon", "lat", "alt", "value_radar001", "value_radar002", ...]
Values are stored in wide format (one row per timestamp).
Parameters
----------
dataset : dict
Single dataset dict OR multi-radar dict keyed by RADAR###
fname : str
File path to write/append to
Returns
-------
fname : str
Output filename
"""
def _is_multiradar(ds):
return isinstance(ds, dict) and any(
isinstance(k, str) and re.match(r"^RADAR\d{3}$", k) for k in ds.keys()
)
def _to_dt_series(t):
if isinstance(t, (list, tuple, np.ndarray, pd.Series)):
return pd.to_datetime(pd.Series(t))
return pd.to_datetime(pd.Series([t]))
def _to_1d_values(v, n):
if np.ma.isMaskedArray(v):
v = v.filled(np.nan)
if isinstance(v, (list, tuple, np.ndarray, pd.Series)):
arr = np.asarray(v)
else:
arr = np.asarray([v])
if arr.size == 1 and n > 1:
arr = np.repeat(arr, n)
if arr.dtype == object:
arr = np.array(
[np.nan if x is np.ma.masked else x for x in arr], dtype=float
)
return arr.reshape(-1)
file_exists = os.path.exists(fname)
file_empty = file_exists and (os.path.getsize(fname) == 0)
multiradar = _is_multiradar(dataset)
if not multiradar:
# ---- single radar (legacy-ish) ----
# dataset["time"] may be scalar or vector; grid writer historically wrote one row,
# but we support vectors too (like the polar writer).
t = _to_dt_series(dataset["time"])
n = len(t)
# Used coordinates can be scalars
used = dataset["used_coordinates_WGS84_lon_lat_alt"]
lon = float(used[0])
lat = float(used[1])
alt = float(used[2])
val = _to_1d_values(dataset["value"], n)
df = pd.DataFrame(
{
"date": t.dt.strftime("%Y%m%d%H%M%S"),
"lon": lon,
"lat": lat,
"alt": alt,
"value": val[:n],
}
)
start_time = t.iloc[0].to_pydatetime()
header = (
"# Gridded data timeseries data file\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of a gridded data over a fixed location.\n"
f"# Nominal location [lon, lat, alt]: {dataset.get('point_coordinates_WGS84_lon_lat_alt')}\n"
f"# Grid points used [iz, iy, ix]: {dataset.get('grid_points_iz_iy_ix')}\n"
f"# Data: {generate_field_name_str(dataset.get('datatype'))}\n"
f"# Fill Value: {get_fillvalue()}\n"
f"# Start: {start_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
"#\n"
)
col_order = ["date", "lon", "lat", "alt", "value"]
else:
# ---- multi radar (wide format, outer-joined timestamps) ----
radar_keys = sorted(
[
k
for k in dataset.keys()
if isinstance(k, str) and re.match(r"^RADAR\d{3}$", k)
]
)
if not radar_keys:
raise ValueError("Multi-radar dataset detected but no RADAR### keys found")
ds0 = dataset[radar_keys[0]]
fillvalue = get_fillvalue()
radar_frames = []
for rk in radar_keys:
ds = dataset[rk]
t = _to_dt_series(ds["time"])
n = len(t)
val = _to_1d_values(ds["value"], n)
if len(val) < n:
val = np.pad(
val,
(0, n - len(val)),
mode="constant",
constant_values=fillvalue,
)
radar_df = pd.DataFrame(
{
"time": t,
f"value_{rk.lower()}": val[:n],
}
)
radar_frames.append(radar_df)
# Outer join all radar time axes
df = radar_frames[0]
for radar_df in radar_frames[1:]:
df = pd.merge(df, radar_df, on="time", how="outer")
df = df.sort_values("time").reset_index(drop=True)
value_cols = [f"value_{rk.lower()}" for rk in radar_keys]
df[value_cols] = df[value_cols].fillna(fillvalue)
# Take lon/lat/alt columns from reference radar
used0 = ds0["used_coordinates_WGS84_lon_lat_alt"]
df["lon"] = float(used0[0])
df["lat"] = float(used0[1])
df["alt"] = float(used0[2])
start_time = df["time"].iloc[0].to_pydatetime()
df["date"] = df["time"].dt.strftime("%Y%m%d%H%M%S")
# Header: include per-radar nominal location and grid points if present
loc_lines = []
gp_lines = []
for rk in radar_keys:
ds = dataset[rk]
loc_lines.append(
f"# {rk} Nominal location [lon, lat, alt]: "
f"{ds.get('point_coordinates_WGS84_lon_lat_alt')}"
)
if "grid_points_iz_iy_ix" in ds:
gp_lines.append(
f"# {rk} Grid points used [iz, iy, ix]: "
f"{ds.get('grid_points_iz_iy_ix')}"
)
header = (
"# Gridded data timeseries data file (multi-radar)\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of gridded data over a fixed location "
"(multiple radars).\n"
+ "\n".join(loc_lines)
+ "\n"
+ ("\n".join(gp_lines) + "\n" if gp_lines else "")
+ f"# Data: {generate_field_name_str(ds0.get('datatype'))}\n"
+ f"# Fill Value: {fillvalue}\n"
+ f"# Start: {start_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
+ "#\n"
)
col_order = ["date", "lon", "lat", "alt"] + value_cols
df = df[col_order]
# Remove duplicates on time
df.drop_duplicates(subset=["date"], keep="last", inplace=True)
# Write / append
write_header = (not file_exists) or file_empty
mode = "a" if (file_exists and not write_header) else "w"
with open(fname, mode, encoding="utf-8", newline="") as f:
if write_header:
f.write(header)
df.to_csv(f, index=False)
else:
df.to_csv(f, index=False, header=False)
return fname
def write_ts_ml(
dt_ml, ml_top_avg, ml_top_std, thick_avg, thick_std, nrays_valid, nrays_total, fname
):
"""
writes time series of melting layer data
Parameters
----------
dt_ml : date time array
array of time steps
ml_top_avg, ml_top_std: float arrays
the average and the standard deviation of the melting layer top height
thick_avg, thick_std: float arrays
the average and the standard deviation of the metling layer thickness
nrays_valid, nrays_total: int arrays
the number of rays where melting layer has been identified and the
total number of arrays in the scan
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
filelist = glob.glob(fname)
if not filelist:
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Weather radar detected melting layer data file\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description: \n"
+ "# Time series of melting layer data detected by weather radar.\n"
+ "# Fill Value: "
+ str(get_fillvalue())
+ "\n"
+ "# Start: "
+ dt_ml.strftime("%Y-%m-%d %H:%M:%S UTC")
+ "\n"
+ "#\n"
)
fieldnames = [
"date-time [UTC]",
"mean ml top height [m MSL]",
"std ml top height [m MSL]",
"mean ml thickness [m]",
"std ml thickness [m]",
"N valid rays",
"rays total",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
writer.writerow(
{
"date-time [UTC]": dt_ml.strftime("%Y-%m-%d %H:%M:%S"),
"mean ml top height [m MSL]": ml_top_avg.filled(
fill_value=get_fillvalue()
),
"std ml top height [m MSL]": ml_top_std.filled(
fill_value=get_fillvalue()
),
"mean ml thickness [m]": thick_avg.filled(
fill_value=get_fillvalue()
),
"std ml thickness [m]": thick_std.filled(
fill_value=get_fillvalue()
),
"N valid rays": nrays_valid,
"rays total": nrays_total,
}
)
csvfile.close()
else:
with open(fname, "a", newline="") as csvfile:
fieldnames = [
"date-time [UTC]",
"mean ml top height [m MSL]",
"std ml top height [m MSL]",
"mean ml thickness [m]",
"std ml thickness [m]",
"N valid rays",
"rays total",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writerow(
{
"date-time [UTC]": dt_ml.strftime("%Y-%m-%d %H:%M:%S"),
"mean ml top height [m MSL]": ml_top_avg.filled(
fill_value=get_fillvalue()
),
"std ml top height [m MSL]": ml_top_std.filled(
fill_value=get_fillvalue()
),
"mean ml thickness [m]": thick_avg.filled(
fill_value=get_fillvalue()
),
"std ml thickness [m]": thick_std.filled(
fill_value=get_fillvalue()
),
"N valid rays": nrays_valid,
"rays total": nrays_total,
}
)
csvfile.close()
return fname
[docs]
def write_ts_stats(dt, value, fname, stat="mean"):
"""
writes time series of statistics
Parameters
----------
dt : date time array
array of time steps
value: float arrays
the average and the standard deviation of the melting layer top height
fname : str
file name where to store the data
stat : str
Statistic that is written
Returns
-------
fname : str
the name of the file where data has written
"""
filelist = glob.glob(fname)
if not filelist:
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Statistics\n"
+ '# Comment lines are preceded by "#"\n'
+ "# Description: \n"
+ "# Time series of "
+ stat
+ ".\n"
+ "# Fill Value: "
+ str(get_fillvalue())
+ "\n"
+ "# Start: "
+ datetime.datetime.strftime("%Y-%m-%d %H:%M:%S UTC")
+ "\n"
+ "#\n"
)
fieldnames = ["date-time [UTC]", "value"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
writer.writerow(
{
"date-time [UTC]": datetime.datetime.strftime("%Y-%m-%d %H:%M:%S"),
"value": value.filled(fill_value=get_fillvalue())[0],
}
)
csvfile.close()
else:
with open(fname, "a", newline="") as csvfile:
fieldnames = ["date-time [UTC]", "value"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writerow(
{
"date-time [UTC]": datetime.datetime.strftime("%Y-%m-%d %H:%M:%S"),
"value": value.filled(fill_value=get_fillvalue())[0],
}
)
csvfile.close()
return fname
[docs]
def write_ts_cum(dataset, fname):
"""
writes time series accumulation of data
Parameters
----------
dataset : dict
dictionary containing the time series parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
radar_value = dataset["radar_value"].filled(fill_value=get_fillvalue())
sensor_value = dataset["sensor_value"].filled(fill_value=get_fillvalue())
np_radar = dataset["np_radar"]
np_sensor = dataset["np_sensor"]
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Precipitation accumulation data file\n")
csvfile.write('# Comment lines are preceded by "#"\n')
csvfile.write("# Description: \n")
csvfile.write(
"# Time series of a precipitation accumulation of "
+ "weather radar data and another sensor over a "
+ "fixed location.\n"
)
csvfile.write(
"# Location [lon, lat, alt]: "
+ str(dataset["point_coordinates_WGS84_lon_lat_alt"])
+ "\n"
)
if "antenna_coordinates_az_el_r" in dataset:
csvfile.write(
"# Nominal antenna coordinates used [az, el, r]: "
+ str(dataset["antenna_coordinates_az_el_r"])
+ "\n"
)
csvfile.write("# sensor type: " + dataset["sensor"] + "\n")
csvfile.write("# sensor ID: " + dataset["sensorid"] + "\n")
csvfile.write(
"# Data: Precipitation accumulation over "
+ str(dataset["cum_time"])
+ " s\n"
)
csvfile.write("# Fill Value: " + str(get_fillvalue()) + "\n")
csvfile.write(
"# Start: " + dataset["time"][0].strftime("%Y-%m-%d %H:%M:%S UTC") + "\n"
)
csvfile.write("#\n")
fieldnames = ["date", "np_radar", "radar_value", "np_sensor", "sensor_value"]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, rad_val in enumerate(radar_value):
writer.writerow(
{
"date": dataset["time"][i],
"np_radar": np_radar[i],
"radar_value": rad_val,
"np_sensor": np_sensor[i],
"sensor_value": sensor_value[i],
}
)
csvfile.close()
return fname
[docs]
def write_monitoring_ts(
start_time, np_t, values, quantiles, mean_list, datatype, fname, rewrite=False
):
nvalues = np.size(start_time)
start_time_aux = (
np.asarray([start_time]) if nvalues == 1 else np.asarray(start_time)
)
np_t_aux = np.asarray([np_t]) if nvalues == 1 else np.asarray(np_t)
values_aux = (
np.asarray([values.filled(get_fillvalue())])
if nvalues == 1
else values.filled(get_fillvalue())
)
mean0_aux = np.asarray([mean_list[0]]) if nvalues == 1 else np.asarray(mean_list[0])
mean1_aux = np.asarray([mean_list[1]]) if nvalues == 1 else np.asarray(mean_list[1])
data = {
"date": [t.strftime("%Y%m%d%H%M%S") for t in start_time_aux],
"NP": np_t_aux,
"geometric_mean_dB": mean0_aux,
"linear_mean_dB": mean1_aux,
}
for i, q in enumerate(quantiles):
data[f"quantile_{q}"] = values_aux[:, i]
df = pd.DataFrame(data)
# Clean new rows before writing
df["date"] = df["date"].astype(str).str.strip().str.replace(r"\.0$", "", regex=True)
df["date_dt"] = pd.to_datetime(
df["date"],
format="%Y%m%d%H%M%S",
utc=True,
errors="coerce",
)
df["NP"] = pd.to_numeric(df["NP"], errors="coerce")
# Drop rows like: 0.5,,,,,,,
df = df.dropna(subset=["date_dt", "NP"])
df["NP"] = df["NP"].astype(int)
df["date"] = df["date_dt"].dt.strftime("%Y%m%d%H%M%S")
df = df.drop(columns="date_dt")
numeric_cols = [c for c in df.columns if c != "date"]
df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors="coerce")
file_exists = os.path.exists(fname)
with open(fname, "r+" if file_exists and not rewrite else "w") as f:
lock_file(f)
if rewrite or not file_exists:
header = (
"# Weather radar monitoring timeseries data file\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of a monitoring of weather radar data.\n"
f"# Quantiles: {', '.join(map(str, quantiles))} percent.\n"
f"# Data: {generate_field_name_str(datatype)}\n"
f"# Fill Value: {get_fillvalue()}\n"
f"# Start: {start_time_aux[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
"#\n"
)
f.write(header)
df.to_csv(f, index=False)
else:
header = [line.strip() for line in f if line.startswith("#")]
f.seek(0)
existing_df = pd.read_csv(f, comment="#")
# Clean existing file too, in case it already contains bad rows
existing_df["date"] = (
existing_df["date"]
.astype(str)
.str.strip()
.str.replace(r"\.0$", "", regex=True)
)
existing_df["date_dt"] = pd.to_datetime(
existing_df["date"],
format="%Y%m%d%H%M%S",
utc=True,
errors="coerce",
)
existing_df["NP"] = pd.to_numeric(existing_df["NP"], errors="coerce")
existing_df = existing_df.dropna(subset=["date_dt", "NP"])
existing_df["NP"] = existing_df["NP"].astype(int)
existing_df["date"] = existing_df["date_dt"].dt.strftime("%Y%m%d%H%M%S")
existing_df = existing_df.drop(columns="date_dt")
combined_df = pd.concat([existing_df, df], ignore_index=True)
combined_df = combined_df.drop_duplicates(subset=["date"], keep="last")
f.seek(0)
f.truncate()
f.write("\n".join(header) + "\n")
combined_df.to_csv(f, index=False)
unlock_file(f)
return fname
def write_sensor_scores(
start_time,
scores,
double_conditional_threshold,
datatype,
fname,
rewrite=False,
):
"""
Write sensor intercomparison scores to a CSV-with-comment-header file.
Legacy (mono-radar) mode
------------------------
`scores` is a dict keyed by bounds, e.g.:
scores = {"all": {"RMSE": [...], "N": [...]}}
Multi-radar mode
----------------
`scores` is a dict keyed by radar IDs (RADAR001/radar001/...), each mapping to
a mono-radar `scores` dict keyed by bounds, e.g.:
scores = {
"RADAR001": {"all": {"RMSE": [...], "N": [...]}},
"RADAR002": {"all": {"RMSE": [...], "N": [...]}},
}
In multi-radar mode, output columns are suffixed with _radar###, e.g.:
RMSE_radar001, N_radar001, RMSE_radar002, ...
Duplicates are removed by "date" (keep last).
"""
# ----------------------------
# helpers
# ----------------------------
radar_key_re = re.compile(r"^radar\d{3}$", re.IGNORECASE)
def _is_multiradar_scores(sc):
if not isinstance(sc, dict) or not sc:
return False
keys = [k for k in sc.keys() if isinstance(k, str)]
return any(radar_key_re.match(k) for k in keys)
def _normalize_radarkey(k):
# "RADAR001" -> "radar001"
return k.lower()
def _as_1d_array(x, n):
# Accept scalar or list/array; return array length n (repeat scalars)
arr = np.asarray(x)
if arr.size == 1 and n > 1:
arr = np.repeat(arr, n)
return arr
# ----------------------------
# time vector
# ----------------------------
nvalues = np.size(start_time)
start_time_aux = (
np.asarray([start_time]) if nvalues == 1 else np.asarray(start_time)
)
date_str = [t.strftime("%Y%m%d%H%M%S") for t in start_time_aux]
n = len(date_str)
# ----------------------------
# build dataframe
# ----------------------------
data = {"date": date_str}
multiradar = _is_multiradar_scores(scores)
if not multiradar:
# ---- legacy: scores[bounds][metric] ----
bounds = list(scores.keys())[0]
metric_dict = scores[bounds]
for k, v in metric_dict.items():
data[k] = _as_1d_array(v, n)
df = pd.DataFrame(data)
score_names = list(metric_dict.keys())
bounds_line = bounds
radar_list_for_header = None
else:
# ---- multi-radar: scores[radar][bounds][metric] ----
radar_keys = sorted(
[k for k in scores.keys() if isinstance(k, str) and radar_key_re.match(k)]
)
if not radar_keys:
raise ValueError("Multi-radar scores detected but no RADAR### keys found")
# assume same bounds across radars; take from first
bounds = list(scores[radar_keys[0]].keys())[0]
# collect all metrics across radars (union)
metrics_all = set()
for rk in radar_keys:
metrics_all.update(scores[rk][bounds].keys())
metrics_all = sorted(metrics_all)
# fill columns metric_radar###
for rk in radar_keys:
rk_norm = _normalize_radarkey(rk) # radar001
md = scores[rk].get(bounds, {})
for met in metrics_all:
col = f"{met}_{rk_norm}"
if met in md:
data[col] = _as_1d_array(md[met], n)
else:
data[col] = np.full(n, np.nan)
df = pd.DataFrame(data)
score_names = metrics_all
bounds_line = bounds
radar_list_for_header = [rk.upper() for rk in radar_keys]
# enforce date as string for stable dedup/sort
df["date"] = df["date"].astype(str)
# ----------------------------
# write / merge-dedup
# ----------------------------
file_exists = os.path.exists(fname)
# Build header text (written only when creating or rewrite)
header = (
"# Weather radar validation scores with sensors data file\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of sensor intercomparison scores.\n"
f"# Scores: {', '.join(map(str, score_names))}.\n"
f"# Bounds: {bounds_line}.\n"
+ (
f"# Radars: {', '.join(radar_list_for_header)}.\n"
if radar_list_for_header
else ""
)
+ f"# Double conditional threshold: {double_conditional_threshold}.\n"
+ f"# Data: {generate_field_name_str(datatype)}\n"
+ f"# Fill Value: {get_fillvalue()}\n"
+ f"# Start: {start_time_aux[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
+ "#\n"
)
with open(
fname,
"r+" if file_exists and not rewrite else "w",
encoding="utf-8",
newline="",
) as f:
lock_file(f)
try:
if rewrite or not file_exists:
f.seek(0)
f.truncate(0)
f.write(header)
df.to_csv(f, index=False)
else:
# Read existing header lines
f.seek(0)
old_header_lines = [
line.rstrip("\n") for line in f if line.startswith("#")
]
# Read existing data
f.seek(0)
existing_df = pd.read_csv(f, comment="#")
if "date" in existing_df.columns:
existing_df["date"] = existing_df["date"].astype(str)
# Align columns: union to avoid dropping new radar columns
combined_df = pd.concat(
[existing_df, df], ignore_index=True, sort=False
)
# Dedup by date (keep last written)
combined_df.drop_duplicates(subset=["date"], keep="last", inplace=True)
combined_df.sort_values(by="date", inplace=True)
# Rewrite file with preserved header (or new header if missing)
f.seek(0)
f.truncate(0)
if old_header_lines:
f.write("\n".join(old_header_lines) + "\n")
else:
f.write(header)
combined_df.to_csv(f, index=False)
finally:
unlock_file(f)
return fname
[docs]
def write_excess_gates(excess_dict, fname):
"""
Writes the position and values of gates that have a frequency of
occurrence higher than a particular threshold
Parameters
----------
excess_dict : dict
dictionary containing the gates parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
ngates = len(excess_dict["ray_ind"])
with open(fname, "w", newline="") as csvfile:
csvfile.write(
"# Gates exceeding "
+ str(excess_dict["quant_min"])
+ " percentile data file\n"
)
csvfile.write('# Comment lines are preceded by "#"\n')
csvfile.write(
"# Data collection start time: "
+ excess_dict["starttime"].strftime("%Y-%m-%d %H:%M:%S UTC")
+ "\n"
)
csvfile.write(
"# Data collection end time: "
+ excess_dict["endtime"].strftime("%Y-%m-%d %H:%M:%S UTC")
+ "\n"
)
csvfile.write("# Number of gates in file: " + str(ngates) + "\n")
csvfile.write("#\n")
fieldnames = [
"ray_ind",
"rng_ind",
"ele",
"azi",
"rng",
"nsamples",
"occurrence",
"freq_occu",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, ray_ind in enumerate(excess_dict["ray_ind"]):
writer.writerow(
{
"ray_ind": ray_ind,
"rng_ind": excess_dict["rng_ind"][i],
"ele": excess_dict["ele"][i],
"azi": excess_dict["azi"][i],
"rng": excess_dict["rng"][i],
"nsamples": excess_dict["nsamples"][i],
"occurrence": excess_dict["occurrence"][i],
"freq_occu": excess_dict["freq_occu"][i],
}
)
csvfile.close()
return fname
[docs]
def write_intercomp_scores_ts(
start_time,
stats,
field_name,
fname,
rad1_name="RADAR001",
rad2_name="RADAR002",
rewrite=False,
):
"""
Writes time series of radar intercomparison scores.
If an existing file already contains a given date, the newly provided
value replaces the old one.
"""
fieldnames = [
"date",
"NP",
"mean_bias",
"median_bias",
"quant25_bias",
"quant75_bias",
"mode_bias",
"corr",
"slope_of_linear_regression",
"intercep_of_linear_regression",
"intercep_of_linear_regression_of_slope_1",
]
nvalues = np.size(start_time)
def _as_array(value):
value = value.filled(fill_value=get_fillvalue())
if nvalues == 1:
return np.asarray([value])
return np.asarray(value)
if nvalues == 1:
start_time_aux = np.asarray([start_time])
np_t = np.asarray([stats["npoints"]])
else:
start_time_aux = np.asarray(start_time)
np_t = np.asarray(stats["npoints"])
new_df = pd.DataFrame(
{
"date": pd.to_datetime(start_time_aux),
"NP": np_t,
"mean_bias": _as_array(stats["meanbias"]),
"median_bias": _as_array(stats["medianbias"]),
"quant25_bias": _as_array(stats["quant25bias"]),
"quant75_bias": _as_array(stats["quant75bias"]),
"mode_bias": _as_array(stats["modebias"]),
"corr": _as_array(stats["corr"]),
"slope_of_linear_regression": _as_array(stats["slope"]),
"intercep_of_linear_regression": _as_array(stats["intercep"]),
"intercep_of_linear_regression_of_slope_1": _as_array(
stats["intercep_slope_1"]
),
}
)
file_exists = bool(glob.glob(fname)) and not rewrite
if file_exists:
old_df = read_intercomp_scores_ts(fname, return_dataframe=True)
if len(old_df):
old_df["date"] = pd.to_datetime(old_df["date"], errors="coerce")
new_df["date"] = pd.to_datetime(new_df["date"], errors="coerce")
out_df = pd.concat([old_df, new_df], ignore_index=True)
else:
out_df = new_df
else:
out_df = new_df
# Keep only the newest value for each date
out_df = out_df.drop_duplicates(subset="date", keep="last")
# Optional: sort chronologically
out_df = out_df.sort_values("date")
# Convert back to original CSV date format before writing
out_df["date"] = out_df["date"].dt.strftime("%Y%m%d%H%M%S")
with open(fname, "w", newline="") as csvfile:
lock_file(csvfile)
csvfile.write(
"# Weather radar intercomparison scores timeseries file\n"
'# Comment lines are preceded by "#"\n'
"# Description: \n"
"# Time series of the intercomparison between two radars.\n"
f"# Radar 1: {rad1_name}\n"
f"# Radar 2: {rad2_name}\n"
f"# Field name: {field_name}\n"
f"# Fill Value: {get_fillvalue()}\n"
f"# Start: {start_time_aux[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
"#\n"
)
out_df.to_csv(csvfile, index=False, columns=fieldnames)
unlock_file(csvfile)
return fname
[docs]
def write_colocated_gates(coloc_gates, fname):
"""
Writes the position of gates colocated with two radars
Parameters
----------
coloc_gates : dict
dictionary containing the colocated gates parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Colocated radar gates data file\n")
csvfile.write('# Comment lines are preceded by "#"\n')
csvfile.write("#\n")
fieldnames = [
"rad1_ray_ind",
"rad1_rng_ind",
"rad1_ele",
"rad1_azi",
"rad1_rng",
"rad2_ray_ind",
"rad2_rng_ind",
"rad2_ele",
"rad2_azi",
"rad2_rng",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, rad1_ray_ind in enumerate(coloc_gates["rad1_ray_ind"]):
writer.writerow(
{
"rad1_ray_ind": rad1_ray_ind,
"rad1_rng_ind": coloc_gates["rad1_rng_ind"][i],
"rad1_ele": coloc_gates["rad1_ele"][i],
"rad1_azi": coloc_gates["rad1_azi"][i],
"rad1_rng": coloc_gates["rad1_rng"][i],
"rad2_ray_ind": coloc_gates["rad2_ray_ind"][i],
"rad2_rng_ind": coloc_gates["rad2_rng_ind"][i],
"rad2_ele": coloc_gates["rad2_ele"][i],
"rad2_azi": coloc_gates["rad2_azi"][i],
"rad2_rng": coloc_gates["rad2_rng"][i],
}
)
csvfile.close()
return fname
[docs]
def write_colocated_data(coloc_data, fname):
"""
Writes the time averaged data of gates colocated with two radars, if QC metadata
is available, it will be written as well. If the file already exists new records will be
appended if they are not already present (based on rad1_time and rad2_time columns)
Parameters
----------
coloc_data : dict
dictionary containing the colocated data parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
file_exists = bool(glob.glob(fname))
dataset_dict = {
"rad1_time": pd.to_datetime(coloc_data["rad1_time"]).strftime("%Y%m%d%H%M%S"),
"rad1_ray_ind": coloc_data["rad1_ray_ind"],
"rad1_rng_ind": coloc_data["rad1_rng_ind"],
"rad1_ele": coloc_data["rad1_ele"],
"rad1_azi": coloc_data["rad1_azi"],
"rad1_rng": coloc_data["rad1_rng"],
"rad1_val": coloc_data.get("rad1_val"),
"rad2_time": pd.to_datetime(coloc_data["rad2_time"]).strftime("%Y%m%d%H%M%S"),
"rad2_ray_ind": coloc_data["rad2_ray_ind"],
"rad2_rng_ind": coloc_data["rad2_rng_ind"],
"rad2_ele": coloc_data["rad2_ele"],
"rad2_azi": coloc_data["rad2_azi"],
"rad2_rng": coloc_data["rad2_rng"],
"rad2_val": coloc_data["rad2_val"],
}
if "rad1_PhiDPavg" in coloc_data:
dataset_dict["rad1_PhiDPavg"] = coloc_data["rad1_PhiDPavg"]
dataset_dict["rad1_Flagavg"] = coloc_data["rad1_Flagavg"]
dataset_dict["rad2_PhiDPavg"] = coloc_data["rad2_PhiDPavg"]
dataset_dict["rad2_Flagavg"] = coloc_data["rad2_Flagavg"]
df = pd.DataFrame(dataset_dict)
# ---------------------------
# REMOVE DUPLICATES ON APPEND
# ---------------------------
if file_exists:
# Read only needed columns (fast)
df_existing = pd.read_csv(
fname, comment="#", usecols=["rad1_time", "rad2_time"]
)
# Build set of existing pairs
existing_pairs = set(zip(df_existing.rad1_time, df_existing.rad2_time))
# Filter new data
mask = [
(r1, r2) not in existing_pairs for r1, r2 in zip(df.rad1_time, df.rad2_time)
]
df = df[mask]
# If nothing new → exit early
if df.empty:
return fname
# ---------------------------
# WRITE FILE
# ---------------------------
if not file_exists:
with open(fname, "w") as f:
f.write("# Colocated radar gates data file\n")
f.write('# Comment lines are preceded by "#"\n')
f.write("#\n")
df.to_csv(fname, mode="a", index=False)
else:
df.to_csv(fname, mode="a", index=False, header=False)
return fname
[docs]
def write_sun_hits(sun_hits, fname):
"""
Writes sun hits data.
Parameters
----------
sun_hits : dict
dictionary containing the sun hits parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
dBm_sun_hit = sun_hits["dBm_sun_hit"].filled(fill_value=get_fillvalue())
std_dBm_sun_hit = sun_hits["std(dBm_sun_hit)"].filled(fill_value=get_fillvalue())
dBmv_sun_hit = sun_hits["dBmv_sun_hit"].filled(fill_value=get_fillvalue())
std_dBmv_sun_hit = sun_hits["std(dBmv_sun_hit)"].filled(fill_value=get_fillvalue())
zdr_sun_hit = sun_hits["ZDR_sun_hit"].filled(fill_value=get_fillvalue())
std_zdr_sun_hit = sun_hits["std(ZDR_sun_hit)"].filled(fill_value=get_fillvalue())
filelist = glob.glob(fname)
if not filelist:
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Weather radar sun hits data file\n")
csvfile.write('# Comment lines are preceded by "#"\n')
csvfile.write("# Fill Value: " + str(get_fillvalue()) + "\n")
csvfile.write("#\n")
fieldnames = [
"time",
"ray",
"NPrng",
"rad_el",
"rad_az",
"sun_el",
"sun_az",
"dBm_sun_hit",
"std(dBm_sun_hit)",
"NPh",
"NPhval",
"dBmv_sun_hit",
"std(dBmv_sun_hit)",
"NPv",
"NPvval",
"ZDR_sun_hit",
"std(ZDR_sun_hit)",
"NPzdr",
"NPzdrval",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
for i, sh_time in enumerate(sun_hits["time"]):
writer.writerow(
{
"time": sh_time.strftime("%Y-%m-%d %H:%M:%S.%f"),
"ray": sun_hits["ray"][i],
"NPrng": sun_hits["NPrng"][i],
"rad_el": sun_hits["rad_el"][i],
"rad_az": sun_hits["rad_az"][i],
"sun_el": sun_hits["sun_el"][i],
"sun_az": sun_hits["sun_az"][i],
"dBm_sun_hit": dBm_sun_hit[i],
"std(dBm_sun_hit)": std_dBm_sun_hit[i],
"NPh": sun_hits["NPh"][i],
"NPhval": sun_hits["NPhval"][i],
"dBmv_sun_hit": dBmv_sun_hit[i],
"std(dBmv_sun_hit)": std_dBmv_sun_hit[i],
"NPv": sun_hits["NPv"][i],
"NPvval": sun_hits["NPvval"][i],
"ZDR_sun_hit": zdr_sun_hit[i],
"std(ZDR_sun_hit)": std_zdr_sun_hit[i],
"NPzdr": sun_hits["NPzdr"][i],
"NPzdrval": sun_hits["NPzdrval"][i],
}
)
csvfile.close()
else:
with open(fname, "a", newline="") as csvfile:
fieldnames = [
"time",
"ray",
"NPrng",
"rad_el",
"rad_az",
"sun_el",
"sun_az",
"dBm_sun_hit",
"std(dBm_sun_hit)",
"NPh",
"NPhval",
"dBmv_sun_hit",
"std(dBmv_sun_hit)",
"NPv",
"NPvval",
"ZDR_sun_hit",
"std(ZDR_sun_hit)",
"NPzdr",
"NPzdrval",
]
writer = csv.DictWriter(csvfile, fieldnames)
for i, sh_time in enumerate(sun_hits["time"]):
writer.writerow(
{
"time": sh_time.strftime("%Y-%m-%d %H:%M:%S.%f"),
"ray": sun_hits["ray"][i],
"NPrng": sun_hits["NPrng"][i],
"rad_el": sun_hits["rad_el"][i],
"rad_az": sun_hits["rad_az"][i],
"sun_el": sun_hits["sun_el"][i],
"sun_az": sun_hits["sun_az"][i],
"dBm_sun_hit": dBm_sun_hit[i],
"std(dBm_sun_hit)": std_dBm_sun_hit[i],
"NPh": sun_hits["NPh"][i],
"NPhval": sun_hits["NPhval"][i],
"dBmv_sun_hit": dBmv_sun_hit[i],
"std(dBmv_sun_hit)": std_dBmv_sun_hit[i],
"NPv": sun_hits["NPv"][i],
"NPvval": sun_hits["NPvval"][i],
"ZDR_sun_hit": zdr_sun_hit[i],
"std(ZDR_sun_hit)": std_zdr_sun_hit[i],
"NPzdr": sun_hits["NPzdr"][i],
"NPzdrval": sun_hits["NPzdrval"][i],
}
)
csvfile.close()
return fname
[docs]
def write_sun_retrieval(sun_retrieval, fname):
"""
Writes sun retrieval data.
Parameters
----------
sun_retrieval : dict
dictionary containing the sun retrieval parameters
fname : str
file name where to store the data
Returns
-------
fname : str
the name of the file where data has written
"""
first_hit_time = sun_retrieval["first_hit_time"].strftime("%Y%m%d%H%M%S")
last_hit_time = sun_retrieval["last_hit_time"].strftime("%Y%m%d%H%M%S")
el_width_h = sun_retrieval["el_width_h"].filled(fill_value=get_fillvalue())
az_width_h = sun_retrieval["az_width_h"].filled(fill_value=get_fillvalue())
el_bias_h = sun_retrieval["el_bias_h"].filled(fill_value=get_fillvalue())
az_bias_h = sun_retrieval["az_bias_h"].filled(fill_value=get_fillvalue())
dBm_sun_est = sun_retrieval["dBm_sun_est"].filled(fill_value=get_fillvalue())
std_dBm_sun_est = sun_retrieval["std(dBm_sun_est)"].filled(
fill_value=get_fillvalue()
)
sf_h = sun_retrieval["sf_h"].filled(fill_value=get_fillvalue())
el_width_v = sun_retrieval["el_width_v"].filled(fill_value=get_fillvalue())
az_width_v = sun_retrieval["az_width_v"].filled(fill_value=get_fillvalue())
el_bias_v = sun_retrieval["el_bias_v"].filled(fill_value=get_fillvalue())
az_bias_v = sun_retrieval["az_bias_v"].filled(fill_value=get_fillvalue())
dBmv_sun_est = sun_retrieval["dBmv_sun_est"].filled(fill_value=get_fillvalue())
std_dBmv_sun_est = sun_retrieval["std(dBmv_sun_est)"].filled(
fill_value=get_fillvalue()
)
sf_v = sun_retrieval["sf_v"].filled(fill_value=get_fillvalue())
zdr_sun_est = sun_retrieval["ZDR_sun_est"].filled(fill_value=get_fillvalue())
std_zdr_sun_est = sun_retrieval["std(ZDR_sun_est)"].filled(
fill_value=get_fillvalue()
)
sf_ref = sun_retrieval["sf_ref"].filled(fill_value=get_fillvalue())
ref_time = "None"
if sun_retrieval["ref_time"] is not None:
ref_time = sun_retrieval["ref_time"].strftime("%Y%m%d%H%M%S")
filelist = glob.glob(fname)
if not filelist:
with open(fname, "w", newline="") as csvfile:
csvfile.write("# Weather radar sun retrievals data file\n")
csvfile.write('# Comment lines are preceded by "#"\n')
csvfile.write("# Fill Value: " + str(get_fillvalue()) + "\n")
csvfile.write("#\n")
fieldnames = [
"first_hit_time",
"last_hit_time",
"nhits_h",
"el_width_h",
"az_width_h",
"el_bias_h",
"az_bias_h",
"dBm_sun_est",
"std(dBm_sun_est)",
"sf_h",
"nhits_v",
"el_width_v",
"az_width_v",
"el_bias_v",
"az_bias_v",
"dBmv_sun_est",
"std(dBmv_sun_est)",
"sf_v",
"nhits_zdr",
"ZDR_sun_est",
"std(ZDR_sun_est)",
"sf_ref",
"ref_time",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writeheader()
writer.writerow(
{
"first_hit_time": first_hit_time,
"last_hit_time": last_hit_time,
"nhits_h": sun_retrieval["nhits_h"],
"el_width_h": el_width_h,
"az_width_h": az_width_h,
"el_bias_h": el_bias_h,
"az_bias_h": az_bias_h,
"dBm_sun_est": dBm_sun_est,
"std(dBm_sun_est)": std_dBm_sun_est,
"sf_h": sf_h,
"nhits_v": sun_retrieval["nhits_v"],
"el_width_v": el_width_v,
"az_width_v": az_width_v,
"el_bias_v": el_bias_v,
"az_bias_v": az_bias_v,
"dBmv_sun_est": dBmv_sun_est,
"std(dBmv_sun_est)": std_dBmv_sun_est,
"sf_v": sf_v,
"nhits_zdr": sun_retrieval["nhits_zdr"],
"ZDR_sun_est": zdr_sun_est,
"std(ZDR_sun_est)": std_zdr_sun_est,
"sf_ref": sf_ref,
"ref_time": ref_time,
}
)
csvfile.close()
else:
with open(fname, "a", newline="") as csvfile:
fieldnames = [
"first_hit_time",
"last_hit_time",
"nhits_h",
"el_width_h",
"az_width_h",
"el_bias_h",
"az_bias_h",
"dBm_sun_est",
"std(dBm_sun_est)",
"sf_h",
"nhits_v",
"el_width_v",
"az_width_v",
"el_bias_v",
"az_bias_v",
"dBmv_sun_est",
"std(dBmv_sun_est)",
"sf_v",
"nhits_zdr",
"ZDR_sun_est",
"std(ZDR_sun_est)",
"sf_ref",
"ref_time",
]
writer = csv.DictWriter(csvfile, fieldnames)
writer.writerow(
{
"first_hit_time": first_hit_time,
"last_hit_time": last_hit_time,
"nhits_h": sun_retrieval["nhits_h"],
"el_width_h": el_width_h,
"az_width_h": az_width_h,
"el_bias_h": el_bias_h,
"az_bias_h": az_bias_h,
"dBm_sun_est": dBm_sun_est,
"std(dBm_sun_est)": std_dBm_sun_est,
"sf_h": sf_h,
"nhits_v": sun_retrieval["nhits_v"],
"el_width_v": el_width_v,
"az_width_v": az_width_v,
"el_bias_v": el_bias_v,
"az_bias_v": az_bias_v,
"dBmv_sun_est": dBmv_sun_est,
"std(dBmv_sun_est)": std_dBmv_sun_est,
"sf_v": sf_v,
"nhits_zdr": sun_retrieval["nhits_zdr"],
"ZDR_sun_est": zdr_sun_est,
"std(ZDR_sun_est)": std_zdr_sun_est,
"sf_ref": sf_ref,
"ref_time": ref_time,
}
)
csvfile.close()
return fname