Newer
Older
This is the module defining fire objects used for fire tracking
a. Allfires: the class of all fire events
b. Fire: the class of a fire event
c. Cluster: the class of active fire pixel cluster (only for supporting)
d. FirePixel: the class of an active fire pixel
"""
# a. Object - Allfires
class Allfires:
""" Class of allfire events at a particular time step
"""
# initilization
def __init__(self, t):
""" Initiate the object with current time
Parameters
----------
t : tuple, (int,int,int,str)
the year, month, day and 'AM'|'PM'
# self.t = FireTime.t_nb(t,nb='previous') # initialize the object at the previous time step
self.t = t
# Allfires object contains a dict of Fire objects with fireID as the key (will be added after reading active fire data)
# Initiate lists storing fire ids which will changes at the current time step
self.fids_expanded = (
[]
) # a list of ids for fires with expansion at current time step
self.fids_new = (
[]
) # a list of ids for all new fires formed at current time step
self.fids_merged = (
[]
) # a list of ids for fires with merging at current time step
self.fids_invalid = (
[]
) # a list of ids for fires invalidated at current time step
self.heritages = [] # a list of fire heritage relationships (source, target)
self.id_dict = (
[]
) # this list relates the list position of the fire in the allfires object to the fire id
# (list_index, fireid) (list position can be variable when writing out only active fires)
Parameters
----------
ampm : str, 'AM'|'PM'
the ampm option calculated from t
""" List of fire ids
"""
return [i for i, f in self.fires.items()]
""" Total number of fires (active and inactive) at this time step
"""
return len(self.fires)
@property
def fids_active(self):
""" List of active fire ids
"""
return [i for i, f in self.fires.items() if f.isactive]
""" Total number of active fires at this time step
"""
return len(self.fids_active)
@property
def activefires(self):
""" dict of active fires
"""
return {i: f for i, f in self.fires.items() if f.isactive}
""" dict of active fires and sleepers
"""
return {i: f for i, f in self.fires.items() if (f.isactive or f.mayreactivate)}
""" dict of inactive fires not going to be reactivated
"""
return {
i: f for i, f in self.fires.items() if not (f.isactive or f.mayreactivate)
}
""" List of fire ids that is not going to be reactivated
"""
return [i for i, f in self.fires.items() if not (f.isactive or f.mayreactivate)]
""" List of fire ids that may reactivate
"""
return [i for i, f in self.fires.items() if f.mayreactivate]
""" Total number of sleep fires at this time step
"""
# return [self.fires[f].fireID for f in self.fires if self.fires[f].invalid is False]
return [i for i, f in self.fires.items() if f.invalid is False]
""" Total number of valid fires at this time step
"""
return len(self.fids_valid)
@property
def validfires(self):
return {i: f for i, f in self.fires.items() if f.invalid is False}
""" List of fire id which is updated at this time step
"""
fids_updated = list(
set(
self.fids_expanded
+ self.fids_new
+ self.fids_merged
+ self.fids_invalid
)
)
""" List of fire id which is newly formed or expanded
"""
fids_ne = sorted(set(self.fids_expanded + self.fids_new))
return fids_ne
# functions to be run before tracking VIIRS active fire pixels at each time step
def update_t(self, t):
""" Update the time (cday and ampm) for the Allfire object.
Parameters
----------
t : tuple, (int,int,int,str)
the year, month, day and 'AM'|'PM'
def update_t_allfires(self, t):
""" Update the time (t) for each Fire object in the Allfire object.
Parameters
----------
t : tuple, (int,int,int,str)
the year, month, day and 'AM'|'PM'
""" Reset newpixels to [] for each fire.
"""
for i, f in self.fires.items():
def cleanup(self, t):
""" Clean up Allfires obj at each time step
- update t (for allfires and all fires)
- clean up lists to record fire changes
- clean up newpixels for each fire
self.update_t_allfires(t) # update t
# reset the fids used to record changes
self.fids_expanded = (
[]
) # a list of ids for fires with expansion at current time step
self.fids_new = (
[]
) # a list of ids for all new fires formed at current time step
self.fids_merged = (
[]
) # a list of ids for fires with merging at current time step
self.fids_invalid = (
[]
) # a list of ids for fires invalidated at current time step
def newyear_reset(self, regnm):
""" reset fire ids at the start of a new year
"""
newfires[i] = self.fires[fid] # record new fireID and fire object
newfires[i].fireID = i # also update fireID attribute of fire object
fidmapping.append((fid, i))
self.fires = newfires
# lastyearfires = {}
# fidmapping = []
# nfid = 0
# for f in self.activefires:
# ofid = f.fireID
# f.fireID = nfid
# # lastyearfires.append(f)
# lastyearfires[nfid] = f
# fidmapping.append((ofid,nfid))
# nfid += 1
# self.fires = lastyearfires
# clean heritages
self.heritages = []
# save the mapping table
if len(fidmapping) > 0:
FireIO.save_newyearfidmapping(fidmapping, self.t[0], regnm)
# functions to be run after tracking VIIRS active fire pixels at each time step
def record_fids_change(
self, fids_expanded=None, fids_new=None, fids_merged=None, fids_invalid=None
):
""" Update the list of ids for fires with changes at current time step.
Parameters
----------
fids_expanded : list
ids of expanded fires
fids_new : list
ids of new formed fires
fids_merged : list
ids of fires with other fires merging to them
fids_invalid : list
ids of fires invalidated (due to merging with other fires)
self.fids_merged = fids_merged # fires with merging with other fires
self.fids_invalid = (
fids_invalid # fires invalidated due to merging with other fires
)
""" If pixel density of an active fire is too large, assume it's static
try:
for i, f in self.activefires.items():
if (f.pixden > 20) & (f.farea < 20):
# invalidate the fire
f.invalid = True
# add the fire id into the fids_invalid list
self.fids_invalid.append(f.fireID)
print('one static fire invalidated')
except Exception as e:
print(e)
# def updateLCTmax(self):
# ''' update Land cover type (dominant LCT for all fire pixels) for active
# fires that is small in size (<1000 pixels)
# '''
# import FireIO
# for f in self.activefires:
# if (f.n_pixels < 1000):
# # get all LCT for the fire pixels
# vLCT = FireIO.get_LCT(f.locs)
# # extract the LCT with most pixel counts
# LCTmax = max(set(vLCT), key = vLCT.count)
# f.LCTmax = LCTmax
# def updateftypes(self):
# ''' update fire type and dominant LCT for active fires
# '''
# import FireFuncs
# for f in self.activefires:
# if f.n_newpixels > 0:
# f.ftype = FireFuncs.set_ftype(f)
# b. Object - Fire
class Fire:
""" Class of a single fire event at a particular time step
"""
# initilization
def __init__(self, id, t, pixels, sensor="viirs"):
""" Initialize Fire class with active fire pixels locations and t. This
is only called when fire clusters forming a new Fire object.
Parameters
----------
id : int
fire id
t : tuple, (int,int,int,str)
the year, month, day and 'AM'|'PM'
pixels : list (nx5)
latitude, longitude, line, sample, and FRP values of active fire pixels
sensor : str
the remote sensing instrument, 'viirs' | 'modis'; no differentiation
between SNPP and NOAA20
self.mergeid = id # mergeid is the final fire id the current fire being merged; use current fire id at initialization
self.sensor = sensor #
# initialize current time, fire start time, and fire final time
self.t = tlist # current time
self.t_st = tlist
self.t_ed = tlist
# initialize pixels
# fpixels = [FirePixel((p[0],p[1]),(p[2],p[3],p[4]),tlist,id) for p in pixels]
fpixels = pixels
self.pixels = fpixels # all pixels
self.newpixels = fpixels # new detected pixels
# self.actpixels = fpixels # new detected pixels of last active fire detection
# initialize hull using the pixels
locs = [p.loc for p in fpixels] # list of [lat,lon]
# locs_geo = [p.loc_geo for p in fpixels] # list of [lat,lon]
hull = FireVector.cal_hull(locs, sensor) # the hull from all locs
self.hull = hull # note fire.hull is not automatically updated (need explicit calculation if changes occur)
# initialize the exterior pixels (pixels within the inward extbuffer of
# the hull; used for saving time for hull calculation of large fires)
self.extpixels = FireVector.cal_extpixels(fpixels, hull)
# fline of latest active timestep, used for sleeper threshold
self.fline_prior = None
# always set validate at initialization
self.invalid = False
# get and record fm1000 value at ignition (at the time of initilization)
# self.stFM1000 = FireIO.get_stFM1000(hull,locs,tlist)
# LCTmax
# vLCT = FireIO.get_LCT(locs)
# self.LCTmax = max(set(vLCT), key = vLCT.count)
# properties
@property
def cday(self):
@property
def cdoy(self):
return self.cday.timetuple().tm_yday
""" Time difference between first and last active fire detection
"""
duration = FireTime.t_dif(self.t_st, self.t_ed) # + 0.5
""" Time difference between current time and the last active fire detection
"""
from FireConsts import maxoffdays
# invalidated fires are always inactive
if self.invalid:
return False
# otherwise, set to True if no new pixels detected for 5 consecutive days
from FireConsts import limoffdays
# invalidated fires are always inactive
if self.invalid:
return False
# otherwise, set to True if no new pixels detected for 5 consecutive days
""" Fire sleeper status
"""
from FireConsts import maxoffdays, limoffdays
# invalidated fires are always inactive
if self.invalid:
return False
# otherwise, set to True if no new pixels detected for 5 consecutive days
@property
def isignition(self):
if (
len(self.newpixels) == 0
): # in this case t_st = t_ed because t_ed has not been updated
ign = (FireTime.t_dif(self.t_st, self.t_ed) == 0) * 1
mp = MultiPoint([(p[0], p[1]) for p in self.locs])
return len(self.pixels)
@property
def newlocs(self):
""" List of new fire pixels locations (lat,lon)
"""
""" List of new fire pixels locations (lat,lon)
"""
mp = MultiPoint([(p[0], p[1]) for p in self.newlocs])
""" List of new fire pixels attributes
"""
return [
(p.lon, p.lat, p.frp, p.DS, p.DT, p.datetime, p.ampm, p.sat)
for p in self.newpixels
]
return len(self.newpixels)
@property
def extlocs(self):
""" List of exterior fire pixel locations (lat,lon)
"""
mp = MultiPoint([(p[0], p[1]) for p in self.extlocs])
return len(self.extpixels)
@property
def ignlocs(self):
""" List of fire pixel locations (lat,lon) at ignition time step
"""
mp = MultiPoint([(p[0], p[1]) for p in self.ignlocs])
""" List of fire pixel locations (lat,lon) at ignition time step
"""
return [p.loc_geo for p in self.ignpixels]
@property
def ignlocsMP_geo(self):
mp = MultiPoint([(p[0], p[1]) for p in self.ignlocs_geo])
return len(self.ignpixels)
@property
def farea(self):
""" Fire spatail size of the fire event (km2)
"""
# get hull
fhull = self.hull
# If no hull, return area calculated from number of pixels
if fhull is None:
return self.n_pixels * area_VI
# otherwise, use calConcHarea to calculate area,
# but no smaller than area_VI (sometimes calculated hull area is very small)
# from pyproj import Geod
# geod = Geod(ellps="WGS84")
# area_cal = np.abs(geod.geometry_area_perimeter(self.hull)[0]/1e6)
area_cal = fhull.area / 1e6
return max(area_cal, area_VI)
# import FireVector
# return max(FireVector.calConcHarea(fhull),area_VI)
# @property
# def centroid(self):
# ''' Centroid of fire object (lat,lon)
# '''
# import FireClustering
#
# # get hull
# fhull = self.hull
#
# if fhull is not None: # centroid of the hull
# cent = (fhull.centroid.y,fhull.centroid.x)
# else: # when no hull, use the centroid of all pixels
# cent = FireClustering.cal_centroid(self.locs)
# return cent
""" Fire pixel density (number of pixels per km2 fire area)
"""
else:
m = 0
return m
@property
def ftypename(self):
# get hull
fhull = self.hull
if fhull is None: # if no hull, return zero
perim = 0
else: # otherwise, use the hull length
# from pyproj import Geod
# geod = Geod(ellps="WGS84")
# perim = geod.geometry_length(fhull)/1000 # in km
""" List of all fire pixels near the fire perimeter (fine line pixels)
"""
import FireVector
from FireConsts import fpbuffer
else:
try:
# otherwise, extract the pixels nearl the hull
# if hull is a polygon, return new pixels near the hull
if fhull.type == "Polygon":
# lr = fhull.exterior.buffer(fpbuffer)
lr = FireVector.addbuffer(fhull.exterior, fpbuffer)
return [p for p in nps if lr.contains(Point(p.loc[0], p.loc[1]))]
# if hull is a multipolygon, return new pixels near the hull
elif fhull.type == "MultiPolygon":
# mlr = MultiLineString([x.exterior for x in fhull]).buffer(fpbuffer)
mlr = MultiLineString([x.exterior for x in fhull])
mlr = FireVector.addbuffer(mlr, fpbuffer)
return [p for p in nps if mlr.contains(Point(p.loc[0], p.loc[1]))]
except Exception as e:
print(e)
return []
""" List of fire line pixel locations (lat,lon)
"""
return [p.loc for p in self.flinepixels]
@property
def flplocsMP(self):
from shapely.geometry import Point
import geopandas as gpd
return gpd.GeoSeries(mp)
@property
def n_flinepixels(self):
""" Active fire line MultiLineString shape (segment of fire perimeter with active fires nearby)
"""
if (
self.n_flinepixels == 0
): # this happens is last active pixels are within the fire scar
# get fireline pixel locations, different from flplocsMP since it contains a VIIRS pixel buffer
flinelocsMP = FireVector.doMultP(self.flplocs, VIIRSbuf)
# get the hull
fhull = self.hull
# calculate the fire line
return None
else: # otherwise, create shape of the active fire line
# extract exterior of fire perimeter
mls = MultiLineString([plg.exterior for plg in fhull])
# return the part which intersects with bufferred flinelocsMP
# return mls.intersection(flinelocsMP.buffer(flbuffer))
flinelocsMP_buf = FireVector.addbuffer(flinelocsMP, flbuffer)
mls = fhull.exterior
# return mls.intersection(flinelocsMP.buffer(flbuffer))
flinelocsMP_buf = FireVector.addbuffer(flinelocsMP, flbuffer)
else: # if fhull type is not 'MultiPolygon' or 'Polygon', return flinelocsMP
fline = flinelocsMP
# we save the fire line to a new property (this is only updated when fline not None)
self.fline_prior = fline
return fline
# geod = Geod(ellps="WGS84")
# flinelen = geod.geometry_length(self.fline)/1000 # in km
# do not use ftype as property since it may mess up when t updates (without pixel addition)
def updatefhull(self, newlocs):
""" Update the hull using old hull and new locs
"""
hull = FireVector.cal_hull(newlocs, sensor=self.sensor)
# use the union to include hull in past time step
phull = self.hull
self.hull = phull.union(hull)
def updateextpixels(self, newpixels):
""" Update the external pixels
"""
self.extpixels = FireVector.cal_extpixels(pextpixels + newpixels, self.hull)
# c. Object - Cluster
class Cluster:
""" class of active fire pixel cluster at a particular time
"""
# initilization
def __init__(self, id, pixels, t, sensor="viirs"):
""" initilization
Parameters
----------
id : int
cluster id number
pixels : 3-element list
(lat, lon, FRP) of AF pixels
t : tuple, (int,int,int,str)
the year, month, day and 'AM'|'PM'
""" List of pixel locations (lat,lon)
"""
return [(p.x, p.y) for p in self.pixels]
return FireClustering.cal_centroid(self.locs)
@property
def n_pixels(self):
hull = FireVector.cal_hull(self.locs, self.sensor)
@property
def b_box(self):
b_box = self.hull.bounds
return b_box
# d. Object - FirePixel
class FirePixel:
""" class of an acitve fire pixel, which includes
loc : location (lat, lon)
atts : line & sample or viirs pixel, fire radiative power
t : time (y,m,d,ampm) of record
origin : the fire id originally recorded (before merging)
"""
def __init__(self, x, y, lon, lat, frp, DS, DT, ampm, t, Sat, origin):
# self.t = list(t) # (year,month,day,ampm)
self.DS = DS
self.DT = DT
self.datetime = t # YYYYMMDD_HHMM
self.sat = Sat # satellite
self.origin = origin # intially assigned fire id