from PIL import Image
import matplotlib as mpl
from utils import prep_for_plot
import torch.multiprocessing
import torchvision.transforms as T
from utils_gee import extract_img, transform_ee_img
import plotly.graph_objects as go
import plotly.express as px
import numpy as np
from plotly.subplots import make_subplots
import os
os.environ['KMP_DUPLICATE_LIB_OK'] = 'True'
colors = ('red', 'palegreen', 'green', 'steelblue', 'blue', 'yellow', 'lightgrey')
class_names = ('Buildings', 'Cultivation', 'Natural green', 'Wetland', 'Water', 'Infrastructure', 'Background')
cmap = mpl.colors.ListedColormap(colors)
colors = ('red', 'palegreen', 'green', 'steelblue', 'blue', 'yellow', 'lightgrey')
class_names = ('Buildings', 'Cultivation', 'Natural green', 'Wetland', 'Water', 'Infrastructure', 'Background')
scores_init = [1,2,4,3,4,1,0]
# Function that look for img on EE and segment it
# -- 3 ways possible to avoid cloudy environment -- monthly / bi-monthly / yearly meaned img
def segment_loc(model, location, month, year, how = "month", month_end = '12', year_end = None) :
if how == 'month':
img = extract_img(location, year +'-'+ month +'-01', year +'-'+ month +'-28')
elif how == 'year' :
if year_end == None :
img = extract_img(location, year +'-'+ month +'-01', year +'-'+ month_end +'-28', width = 0.04 , len = 0.04)
else :
img = extract_img(location, year +'-'+ month +'-01', year_end +'-'+ month_end +'-28', width = 0.04 , len = 0.04)
img_test= transform_ee_img(img, max = 0.25)
# Preprocess opened img
x = preprocess(img_test)
x = torch.unsqueeze(x, dim=0).cpu()
# model=model.cpu()
with torch.no_grad():
feats, code = model.net(x)
linear_preds = model.linear_probe(x, code)
linear_preds = linear_preds.argmax(1)
outputs = {
'img': x[:model.cfg.n_images].detach().cpu(),
'linear_preds': linear_preds[:model.cfg.n_images].detach().cpu()
}
return outputs
# Function that look for all img on EE and extract all segments with the date as first output arg
def segment_group(location, start_date, end_date, how = 'month') :
outputs = []
st_month = int(start_date[5:7])
end_month = int(end_date[5:7])
st_year = int(start_date[0:4])
end_year = int(end_date[0:4])
for year in range(st_year, end_year+1) :
if year != end_year :
last = 12
else :
last = end_month
if year != st_year:
start = 1
else :
start = st_month
if how == 'month' :
for month in range(start, last + 1):
month_str = f"{month:0>2d}"
year_str = str(year)
outputs.append((year_str + '-' + month_str, segment_loc(location, month_str, year_str)))
elif how == 'year' :
outputs.append((str(year) + '-' + f"{start:0>2d}", segment_loc(location, f"{start:0>2d}", str(year), how = 'year', month_end=f"{last:0>2d}")))
elif how == '2months' :
for month in range(start, last + 1):
month_str = f"{month:0>2d}"
year_str = str(year)
month_end = (month) % 12 +1
if month_end < month :
year_end = year +1
else :
year_end = year
month_end= f"{month_end:0>2d}"
year_end = str(year_end)
outputs.append((year_str + '-' + month_str, segment_loc(location, month_str, year_str,how = 'year', month_end=month_end, year_end=year_end)))
return outputs
def values_from_output(output):
imgs = transform_to_pil(output, alpha = 0.3)
img = imgs[0]
img = np.array(img.convert('RGB'))
labeled_img = imgs[2]
labeled_img = np.array(labeled_img.convert('RGB'))
nb_values = []
for i in range(7):
nb_values.append(np.count_nonzero(output['linear_preds'][0] == i+1))
score = sum(x * y for x, y in zip(scores_init, nb_values)) / sum(nb_values) / max(scores_init)
return img, labeled_img, nb_values, score
# Function that extract from outputs (from segment_group function) all dates/ all images
def values_from_outputs(outputs) :
months = []
imgs = []
imgs_label = []
nb_values = []
scores = []
for output in outputs:
img, labeled_img, nb_value, score = values_from_output(output[1])
months.append(output[0])
imgs.append(img)
imgs_label.append(labeled_img)
nb_values.append(nb_value)
scores.append(score)
return months, imgs, imgs_label, nb_values, scores
def plot_imgs_labels(months, imgs, imgs_label, nb_values, scores) :
fig2 = px.imshow(np.array(imgs), animation_frame=0, binary_string=True)
fig3 = px.imshow(np.array(imgs_label), animation_frame=0, binary_string=True)
# Scores
scatters = []
temp = []
for score in scores :
temp_score = []
temp_date = []
score = scores[i]
temp.append(score)
text_temp = ["" for i in temp]
text_temp[-1] = str(round(score,2))
scatters.append(go.Scatter(x=text_temp, y=temp, mode="lines+markers+text", marker_color="black", text = text_temp, textposition="top center"))
# Scores
fig = make_subplots(
rows=1, cols=4,
# specs=[[{"rowspan": 2}, {"rowspan": 2}, {"type": "pie"}, None]]
# row_heights=[0.8, 0.2],
column_widths = [0.6, 0.6,0.3, 0.3],
subplot_titles=("Localisation visualization", "labeled visualisation", "Segments repartition", "Biodiversity scores")
)
fig.add_trace(fig2["frames"][0]["data"][0], row=1, col=1)
fig.add_trace(fig3["frames"][0]["data"][0], row=1, col=2)
fig.add_trace(go.Pie(labels = class_names,
values = nb_values[0],
marker_colors = colors,
name="Segment repartition",
textposition='inside',
texttemplate = "%{percent:.0%}",
textfont_size=14
),
row=1, col=3)
fig.add_trace(scatters[0], row=1, col=4)
# fig.add_annotation(text='score:' + str(scores[0]),
# showarrow=False,
# row=2, col=2)
number_frames = len(imgs)
frames = [dict(
name = k,
data = [ fig2["frames"][k]["data"][0],
fig3["frames"][k]["data"][0],
go.Pie(labels = class_names,
values = nb_values[k],
marker_colors = colors,
name="Segment repartition",
textposition='inside',
texttemplate = "%{percent:.0%}",
textfont_size=14
),
scatters[k]
],
traces=[0, 1,2,3] # the elements of the list [0,1,2] give info on the traces in fig.data
# that are updated by the above three go.Scatter instances
) for k in range(number_frames)]
updatemenus = [dict(type='buttons',
buttons=[dict(label='Play',
method='animate',
args=[[f'{k}' for k in range(number_frames)],
dict(frame=dict(duration=500, redraw=False),
transition=dict(duration=0),
easing='linear',
fromcurrent=True,
mode='immediate'
)])],
direction= 'left',
pad=dict(r= 10, t=85),
showactive =True, x= 0.1, y= 0.13, xanchor= 'right', yanchor= 'top')
]
sliders = [{'yanchor': 'top',
'xanchor': 'left',
'currentvalue': {'font': {'size': 16}, 'prefix': 'Frame: ', 'visible': False, 'xanchor': 'right'},
'transition': {'duration': 500.0, 'easing': 'linear'},
'pad': {'b': 10, 't': 50},
'len': 0.9, 'x': 0.1, 'y': 0,
'steps': [{'args': [[k], {'frame': {'duration': 500.0, 'easing': 'linear', 'redraw': False},
'transition': {'duration': 0, 'easing': 'linear'}}],
'label': months[k], 'method': 'animate'} for k in range(number_frames)
]}]
fig.update(frames=frames)
for i,fr in enumerate(fig["frames"]):
fr.update(
layout={
"xaxis": {
"range": [0,imgs[0].shape[1]+i/100000]
},
"yaxis": {
"range": [imgs[0].shape[0]+i/100000,0]
},
})
fr.update(layout_title_text= months[i])
fig.update(layout_title_text= 'tot')
fig.update(
layout={
"xaxis": {
"range": [0,imgs[0].shape[1]+i/100000],
'showgrid': False, # thin lines in the background
'zeroline': False, # thick line at x=0
'visible': False, # numbers below
},
"yaxis": {
"range": [imgs[0].shape[0]+i/100000,0],
'showgrid': False, # thin lines in the background
'zeroline': False, # thick line at y=0
'visible': False,},
"xaxis3": {
"range": [0,len(scores)+1],
'autorange': False, # thin lines in the background
'showgrid': False, # thin lines in the background
'zeroline': False, # thick line at y=0
'visible': False
},
"yaxis3": {
"range": [0,1.5],
'autorange': False,
'showgrid': False, # thin lines in the background
'zeroline': False, # thick line at y=0
'visible': False # thin lines in the background
}
},
legend=dict(
yanchor="bottom",
y=0.99,
xanchor="center",
x=0.01
)
)
fig.update_layout(updatemenus=updatemenus,
sliders=sliders)
fig.update_layout(margin=dict(b=0, r=0))
# fig.show() #in jupyter notebook
return fig
# Last function (global one)
# how = 'month' or '2months' or 'year'
def segment_region(location, start_date, end_date, how = 'month'):
#extract the outputs for each image
outputs = segment_group(location, start_date, end_date, how = how)
#extract the intersting values from image
months, imgs, imgs_label, nb_values, scores = values_from_outputs(outputs)
#Create the figure
fig = plot_imgs_labels(months, imgs, imgs_label, nb_values, scores)
return fig
#normalize img
preprocess = T.Compose([
T.ToPILImage(),
T.Resize((320,320)),
# T.CenterCrop(224),
T.ToTensor(),
T.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
# Function that look for img on EE and segment it
# -- 3 ways possible to avoid cloudy environment -- monthly / bi-monthly / yearly meaned img
def segment_loc(model,location, month, year, how = "month", month_end = '12', year_end = None) :
if how == 'month':
img = extract_img(location, year +'-'+ month +'-01', year +'-'+ month +'-28')
elif how == 'year' :
if year_end == None :
img = extract_img(location, year +'-'+ month +'-01', year +'-'+ month_end +'-28', width = 0.04 , len = 0.04)
else :
img = extract_img(location, year +'-'+ month +'-01', year_end +'-'+ month_end +'-28', width = 0.04 , len = 0.04)
img_test= transform_ee_img(img, max = 0.25)
# Preprocess opened img
x = preprocess(img_test)
x = torch.unsqueeze(x, dim=0).cpu()
# model=model.cpu()
with torch.no_grad():
feats, code = model.net(x)
linear_preds = model.linear_probe(x, code)
linear_preds = linear_preds.argmax(1)
outputs = {
'img': x[:model.cfg.n_images].detach().cpu(),
'linear_preds': linear_preds[:model.cfg.n_images].detach().cpu()
}
return outputs
# Function that look for all img on EE and extract all segments with the date as first output arg
def segment_group(location, start_date, end_date, how = 'month') :
outputs = []
st_month = int(start_date[5:7])
end_month = int(end_date[5:7])
st_year = int(start_date[0:4])
end_year = int(end_date[0:4])
for year in range(st_year, end_year+1) :
if year != end_year :
last = 12
else :
last = end_month
if year != st_year:
start = 1
else :
start = st_month
if how == 'month' :
for month in range(start, last + 1):
month_str = f"{month:0>2d}"
year_str = str(year)
outputs.append((year_str + '-' + month_str, segment_loc(location, month_str, year_str)))
elif how == 'year' :
outputs.append((str(year) + '-' + f"{start:0>2d}", segment_loc(location, f"{start:0>2d}", str(year), how = 'year', month_end=f"{last:0>2d}")))
elif how == '2months' :
for month in range(start, last + 1):
month_str = f"{month:0>2d}"
year_str = str(year)
month_end = (month) % 12 +1
if month_end < month :
year_end = year +1
else :
year_end = year
month_end= f"{month_end:0>2d}"
year_end = str(year_end)
outputs.append((year_str + '-' + month_str, segment_loc(location, month_str, year_str,how = 'year', month_end=month_end, year_end=year_end)))
return outputs
# Function that transforms an output to PIL images
def transform_to_pil(outputs,alpha=0.3):
# Transform img with torch
img = torch.moveaxis(prep_for_plot(outputs['img'][0]),-1,0)
img=T.ToPILImage()(img)
# Transform label by saving it then open it
# label = outputs['linear_preds'][0]
# plt.imsave('label.png',label,cmap=cmap)
# label = Image.open('label.png')
cmaplist = np.array([np.array(cmap(i)) for i in range(cmap.N)])
labels = np.array(outputs['linear_preds'][0])-1
label = T.ToPILImage()((cmaplist[labels]*255).astype(np.uint8))
# Overlay labels with img wit alpha
background = img.convert("RGBA")
overlay = label.convert("RGBA")
labeled_img = Image.blend(background, overlay, alpha)
return img, label, labeled_img
def values_from_output(output):
imgs = transform_to_pil(output,alpha = 0.3)
img = imgs[0]
img = np.array(img.convert('RGB'))
labeled_img = imgs[2]
labeled_img = np.array(labeled_img.convert('RGB'))
nb_values = []
for i in range(7):
nb_values.append(np.count_nonzero(output['linear_preds'][0] == i+1))
score = sum(x * y for x, y in zip(scores_init, nb_values)) / sum(nb_values) / max(scores_init)
return img, labeled_img, nb_values, score
# Function that extract labeled_img(PIL) and nb_values(number of pixels for each class) and the score for each observation
# Function that extract from outputs (from segment_group function) all dates/ all images
def values_from_outputs(outputs) :
months = []
imgs = []
imgs_label = []
nb_values = []
scores = []
for output in outputs:
img, labeled_img, nb_value, score = values_from_output(output[1])
months.append(output[0])
imgs.append(img)
imgs_label.append(labeled_img)
nb_values.append(nb_value)
scores.append(score)
return months, imgs, imgs_label, nb_values, scores
# Last function (global one)
# how = 'month' or '2months' or 'year'
def segment_region(latitude, longitude, start_date, end_date, how = 'month'):
location = [float(latitude),float(longitude)]
how = how[0]
#extract the outputs for each image
outputs = segment_group(location, start_date, end_date, how = how)
#extract the intersting values from image
months, imgs, imgs_label, nb_values, scores = values_from_outputs(outputs)
print(months, imgs, imgs_label, nb_values, scores)
#Create the figure
fig = plot_imgs_labels(months, imgs, imgs_label, nb_values, scores)
return fig