forked from mhoek2/tcp-share
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
162 lines (128 loc) · 6.51 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
from cgi import print_exception
from tkinter import *
import threading
import time
# app core modules
from modules.app.settings import Settings
from modules.app.gui import Gui
from modules.app.read_write import ReadWrite
from modules.translate import Translate
from modules.qrcode import QRCode
from modules.logging import Logging
# app modules
from modules.crypt import Crypt
from modules.gui.share_files import GUI_ShareFiles
from modules.gui.view_files import GUI_ViewFiles
from modules.gui.view_logs import GUI_ViewLogs
from modules.tcp import TCP
from modules.to_pdf import ToPDF
class Application( Tk ):
"""Base class for this application"""
def __init__( self, pytest=False ) -> None:
self.settings : Settings = Settings()
self.tcp : TCP = TCP( self )
self.read_write : ReadWrite = ReadWrite( self )
self.tcp.load_lan_devices();
self.translate : Translate = Translate( self )
self.qrcode : QRCode = QRCode( self )
self.tk_root : Gui = Gui( self )
# empty placeholder modules
self.crypt : Crypt = Crypt( self )
self.to_pdf : ToPDF = ToPDF( self )
self.log : Logging = Logging( self )
if not pytest:
# background worker to check incoming TCP connections
self.tcp_worker()
# background worker for appliction
self.bg_worker()
self.tk_root.mainloop()
#
# Keep the following in main for readability (flow-chart)
#
# TCP thread: The socket server to allow incoming connections:
def tcp_worker( self ) -> None:
self.tcp_server = threading.Thread( target=self.tcp.start_server, daemon=True )
self.tcp_server.start()
# Background thread:
# - Continuesly scan 'files' folder for changes
# - Run GUI updates
def bg_worker( self ) -> None:
self.bg_worker_tick = 0
self.bg_worker_force_gui = False
self.bg_worker_force_files = False
self.bg_worker = threading.Thread( target=self.bg_worker_do, daemon=True )
self.bg_worker.start()
def bg_worker_force_gui_update( self ):
""""Force a pass in the background worker to enter the GUI pass"""
self.bg_worker_force_gui = True
def bg_worker_force_file_update( self ):
""""Force a file update pass in the background worker"""
self.bg_worker_force_files = True
#
# perhaps 'has_files' and 'file_count_changed' should be moved to modules/app/read_write.py?
#
def has_files( self ) -> bool:
"""Check if files in 'txt' or 'pdf' exist.
'hasTextFiles()' and 'hasPdfFiles()' updates the current file count as well."""
try:
has_text_files = self.read_write.hasTextFiles()
has_pdf_files = self.read_write.hasPdfFiles()
return has_text_files or has_pdf_files
except Exception as e:
print(f"An error occurred: {e}")
return False
def file_count_changed( self ) -> bool:
"""Deteriminte wheter the file count has changed since previous iteration."""
if self.read_write.numShareableFiles != self.read_write.prevShareableFiles:
print( f"Files have changed? num txt files: {self.read_write.numShareableFiles}" )
return True
if self.read_write.numPDFFiles != self.read_write.prevPDFFiles:
print( f"Files have changed? num pdf files: {self.read_write.numPDFFiles}" )
return True
return False
def update_previous_file_count( self ) -> None:
"""This method sets the file count to the current count.
This is useful for tracking changes in file counts across operations."""
self.read_write.prevShareableFiles = self.read_write.numShareableFiles
self.read_write.prevPDFFiles = self.read_write.numPDFFiles
def bg_worker_do( self ) -> None:
"""Background worker that continuously monitors and updates the application state.
This method performs the following actions in an infinite loop:
- Checks for changes in the file count and takes following actions:
- If there are changes and files exist while not on the creation page, it reloads the current frame.
- If there are no files or on the creation page, it navigates back to the home frame.
- Updates the recorded file count for the next iteration comparisons.
- Periodically refreshes device data every 10 ticks when in the share files frame.
- If a forced device update is triggered, it updates devices immediately.
The loop sleeps for one second between iterations to manage the frequency.
"""
while True:
print( self.bg_worker_tick )
# Monitor changes in the file count by comparing the current count with the previous iteration.
# If files exist and the current page handles file reading, reload the current frame.
# If the file count is zero or the application is on the creation page, call goHome().
# The goHome() function will determine whether to display the create_files frame or the share_files frame.
has_files = self.has_files()
num_changed = self.file_count_changed()
if num_changed or self.bg_worker_force_files:
if has_files and self.tk_root.current_frame > self.tk_root.FRAME_CREATE_FILES:
# reload the current frame/page
reload_frame : bool = True # redundant bool ..
self.tk_root.after( 10, self.tk_root.show_frame( self.tk_root.current_frame, reload_frame ) )
else:
self.tk_root.after( 10, self.tk_root.goHome() )
self.update_previous_file_count()
if self.bg_worker_force_files:
self.bg_worker_force_files = False
# Update the device list every X seconds when the share_files frame is open.
# If self.bg_worker_force_gui is True, trigger an earlier update.
if self.bg_worker_tick % 10 == 0 or self.bg_worker_force_gui:
if self.tk_root.is_frame_active( self.tk_root.FRAME_SHARE_FILES ):
self.tk_root.after(10, self.tk_root.frames[ self.tk_root.FRAME_SHARE_FILES ].updateDevices() )
print("x")
if self.bg_worker_force_gui:
self.bg_worker_force_gui = False
self.bg_worker_tick += 1
time.sleep(1)
if __name__ == '__main__':
app = Application()