Sideband/sbapp/ui/messages.py
2023-10-22 18:08:28 +02:00

699 lines
30 KiB
Python

import time
import RNS
import LXMF
from kivy.metrics import dp,sp
from kivy.core.clipboard import Clipboard
from kivymd.uix.card import MDCard
from kivymd.uix.menu import MDDropdownMenu
# from kivymd.uix.behaviors import RoundedRectangularElevationBehavior, FakeRectangularElevationBehavior
from kivymd.uix.behaviors import CommonElevationBehavior
from kivy.properties import StringProperty, BooleanProperty
from kivy.uix.gridlayout import GridLayout
from kivy.uix.boxlayout import BoxLayout
from kivy.clock import Clock
from kivymd.uix.button import MDRectangleFlatButton, MDRectangleFlatIconButton
from kivymd.uix.dialog import MDDialog
import os
import plyer
import subprocess
import shlex
if RNS.vendor.platformutils.get_platform() == "android":
from sideband.sense import Telemeter
from ui.helpers import ts_format, file_ts_format, mdc
from ui.helpers import color_received, color_delivered, color_propagated, color_paper, color_failed, color_unknown, intensity_msgs_dark, intensity_msgs_light
else:
from sbapp.sideband.sense import Telemeter
from .helpers import ts_format, file_ts_format, mdc
from .helpers import color_received, color_delivered, color_propagated, color_paper, color_failed, color_unknown, intensity_msgs_dark, intensity_msgs_light
from kivy.lang.builder import Builder
class ListLXMessageCard(MDCard):
# class ListLXMessageCard(MDCard, FakeRectangularElevationBehavior):
text = StringProperty()
heading = StringProperty()
class Messages():
def __init__(self, app, context_dest):
self.app = app
self.context_dest = context_dest
self.screen = self.app.root.ids.screen_manager.get_screen("messages_screen")
self.ids = self.screen.ids
self.new_messages = []
self.added_item_hashes = []
self.added_messages = 0
self.latest_message_timestamp = None
self.earliest_message_timestamp = time.time()
self.loading_earlier_messages = False
self.list = None
self.widgets = []
self.send_error_dialog = None
self.load_more_button = None
self.update()
def reload(self):
if self.list != None:
self.list.clear_widgets()
self.new_messages = []
self.added_item_hashes = []
self.added_messages = 0
self.latest_message_timestamp = None
self.widgets = []
self.update()
def load_more(self, dt):
for new_message in self.app.sideband.list_messages(self.context_dest, before=self.earliest_message_timestamp,limit=5):
self.new_messages.append(new_message)
if len(self.new_messages) > 0:
self.loading_earlier_messages = True
self.list.remove_widget(self.load_more_button)
def update(self, limit=8):
for new_message in self.app.sideband.list_messages(self.context_dest, after=self.latest_message_timestamp,limit=limit):
self.new_messages.append(new_message)
self.db_message_count = self.app.sideband.count_messages(self.context_dest)
if self.load_more_button == None:
self.load_more_button = MDRectangleFlatIconButton(
icon="message-text-clock-outline",
text="Load earlier messages",
font_size=dp(18),
theme_text_color="Custom",
size_hint=[1.0, None],
)
def lmcb(sender):
Clock.schedule_once(self.load_more, 0.15)
self.load_more_button.bind(on_release=lmcb)
if self.list == None:
layout = GridLayout(cols=1, spacing=dp(16), padding=dp(16), size_hint_y=None)
layout.bind(minimum_height=layout.setter('height'))
self.list = layout
c_ts = time.time()
if len(self.new_messages) > 0:
self.update_widget()
if (len(self.added_item_hashes) < self.db_message_count) and not self.load_more_button in self.list.children:
self.list.add_widget(self.load_more_button, len(self.list.children))
if self.app.sideband.config["dark_ui"]:
intensity_msgs = intensity_msgs_dark
else:
intensity_msgs = intensity_msgs_light
for w in self.widgets:
m = w.m
if self.app.sideband.config["dark_ui"]:
w.line_color = (1.0, 1.0, 1.0, 0.25)
else:
w.line_color = (1.0, 1.0, 1.0, 0.5)
if m["state"] == LXMF.LXMessage.SENDING or m["state"] == LXMF.LXMessage.OUTBOUND:
msg = self.app.sideband.message(m["hash"])
if msg["state"] == LXMF.LXMessage.DELIVERED:
w.md_bg_color = msg_color = mdc(color_delivered, intensity_msgs)
txstr = time.strftime(ts_format, time.localtime(msg["sent"]))
titlestr = ""
if msg["title"]:
titlestr = "[b]Title[/b] "+msg["title"].decode("utf-8")+"\n"
w.heading = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Delivered"
m["state"] = msg["state"]
if msg["method"] == LXMF.LXMessage.PAPER:
w.md_bg_color = msg_color = mdc(color_paper, intensity_msgs)
txstr = time.strftime(ts_format, time.localtime(msg["sent"]))
titlestr = ""
if msg["title"]:
titlestr = "[b]Title[/b] "+msg["title"].decode("utf-8")+"\n"
w.heading = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Paper Message"
m["state"] = msg["state"]
if msg["method"] == LXMF.LXMessage.PROPAGATED and msg["state"] == LXMF.LXMessage.SENT:
w.md_bg_color = msg_color = mdc(color_propagated, intensity_msgs)
txstr = time.strftime(ts_format, time.localtime(msg["sent"]))
titlestr = ""
if msg["title"]:
titlestr = "[b]Title[/b] "+msg["title"].decode("utf-8")+"\n"
w.heading = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] On Propagation Net"
m["state"] = msg["state"]
if msg["state"] == LXMF.LXMessage.FAILED:
w.md_bg_color = msg_color = mdc(color_failed, intensity_msgs)
txstr = time.strftime(ts_format, time.localtime(msg["sent"]))
titlestr = ""
if msg["title"]:
titlestr = "[b]Title[/b] "+msg["title"].decode("utf-8")+"\n"
w.heading = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Failed"
m["state"] = msg["state"]
w.dmenu.items.append(w.dmenu.retry_item)
def update_widget(self):
if self.app.sideband.config["dark_ui"]:
intensity_msgs = intensity_msgs_dark
mt_color = [1.0, 1.0, 1.0, 0.8]
else:
intensity_msgs = intensity_msgs_light
mt_color = [1.0, 1.0, 1.0, 0.95]
if self.loading_earlier_messages:
self.new_messages.reverse()
for m in self.new_messages:
if not m["hash"] in self.added_item_hashes:
txstr = time.strftime(ts_format, time.localtime(m["sent"]))
rxstr = time.strftime(ts_format, time.localtime(m["received"]))
titlestr = ""
extra_telemetry = {}
phy_stats_str = ""
if "extras" in m and m["extras"] != None:
phy_stats = m["extras"]
if "q" in phy_stats:
try:
lq = round(float(phy_stats["q"]), 1)
phy_stats_str += "[b]Link Quality[/b] "+str(lq)+"% "
extra_telemetry["quality"] = lq
except:
pass
if "rssi" in phy_stats:
try:
lr = round(float(phy_stats["rssi"]), 1)
phy_stats_str += "[b]RSSI[/b] "+str(lr)+"dBm "
extra_telemetry["rssi"] = lr
except:
pass
if "snr" in phy_stats:
try:
ls = round(float(phy_stats["snr"]), 1)
phy_stats_str += "[b]SNR[/b] "+str(ls)+"dB "
extra_telemetry["snr"] = ls
except:
pass
if m["title"]:
titlestr = "[b]Title[/b] "+m["title"].decode("utf-8")+"\n"
if m["source"] == self.app.sideband.lxmf_destination.hash:
if m["state"] == LXMF.LXMessage.DELIVERED:
msg_color = mdc(color_delivered, intensity_msgs)
heading_str = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Delivered"
elif m["method"] == LXMF.LXMessage.PROPAGATED and m["state"] == LXMF.LXMessage.SENT:
msg_color = mdc(color_propagated, intensity_msgs)
heading_str = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] On Propagation Net"
elif m["method"] == LXMF.LXMessage.PAPER:
msg_color = mdc(color_paper, intensity_msgs)
heading_str = titlestr+"[b]Created[/b] "+txstr+"\n[b]State[/b] Paper Message"
elif m["state"] == LXMF.LXMessage.FAILED:
msg_color = mdc(color_failed, intensity_msgs)
heading_str = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Failed"
elif m["state"] == LXMF.LXMessage.OUTBOUND or m["state"] == LXMF.LXMessage.SENDING:
msg_color = mdc(color_unknown, intensity_msgs)
heading_str = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Sending "
else:
msg_color = mdc(color_unknown, intensity_msgs)
heading_str = titlestr+"[b]Sent[/b] "+txstr+"\n[b]State[/b] Unknown"
else:
msg_color = mdc(color_received, intensity_msgs)
heading_str = titlestr
if phy_stats_str != "" and self.app.sideband.config["advanced_stats"]:
heading_str += phy_stats_str+"\n"
heading_str += "[b]Received[/b] "+rxstr+"\n[b]Sent[/b] "+txstr
item = ListLXMessageCard(
text=m["content"].decode("utf-8"),
heading=heading_str,
md_bg_color=msg_color,
)
item.sb_uid = m["hash"]
item.m = m
item.ids.heading_text.theme_text_color = "Custom"
item.ids.heading_text.text_color = mt_color
item.ids.content_text.theme_text_color = "Custom"
item.ids.content_text.text_color = mt_color
item.ids.msg_submenu.theme_text_color = "Custom"
item.ids.msg_submenu.text_color = mt_color
def gen_del(mhash, item):
def x():
yes_button = MDRectangleFlatButton(text="Yes",font_size=dp(18), theme_text_color="Custom", line_color=self.app.color_reject, text_color=self.app.color_reject)
no_button = MDRectangleFlatButton(text="No",font_size=dp(18))
dialog = MDDialog(
title="Delete message?",
buttons=[ yes_button, no_button ],
# elevation=0,
)
def dl_yes(s):
dialog.dismiss()
self.app.sideband.delete_message(mhash)
def cb(dt):
self.reload()
Clock.schedule_once(cb, 0.2)
def dl_no(s):
dialog.dismiss()
yes_button.bind(on_release=dl_yes)
no_button.bind(on_release=dl_no)
item.dmenu.dismiss()
dialog.open()
return x
def gen_retry(mhash, mcontent, item):
def x():
self.app.messages_view.ids.message_text.text = mcontent.decode("utf-8")
self.app.sideband.delete_message(mhash)
self.app.message_send_action()
item.dmenu.dismiss()
def cb(dt):
self.reload()
Clock.schedule_once(cb, 0.2)
return x
def gen_copy(msg, item):
def x():
Clipboard.copy(msg)
item.dmenu.dismiss()
return x
def gen_copy_telemetry(packed_telemetry, extra_telemetry, item):
def x():
try:
telemeter = Telemeter.from_packed(packed_telemetry)
tlm = telemeter.read_all()
if extra_telemetry and len(extra_telemetry) != 0:
tlm["physical_link"] = extra_telemetry
Clipboard.copy(str(tlm))
item.dmenu.dismiss()
except Exception as e:
RNS.log("An error occurred while decoding telemetry. The contained exception was: "+str(e), RNS.LOG_ERROR)
Clipboard.copy("Could not decode telemetry")
return x
def gen_copy_lxm_uri(lxm, item):
def x():
Clipboard.copy(lxm.as_uri())
item.dmenu.dismiss()
return x
def gen_save_qr(lxm, item):
if RNS.vendor.platformutils.is_android():
def x():
qr_image = lxm.as_qr()
hash_str = RNS.hexrep(lxm.hash[-2:], delimit=False)
filename = "Paper_Message_"+time.strftime(file_ts_format, time.localtime(m["sent"]))+"_"+hash_str+".png"
# filename = "Paper_Message.png"
self.app.share_image(qr_image, filename)
item.dmenu.dismiss()
return x
else:
def x():
try:
qr_image = lxm.as_qr()
hash_str = RNS.hexrep(lxm.hash[-2:], delimit=False)
filename = "Paper_Message_"+time.strftime(file_ts_format, time.localtime(m["sent"]))+"_"+hash_str+".png"
if RNS.vendor.platformutils.is_darwin():
save_path = str(plyer.storagepath.get_downloads_dir()+filename).replace("file://", "")
else:
save_path = plyer.storagepath.get_downloads_dir()+"/"+filename
qr_image.save(save_path)
item.dmenu.dismiss()
ok_button = MDRectangleFlatButton(text="OK",font_size=dp(18))
dialog = MDDialog(
title="QR Code Saved",
text="The paper message has been saved to: "+save_path+"",
buttons=[ ok_button ],
# elevation=0,
)
def dl_ok(s):
dialog.dismiss()
ok_button.bind(on_release=dl_ok)
dialog.open()
except Exception as e:
item.dmenu.dismiss()
ok_button = MDRectangleFlatButton(text="OK",font_size=dp(18))
dialog = MDDialog(
title="Error",
text="Could not save the paper message QR-code to:\n\n"+save_path+"\n\n"+str(e),
buttons=[ ok_button ],
# elevation=0,
)
def dl_ok(s):
dialog.dismiss()
ok_button.bind(on_release=dl_ok)
dialog.open()
return x
def gen_print_qr(lxm, item):
if RNS.vendor.platformutils.is_android():
def x():
item.dmenu.dismiss()
return x
else:
def x():
try:
qr_image = lxm.as_qr()
qr_tmp_path = self.app.sideband.tmp_dir+"/"+str(RNS.hexrep(lxm.hash, delimit=False))
qr_image.save(qr_tmp_path)
print_command = self.app.sideband.config["print_command"]+" "+qr_tmp_path
return_code = subprocess.call(shlex.split(print_command), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
os.unlink(qr_tmp_path)
item.dmenu.dismiss()
except Exception as e:
item.dmenu.dismiss()
ok_button = MDRectangleFlatButton(text="OK",font_size=dp(18))
dialog = MDDialog(
title="Error",
text="Could not print the paper message QR-code.\n\n"+str(e),
buttons=[ ok_button ],
# elevation=0,
)
def dl_ok(s):
dialog.dismiss()
ok_button.bind(on_release=dl_ok)
dialog.open()
return x
retry_item = {
"viewclass": "OneLineListItem",
"text": "Retry",
"height": dp(40),
"on_release": gen_retry(m["hash"], m["content"], item)
}
if m["method"] == LXMF.LXMessage.PAPER:
if RNS.vendor.platformutils.is_android():
qr_save_text = "Share QR Code"
dm_items = [
{
"viewclass": "OneLineListItem",
"text": "Share QR Code",
"height": dp(40),
"on_release": gen_save_qr(m["lxm"], item)
},
{
"viewclass": "OneLineListItem",
"text": "Copy LXM URI",
"height": dp(40),
"on_release": gen_copy_lxm_uri(m["lxm"], item)
},
{
"viewclass": "OneLineListItem",
"text": "Copy message text",
"height": dp(40),
"on_release": gen_copy(m["content"].decode("utf-8"), item)
},
{
"text": "Delete",
"viewclass": "OneLineListItem",
"height": dp(40),
"on_release": gen_del(m["hash"], item)
}
]
else:
dm_items = [
{
"viewclass": "OneLineListItem",
"text": "Print QR Code",
"height": dp(40),
"on_release": gen_print_qr(m["lxm"], item)
},
{
"viewclass": "OneLineListItem",
"text": "Save QR Code",
"height": dp(40),
"on_release": gen_save_qr(m["lxm"], item)
},
{
"viewclass": "OneLineListItem",
"text": "Copy LXM URI",
"height": dp(40),
"on_release": gen_copy_lxm_uri(m["lxm"], item)
},
{
"viewclass": "OneLineListItem",
"text": "Copy message text",
"height": dp(40),
"on_release": gen_copy(m["content"].decode("utf-8"), item)
},
{
"text": "Delete",
"viewclass": "OneLineListItem",
"height": dp(40),
"on_release": gen_del(m["hash"], item)
}
]
else:
if m["state"] == LXMF.LXMessage.FAILED:
dm_items = [
retry_item,
{
"viewclass": "OneLineListItem",
"text": "Copy",
"height": dp(40),
"on_release": gen_copy(m["content"].decode("utf-8"), item)
},
{
"text": "Delete",
"viewclass": "OneLineListItem",
"height": dp(40),
"on_release": gen_del(m["hash"], item)
}
]
else:
if "lxm" in m and m["lxm"] and m["lxm"].fields != None and LXMF.FIELD_TELEMETRY in m["lxm"].fields:
packed_telemetry = m["lxm"].fields[LXMF.FIELD_TELEMETRY]
dm_items = [
{
"viewclass": "OneLineListItem",
"text": "Copy",
"height": dp(40),
"on_release": gen_copy(m["content"].decode("utf-8"), item)
},
{
"viewclass": "OneLineListItem",
"text": "Copy telemetry",
"height": dp(40),
"on_release": gen_copy_telemetry(packed_telemetry, extra_telemetry, item)
},
{
"text": "Delete",
"viewclass": "OneLineListItem",
"height": dp(40),
"on_release": gen_del(m["hash"], item)
}
]
else:
dm_items = [
{
"viewclass": "OneLineListItem",
"text": "Copy",
"height": dp(40),
"on_release": gen_copy(m["content"].decode("utf-8"), item)
},
{
"text": "Delete",
"viewclass": "OneLineListItem",
"height": dp(40),
"on_release": gen_del(m["hash"], item)
}
]
item.dmenu = MDDropdownMenu(
caller=item.ids.msg_submenu,
items=dm_items,
position="auto",
width=dp(256),
elevation=0,
radius=dp(3),
)
item.dmenu.retry_item = retry_item
def callback_factory(ref):
def x(sender):
ref.dmenu.open()
return x
# Bind menu open
item.ids.msg_submenu.bind(on_release=callback_factory(item))
if self.loading_earlier_messages:
insert_pos = len(self.list.children)
else:
insert_pos = 0
self.added_item_hashes.append(m["hash"])
self.widgets.append(item)
self.list.add_widget(item, insert_pos)
if self.latest_message_timestamp == None or m["received"] > self.latest_message_timestamp:
self.latest_message_timestamp = m["received"]
if self.earliest_message_timestamp == None or m["received"] < self.earliest_message_timestamp:
self.earliest_message_timestamp = m["received"]
self.added_messages += len(self.new_messages)
self.new_messages = []
def get_widget(self):
return self.list
def close_send_error_dialog(self, sender=None):
if self.send_error_dialog:
self.send_error_dialog.dismiss()
messages_screen_kv = """
MDScreen:
name: "messages_screen"
BoxLayout:
orientation: "vertical"
MDTopAppBar:
id: messages_toolbar
anchor_title: "left"
title: "Messages"
elevation: 0
left_action_items:
[['menu', lambda x: root.app.nav_drawer.set_state("open")],]
right_action_items:
[
['map-search', lambda x: root.app.peer_show_location_action(self)],
['lan-connect', lambda x: root.app.message_propagation_action(self)],
['close', lambda x: root.app.close_settings_action(self)],
]
ScrollView:
id: messages_scrollview
do_scroll_x: False
do_scroll_y: True
BoxLayout:
id: no_keys_part
orientation: "vertical"
padding: [dp(16), dp(0), dp(16), dp(16)]
spacing: dp(24)
size_hint_y: None
height: self.minimum_height + dp(64)
MDLabel:
id: nokeys_text
text: ""
MDRectangleFlatIconButton:
icon: "key-wireless"
text: "Query Network For Keys"
on_release: root.app.key_query_action(self)
BoxLayout:
id: message_input_part
padding: [dp(16), dp(0), dp(16), dp(16)]
spacing: dp(24)
size_hint_y: None
height: self.minimum_height
MDTextField:
id: message_text
input_type: "text"
keyboard_suggestions: True
multiline: True
hint_text: "Write message"
mode: "rectangle"
max_height: dp(100)
MDRectangleFlatIconButton:
id: message_send_button
icon: "transfer-up"
text: "Send"
padding: [dp(10), dp(13), dp(10), dp(14)]
icon_size: dp(24)
font_size: dp(16)
on_release: root.app.message_send_action(self)
"""
Builder.load_string("""
<ListLXMessageCard>:
style: "outlined"
elevation: 2
padding: dp(8)
radius: [dp(4), dp(4), dp(4), dp(4)]
size_hint: 1.0, None
height: content_text.height + heading_text.height + dp(32)
pos_hint: {"center_x": .5, "center_y": .5}
MDRelativeLayout:
size_hint: 1.0, None
theme_text_color: "ContrastParentBackground"
MDIconButton:
id: msg_submenu
icon: "dots-vertical"
# theme_text_color: 'Custom'
# text_color: rgba(255,255,255,216)
pos:
root.width - (self.width + root.padding[0] + dp(4)), root.height - (self.height + root.padding[0] + dp(4))
MDLabel:
id: heading_text
markup: True
text: root.heading
adaptive_size: True
# theme_text_color: 'Custom'
# text_color: rgba(255,255,255,100)
pos: 0, root.height - (self.height + root.padding[0] + dp(8))
MDLabel:
id: content_text
text: root.text
# adaptive_size: True
size_hint_y: None
text_size: self.width, None
# theme_text_color: 'Custom'
# text_color: rgba(255,255,255,216)
height: self.texture_size[1]
<CustomOneLineIconListItem>
IconLeftWidget:
icon: root.icon
""")