so, I have app in kivy/kivymd- it contains 2 windows, 1st window runs in infinity loop looking for input, but I want it to check form time to time whether or not it has internet connection - probably in 2nd thread.
KV settings:
KV = """
WindowManager:
MainWindow:
SecondWindow:
<MainWindow>:
name: "MainWindow"
canvas:
Color:
rgba: self.background
Rectangle:
pos: self.pos
size: self.size
BoxLayout:
id: layout
orientation: "vertical"
pos: self.pos
size: self.size
Label:
text: ''
font_size: 1
Image:
source: 'logo.jpg'
allow_stretch: True
size_hint_y: None
width: 80
Label:
id: info
text_size: self.size
text: 'scan here'
font_size: 26
size_hint_y: None
text_size: self.width, None
height: self.texture_size[1]
halign: 'center'
Label:
id: clock
text: ''
font_size: 60
TextInput:
id: kod
text: ''
multiline: False
size_hint: (0, .5)
font_size: 15
#opacity: 0
focus: True
password: True
text_validate_unfocus: False
on_text_validate: root.update_label(kod.text)
MDRaisedButton:
text: 'button1'
icon_color: 1, 1, 1, 1
size_hint: 0.3, 0.5
pos_hint: {"x":0.35, "top":1}
on_release:
app.root.current = "SecondWindow"
Label:
id: wifi
text: ''
font_size: 5
Label:
text: 'gui apk'
<SecondWindow>
name: "SecondWindow"
MDBoxLayout:
id: layout
orientation: "vertical"
pos: self.pos
size: self.size
md_bg_color: 0.3, 0.3, 0.3, 1.0
Label:
text: 'log in'
font_size: 22
MDRaisedButton:
text: 'button1'
size_hint: 0.3, 0.5
pos_hint: {"x":0.35, "top":1}
on_release:
app.show_date_picker()
MDRaisedButton:
text: 'exit'
size_hint: 0.3, 0.5
pos_hint: {"x":0.35, "top":1}
on_release:
root.manager.current = "MainWindow"
root.manager.get_screen("MainWindow").ids["kod"].focus = True
Label:
text: 'gui apk'
"""
Code:
import fdb
import configparser
import threading
import time
import socket
from kivymd.app import MDApp
from kivy.properties import ListProperty
from kivy.lang import Builder
from kivy.clock import Clock
from kivy.uix.screenmanager import Screen, ScreenManager
from kivy.properties import StringProperty
from kivy.clock import mainthread
from kivymd.uix.pickers import MDDatePicker
config = configparser.ConfigParser()
config.read(r'C:\\Users\\Przemek\\Desktop\\RCP\\config.txt')
RED_BACKGROUND = [.75, 0, 0, 1.0]
BLACK_BACKGROUND = [0.3, 0.3, 0.3, 1.0]
GREEN_BACKGROUND = [0.2, 0.3, 0.1, 1]
BLUE_BACKGROUND = [0, 0, 1, 1]
class MainWindow(Screen):
background = ListProperty()
def __init__(self, **kwargs):
super(MainWindow, self).__init__(**kwargs)
self.background = BLACK_BACKGROUND
Clock.schedule_interval( self.UpdateClock, 1 )
threading.Thread().start()
@mainthread
def update_label(self, val):
con = fdb.connect(
port = int(config.get('global', 'port')),
database = str(config.get('global', 'database')),
host = str(config.get('global', 'host')),
user = str(config.get('global', 'user')),
password = str(config.get('global', 'password')),
charset = str(config.get('global', 'charset'))
)
cur = con.cursor()
cur.execute("select KOD, FK_KRT_PRC, ID FROM TABLE")
if len(val) > 0:
print('main loop here')
con.commit()
con.close()
self.ids.kod.text = ''
def SetStartScreen( self, dt ):
self.ids.info.text = 'scan'
self.background = BLACK_BACKGROUND
def UpdateClock( self, dt ):
self.ids.clock.text = time.strftime("%H:%M:%S")
class SecondWindow(Screen):
pass
class WindowManager(ScreenManager):
pass
class MyMainApp(MDApp):
def build(self):
self.title = "gui apk"
return Builder.load_string(KV)
def on_save(self, instance, value, date_range):
print(instance, value, date_range)
pass
def on_cancel(self, instance, value):
pass
def show_date_picker(self):
date_dialog = MDDatePicker(mode="range")
date_dialog.bind(on_save=self.on_save, on_cancel=self.on_cancel)
date_dialog.open()
if __name__ == "__main__":
MyMainApp().run()
My function to test internet connection
def check_connection():
try:
host = socket.gethostbyname('www.google.com')
s = socket.create_connection((host, 80), 2)
#print(True)
return True
except:
return False
res = check_connection()
print(res)
I try to put check_connection and call it in various places but w/o success
I tested the code below. One thing I noticed in your question is the line threading.Thread().start(); this will not do anything because no function was submitted as the target of the thread.
import socket
import time
from threading import Thread, Event
def check_connection(internet_on: Event, interval: float = 60.0) -> None:
print("starting function intended to be in a new thread")
internet_on.clear()
# one could also import Queue from queue and return specific error
# messages by putting them in the queue in this thread and checking
# for them in the main thread.
while True:
try:
_host = socket.gethostbyname('www.google.com')
s = socket.create_connection((_host, 80), 2)
except socket.gaierror:
internet_on.clear()
except Exception as e:
internet_on.clear()
else:
internet_on.set()
time.sleep(interval)
_internet = Event()
# set daemon to True so the thread automatically closes when your main program ends
new_thread = Thread(target=check_connection, args=(_internet, 5.0), daemon=True)
new_thread.start()
# below is just an example so this code will execute
if __name__ == '__main__':
while True:
time.sleep(0.5)
print(f"internet on = {_internet.is_set()}")