Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"skyCvr",
"skyCovLayerBase",
"autoRemark",
"operatorRemark",
]:
data[vname] = nc.variables[vname][:]
for qc in ["QCR", "QCD"]:
vname2 = vname + qc
if vname2 in nc.variables:
data[vname2] = nc.variables[vname2][:]
for vname in ["temperature", "dewpoint"]:
data[vname + "C"] = temperature(data[vname], "K").value("C")
data[vname] = temperature(data[vname], "K").value("F")
for vname in ["windSpeed", "windGust"]:
data[vname] = (
masked_array(data[vname], units("meter / second"))
.to(units("knots"))
.magnitude
)
data["altimeter"] = pressure(data["altimeter"], "PA").value("IN")
data["skyCovLayerBase"] = distance(data["skyCovLayerBase"], "M").value(
"FT"
)
data["visibility"] = distance(data["visibility"], "M").value("MI")
data["precipAccum"] = distance(data["precipAccum"], "MM").value("IN")
stations = chartostring(data["stationId"][:])
presentwxs = chartostring(data["presWx"][:])
skycs = chartostring(data["skyCvr"][:])
autoremarks = chartostring(data["autoRemark"][:])
opremarks = chartostring(data["operatorRemark"][:])
def make_wind_plot(ctx, ptype):
"""Generate a wind plot, please."""
df = ctx["df"]
(fig, ax) = plt.subplots(1, 1)
gust = df["gust_sknt"].values
sknt = df["sknt"].values
unit = "kt"
if ptype == "wind":
gust = masked_array(gust, units("knots")).to(units("miles per hour")).m
sknt = masked_array(sknt, units("knots")).to(units("miles per hour")).m
unit = "mph"
ax.bar(
df["local_valid"].values,
gust,
zorder=1,
width=1.0 / 1440.0,
label="Gust",
)
ax.bar(
df["local_valid"].values,
sknt,
zorder=2,
width=1.0 / 1440.0,
label="Speed",
)
ax.set_ylabel(f"Wind Speed / Gust [{unit}]")
score = chain.score(*test)
shape = (len(YAXIS), len(XAXIS))
grid = chain.grid(
region=region,
shape=shape,
projection=projection,
dims=["latitude", "longitude"],
data_names=["precip"],
)
res = grid.to_array()
res = np.ma.where(res < 0, 0, res)
print(
("%s %s rows for %s column min:%.3f max:%.3f score: %.3f")
% (day, len(df.index), idx, np.nanmin(res), np.nanmax(res), score)
)
return masked_array(res, mpunits("inch"))
write_grid(ts, "uwnd", ures)
write_grid(ts, "vwnd", vres)
tmpf = generic_gridder(df, "max_tmpf", domain)
LOG.debug("grid tmpf is done")
if tmpf is None:
LOG.info("Failure for tmpk at %s", ts)
else:
dwpf = generic_gridder(df, "max_dwpf", domain)
LOG.debug("grid dwpf is done")
# require that dwpk <= tmpk
mask = ~np.isnan(dwpf)
mask[mask] &= dwpf[mask] > tmpf[mask]
dwpf = np.where(mask, tmpf, dwpf)
write_grid(
ts, "tmpk", masked_array(tmpf, data_units="degF").to("degK")
)
write_grid(
ts, "dwpk", masked_array(dwpf, data_units="degF").to("degK")
)
res = grid_skyc(df, domain)
LOG.debug("grid skyc is done")
if res is None:
LOG.info("Failure for skyc at %s", ts)
else:
write_grid(ts, "skyc", res)
def mm2inch(val):
"""Helper."""
return masked_array(val, units("mm")).to(units("inch")).m
)
return
res = generic_gridder(df, "highdata")
ds["high_tmpk"].values = datatypes.temperature(res, "F").value("K")
res = generic_gridder(df, "lowdata")
ds["low_tmpk"].values = datatypes.temperature(res, "F").value("K")
hres = generic_gridder(df, "highdwpf")
lres = generic_gridder(df, "lowdwpf")
if hres is not None and lres is not None:
ds["avg_dwpk"].values = datatypes.temperature(
(hres + lres) / 2.0, "F"
).value("K")
res = generic_gridder(df, "avgsknt")
if res is not None:
ds["wind_speed"].values = (
masked_array(res, units.knots).to(units.meters / units.second).m
)
LOG.debug(
"wind_speed min: %s max: %s",
np.nanmin(ds["wind_speed"].values),
np.nanmax(ds["wind_speed"].values),
)
res = generic_gridder(df, "minrh")
if res is not None:
ds["min_rh"].values = res
res = generic_gridder(df, "maxrh")
if res is not None:
ds["max_rh"].values = res
.m
)
plot_units = "F"
clevs = np.arange(-30, 120, 5)
clevstride = 2
elif varname in ["range_tmpk", "range_tmpk_12z"]:
vname1 = "high_tmpk%s" % (
"_12z" if varname == "range_tmpk_12z" else "",
)
vname2 = "low_tmpk%s" % (
"_12z" if varname == "range_tmpk_12z" else "",
)
d1 = nc.variables[vname1][idx0, jslice, islice]
d2 = nc.variables[vname2][idx0, jslice, islice]
data = (
masked_array(d1, units("degK")).to(units("degF")).m
- masked_array(d2, units("degK")).to(units("degF")).m
)
plot_units = "F"
clevs = np.arange(0, 61, 5)
clevstride = 2
if np.ma.is_masked(np.max(data)):
raise NoDataFound("Data Unavailable")
x, y = np.meshgrid(lons, lats)
if ptype == "c":
# in the case of contour, use the centroids on the grids
mp.contourf(
x + 0.125,
y + 0.125,
data,
clevs,
lats, lons = grbs[1].latlons()
else:
total += grbs[1]["values"]
grbs.close()
now += interval
if lts is None:
if ts.hour > 1:
LOG.info("found no data for date: %s", ts)
return
lts = lts - datetime.timedelta(minutes=1)
subtitle = "Total between 12:00 AM and %s" % (lts.strftime("%I:%M %p %Z"),)
routes = "ac"
if not realtime:
routes = "a"
total = masked_array(total, units("mm")).to(units("inch")).m
for sector in ["iowa", "midwest", "conus"]:
pqstr = ("plot %s %s00 %s_stage4_1d.png %s_stage4_1d.png png") % (
routes,
ts.strftime("%Y%m%d%H"),
sector,
sector,
)
mp = MapPlot(
sector=sector,
title=("%s NCEP Stage IV Today's Precipitation")
% (ts.strftime("%-d %b %Y"),),
subtitle=subtitle,
)
clevs = [0.01, 0.1, 0.3, 0.5, 1, 1.5, 2, 2.5, 3, 3.5, 4, 5, 6, 8, 10]