root
@classmethod
def set(cls, tree):
cls._tree = tree if tree is not None else ET.Element("tv")
@classmethod
def get(cls):
return cls._tree
@classmethod
def append(cls, programme):
if cls._tree is not None:
cls._tree.append(programme)
@classmethod
def clear(cls):
cls._tree = ET.Element("tv")
class IPTVWindow(QMainWindow):
def __init__(self, php_dir):
self.channel_video_info = {}
self.epg_grid_window = None
self.channel_logos = {} # โ
initialize before it's used
super().__init__()
self.setWindowTitle("DSPT IPTV Player")
self.setGeometry(100, 100, 1280, 720)
self.php_dir = php_dir
self._is_fullscreen = False
self._normal_geometry = None
self._normal_flags = None
self.current_channel_epg = None
self.epg_last_update = None
self.debug_log = []
self.default_epg_urls = [] # Holds default EPG URLs from playlist(s)
self.custom_epg_sources = [] # Holds both custom EPG files and URLs
self.channel_order = [] # โ
Prevent AttributeError
self.full_channel_order = list(self.channel_order)
self.init_ui()
# Get usable screen area
screen = QGuiApplication.primaryScreen().availableGeometry()
screen_left = screen.left()
screen_top = screen.top()
screen_width = screen.width()
screen_height = screen.height()
# Dimensions for top-right quadrant
window_width = screen_width // 2
window_height = screen_height // 2
# Shift down a bit to show title bar and avoid EPG overlap
safe_margin = 40 # ensures title bar is visible
window_top = screen_top + safe_margin
window_left = screen_left + window_width # right half
# Slightly reduce height to prevent overlapping EPGGrid
adjusted_height = screen_height // 2 - safe_margin
self.setGeometry(window_left, window_top, window_width, adjusted_height)
self.epg_data = ET.Element("tv") # placeholder in case it's accessed early
self.show() # โ
Show Main Window Immediately
# STEP 2: Show blank EPG Grid window (structure only)
self.epg_grid_window = EPGGridWindow(ET.Element("tv"), [], {}, self)
self.epg_grid_window.show()
# STEP 3: Show loading message and schedule playlist download in background
self.show_status_message("๐ฅ Downloading Default Playlist...")
QTimer.singleShot(1000, self.download_playlist)
self.stream_error_label = None # for "Stream not available" overlay
def init_ui(self):
self.splitter = QSplitter(Qt.Horizontal)
# Left panel with channels
self.channel_list = QListWidget()
self.channel_list.itemClicked.connect(self.on_channel_clicked)
self.splitter.addWidget(self.channel_list)
self.channel_list.hide() # โ
Hide the old playlist
# Right panel with video and EPG
right_panel = QWidget()
layout = QVBoxLayout(right_panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self.video_container = QWidget()
self.video_container.setStyleSheet("background-color: black;")
self.video_container.setMinimumHeight(400)
video_container_layout = QVBoxLayout(self.video_container)
video_container_layout.setContentsMargins(0, 0, 0, 0)
self.video_frame = VideoFrame(self.video_container)
self.video_frame.mouseDoubleClickEvent = self.toggle_fullscreen
video_container_layout.addWidget(self.video_frame)
layout.addWidget(self.video_container, stretch=3)
self.player = VLCPlayer(self.video_frame)
# EPG Display with new formatting
self.epg_display = QLabel()
self.epg_display.setStyleSheet("""
background-color: #222;
color: white;
padding: 10px;
""")
self.epg_display.setWordWrap(True)
self.epg_display.setAlignment(Qt.AlignTop | Qt.AlignLeft)
layout.addWidget(self.epg_display, stretch=1)
# Timers
self.clock_timer = QTimer()
self.clock_timer.timeout.connect(self.refresh_epg_time)
self.clock_timer.start(1000)
self.epg_update_timer = QTimer()
self.epg_update_timer.timeout.connect(self.update_epg_data)
self.epg_update_timer.start(3600000) # Update EPG every hour
# Controls
self.controls = QWidget()
ctl_layout = QHBoxLayout(self.controls)
for text, slot in [("โฎ๏ธ", self.play_previous_channel),
("โฏ๏ธ", self.toggle_play_pause),
("โญ๏ธ", self.play_next_channel),
("๐ต", self.toggle_fullscreen),
("๐
", self.toggle_epg_grid),
("๐", self.update_epg_data)]:
btn = QPushButton(text)
btn.clicked.connect(slot)
ctl_layout.addWidget(btn)
self.volume_slider = QSlider(Qt.Horizontal)
self.volume_slider.setRange(0, 200)
self.volume_slider.setValue(70)
self.volume_slider.setToolTip("Volume")
self.volume_slider.valueChanged.connect(self.set_volume)
ctl_layout.addWidget(self.volume_slider)
self.mute_button = QPushButton("๐")
self.mute_button.setCheckable(True)
self.mute_button.clicked.connect(self.toggle_mute)
ctl_layout.addWidget(self.mute_button)
layout.addWidget(self.controls)
self.splitter.addWidget(right_panel)
self.splitter.setSizes([300, 980])
self.setCentralWidget(self.splitter)
self.last_epg_title = ""
self.last_epg_desc = ""
def cleanup_resources(self):
self.log_debug("Cleaning up timers and resources")
if self.epg_grid_window:
self.epg_grid_window.close()
self.epg_grid_window.deleteLater()
self.epg_grid_window = None
self.clock_timer.stop()
self.epg_update_timer.stop()
self.player.stop()
self.player.release()
global php_proc
if php_proc:
try:
php_proc.terminate()
php_proc.wait(2)
except Exception as e:
self.log_debug(f"[PHP Cleanup] Error: {e}")
php_proc = None
self.log_debug("Cleanup finished โ exiting app")
QTimer.singleShot(100, QApplication.instance().quit) # exit only after UI fully cleaned
def closeEvent(self, event):
self.log_debug("Application closing - starting async cleanup")
# Hide the main window immediately for fast visual exit
self.hide()
# Defer cleanup after 100ms so UI can close first
QTimer.singleShot(100, self.cleanup_resources)
event.accept() # โ
allow close to continue
super().closeEvent(event)
def detect_resolution(self, item, base):
if not self.player.media_player.is_playing():
print("[VLC] Stream failed โ displaying fallback message")
self.player.media_player.stop()
# Remove any previous error label
if self.stream_error_label:
self.stream_error_label.deleteLater()
self.stream_error_label = None
# Show persistent "Stream not available" message
self.stream_error_label = QLabel("Stream is not available", self.video_frame)
self.stream_error_label.setAlignment(Qt.AlignCenter)
self.stream_error_label.setStyleSheet("""
color: red;
font-size: 18pt;
background-color: rgba(0, 0, 0, 180);
""")
self.stream_error_label.setGeometry(0, 0, self.video_frame.width(), self.video_frame.height())
self.stream_error_label.show()
# โ
Restart PHP only if needed and it's a PHP-based stream
url = item.data(Qt.UserRole)
if url and url.startswith("http://localhost:8888"):
if php_proc is None or php_proc.poll() is not None:
print("[PHP] Detected PHP server is not running. Restarting...")
restart_php_server(self.php_dir)
else:
print("[PHP] Forcing restart due to stuck stream...")
restart_php_server(self.php_dir)
self.current_channel_resolution = "Stream not available"
self.update_epg_info(self.current_channel_name or base, "Stream not available")
return
# โ
Stream is good โ remove error label
if self.stream_error_label:
self.stream_error_label.deleteLater()
self.stream_error_label = None
# Continue with resolution and FPS
w = self.player.media_player.video_get_width()
h = self.player.media_player.video_get_height()
fps = self.player.media_player.get_fps()
if w > 0 and h > 0:
label = "SD"
if w >= 3840:
label = "4K"
elif w >= 1920:
label = "FHD"
elif w >= 1280:
label = "HD"
info = f"{label}-{fps:.2f} FPS"
self.channel_video_info[base] = (label, f"{fps:.2f}")
item.setText(f"{base} ({info})")
self.current_channel_name = base
self.current_channel_resolution = info
self.update_epg_info(base, info)
if self.epg_grid_window:
self.epg_grid_window.refresh_channel_fps_info()
def download_playlist(self):
try:
os.makedirs(self.php_dir, exist_ok=True)
url = "https://iptv.nywebforum.com/playlist.m3u"
self.log_debug(f"Downloading playlist from: {url}")
response = requests.get(url, timeout=10)
lines = response.text.splitlines()
fixed_lines = []
for line in lines:
if line.startswith("http") and "iptv.nywebforum.com" in line:
line = line.replace("https://iptv.nywebforum.com", "http://localhost:8888")
fixed_lines.append(line)
content = "\n".join(fixed_lines)
# Save modified playlist (with local proxy)
path = os.path.join(self.php_dir, "playlist.m3u")
with open(path, "w", encoding="utf-8") as f:
f.write(content)
self.log_debug(f"Playlist saved to: {path}")
# Load into app
self.load_playlist(path)
# Step 4: Fill dummy EPG data
self.insert_dummy_epg_gaps()
if self.epg_grid_window:
self.epg_grid_window.update_epg(self.channel_order, self.channel_video_info)
# STEP 5: Start internal PHP server in background
threading.Thread(target=lambda: restart_php_server(self.php_dir), daemon=True).start()
# Step 6: Start playing first channel immediately
if self.channel_list.count() > 0:
first = self.channel_list.item(0)
self.channel_list.setCurrentItem(first)
QTimer.singleShot(500, lambda: self.on_channel_clicked(first))
# Step 7: Extract EPG URLs from first line if present
first_line = response.text.split('\n')[0]
if first_line.startswith("#EXTM3U") and "url-tvg=" in first_line:
epg_raw = first_line.split('url-tvg="')[1].split('"')[0]
epg_urls = [url.strip() for url in epg_raw.split(',') if url.strip()]
new_urls = [url for url in epg_urls if url not in self.default_epg_urls]
self.default_epg_urls.extend(new_urls)
if epg_urls:
self.log_debug(f"Found EPG URLs: {epg_urls}")
QTimer.singleShot(2000, self.update_epg_data) # Delayed fetch in background
# โ
Step 8: Load logos in background (after everything else settles)
def start_logo_prefetch():
def prefetch_logos():
os.makedirs(CACHE_DIR, exist_ok=True)
for name, url in self.channel_logos.items():
if not url:
continue
try:
logo_filename = hashlib.md5(url.encode('utf-8')).hexdigest() + ".png"
logo_path = os.path.join(CACHE_DIR, logo_filename)
if not os.path.exists(logo_path):
response = requests.get(url, timeout=5)
if response.status_code == 200:
with open(logo_path, "wb") as f:
f.write(response.content)
except Exception as e:
print(f"[LOGO] Failed to fetch logo for {name}: {e}")
# Refresh EPG visuals from Qt thread
if self.epg_grid_window:
QTimer.singleShot(0, self.epg_grid_window.refresh_channel_fps_info)
threading.Thread(target=prefetch_logos, daemon=True).start()
self.log_debug("๐ผ๏ธ Background logo fetcher started...")
# Delay logo prefetching so it runs AFTER first channel is loaded
QTimer.singleShot(4000, start_logo_prefetch)
except Exception as e:
self.log_debug(f"Failed to download playlist: {e}")
def insert_dummy_epg_gaps(self):
if self.epg_data is None:
return
local_tz = pytz.timezone("Asia/Karachi")
timeline_start = self.epg_grid_window.timeline_start if self.epg_grid_window else datetime.now(local_tz)
timeline_end = self.epg_grid_window.timeline_end if self.epg_grid_window else timeline_start + timedelta(hours=6)
for _, ch_id in self.channel_order:
progs = [p for p in self.epg_data.findall(".//programme") if p.get("channel") == ch_id]
# Convert to (start, stop, element)
parsed = []
for p in progs:
try:
start = datetime.strptime(p.get("start")[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc).astimezone(local_tz)
stop = datetime.strptime(p.get("stop")[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc).astimezone(local_tz)
parsed.append((start, stop, p))
except:
continue
parsed.sort(key=lambda x: x[0])
new_programmes = []
current_time = timeline_start
for start, stop, orig_elem in parsed:
if start > current_time:
# Insert dummy gap
gap_start = current_time
gap_end = start
while gap_start < gap_end:
next_gap = min(gap_end, gap_start + timedelta(hours=1))
dummy = ET.Element("programme")
dummy.set("channel", ch_id)
dummy.set("start", gap_start.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S") + " +0000")
dummy.set("stop", next_gap.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S") + " +0000")
title = ET.SubElement(dummy, "title")
title.text = "No Program Information"
new_programmes.append(dummy)
gap_start = next_gap
new_programmes.append(orig_elem)
current_time = max(current_time, stop)
# Fill any remaining gap to timeline_end
if current_time < timeline_end:
gap_start = current_time
while gap_start < timeline_end:
next_gap = min(timeline_end, gap_start + timedelta(hours=1))
dummy = ET.Element("programme")
dummy.set("channel", ch_id)
dummy.set("start", gap_start.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S") + " +0000")
dummy.set("stop", next_gap.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S") + " +0000")
title = ET.SubElement(dummy, "title")
title.text = "No Program Information"
new_programmes.append(dummy)
gap_start = next_gap
# Remove old and insert new
for p in progs:
self.epg_data.remove(p)
for p in new_programmes:
self.epg_data.append(p)
def keyPressEvent(self, event):
if event.key() == Qt.Key_Escape and self._is_fullscreen:
self.toggle_fullscreen()
else:
super().keyPressEvent(event)
def load_playlist(self, path):
try:
self.log_debug(f"Loading playlist from: {path}")
lines = open(path, encoding="utf-8").read().splitlines()
self.channel_list.clear()
first = None
name = None
channel_id = None
self.channel_order = [] # โ
Reset and initialize channel order
for i, l in enumerate(lines):
if l.startswith("#EXTINF:"):
name = l.split(',')[-1].strip()
if 'tvg-id="' in l:
channel_id = l.split('tvg-id="')[1].split('"')[0]
self.log_debug(f"Found channel ID: {channel_id} for {name}")
if 'tvg-logo="' in l:
logo = l.split('tvg-logo="')[1].split('"')[0]
if name:
self.channel_logos[name] = logo
self.log_debug(f"Logo for {name}: {logo}")
elif l and not l.startswith("#") and name:
it = QListWidgetItem(f"{name} (Loading...)")
it.setData(Qt.UserRole, l) # stream URL
if channel_id:
it.setData(Qt.UserRole + 1, channel_id)
self.channel_order.append((name, channel_id)) # โ
Maintain order for EPG
it.setData(Qt.UserRole + 2, len(self.channel_order) - 1) # โ
Use correct index
self.channel_list.addItem(it)
if first is None:
first = it
name = None
channel_id = None
if first:
self.channel_list.setCurrentItem(first)
self.log_debug(f"Found {self.channel_list.count()} channels, selecting first one")
QTimer.singleShot(1000, lambda: self.on_channel_clicked(first))
except Exception as e:
self.log_debug(f"Failed to load playlist: {e}")
def log_debug(self, message):
"""Add debug message to log and print to console"""
timestamp = datetime.now().strftime("%H:%M:%S")
full_message = f"[{timestamp}] {message}"
self.debug_log.append(full_message)
print(full_message)
if hasattr(self, 'debug_display'):
self.debug_display.addItem(full_message)
self.debug_display.scrollToBottom()
def on_channel_clicked(self, item: QListWidgetItem):
url = item.data(Qt.UserRole)
if not url:
return
base = item.text().split('(')[0].strip()
item.setText(f"{base} (Loading...)")
self.current_channel_name = base
self.current_channel_resolution = "Loading..."
channel_id = item.data(Qt.UserRole + 1)
if channel_id:
self.current_channel_id = channel_id
self.update_current_epg()
# โ
Now that EPG is ready, display it
self.update_epg_info(base, "Loading...")
self.log_debug(f"Playing stream: {url}")
self.player.stop()
self.player.play_stream(url)
def retry_if_failed():
if self.player.media_player.is_playing():
return # โ
First try succeeded
self.log_debug("[RETRY] First attempt failed, retrying stream...")
self.player.stop()
self.player.release()
self.player = VLCPlayer(self.video_frame)
if url.startswith("http://localhost:8888"):
self.log_debug("[RETRY] Waiting for PHP server to be ready...")
# โ
Try to wait up to 3 seconds max
if not wait_for_php_server(timeout=3):
self.log_debug("[RETRY] PHP not ready. Restarting and waiting again...")
restart_php_server(self.php_dir)
time.sleep(1.0)
if not wait_for_php_server(timeout=3):
self.log_debug("[RETRY] PHP still not ready. Giving up.")
return
self.player.play_stream(url)
QTimer.singleShot(10000, retry_if_failed)
channel_id = item.data(Qt.UserRole + 1)
if channel_id:
self.current_channel_id = channel_id
self.update_current_epg()
QTimer.singleShot(5000, lambda: self.detect_resolution(item, base))
def play_next_channel(self):
idx = self.channel_list.currentRow()
if idx < self.channel_list.count() - 1:
self.channel_list.setCurrentRow(idx+1)
self.on_channel_clicked(self.channel_list.currentItem())
def play_previous_channel(self):
idx = self.channel_list.currentRow()
if idx > 0:
self.channel_list.setCurrentRow(idx-1)
self.on_channel_clicked(self.channel_list.currentItem())
def process_epg_data(self, epg_bytes):
try:
# Try to decompress the data
try:
epg_xml = gzip.decompress(epg_bytes).decode('utf-8')
self.log_debug("EPG data decompressed successfully")
except:
epg_xml = epg_bytes.decode('utf-8')
self.log_debug("EPG data was not compressed")
# Save the raw XML for debugging
debug_epg_path = os.path.join(self.php_dir, "last_epg.xml")
with open(debug_epg_path, "w", encoding="utf-8") as f:
f.write(epg_xml)
self.log_debug(f"Saved EPG data to {debug_epg_path}")
# Parse the XML
EPGStore.set(ET.fromstring(epg_xml))
self.insert_dummy_epg_gaps()
self.epg_last_update = datetime.now()
if self.epg_grid_window:
# Defer overlay creation safely inside update_epg()
self.epg_grid_window.update_epg(self.channel_order, self.channel_video_info)
else:
# Create fresh EPGGridWindow AFTER full EPG is available
self.epg_grid_window = EPGGridWindow(
EPGStore.get(),
self.channel_order,
self.channel_video_info,
self,
channel_logos=self.channel_logos
)
self.epg_grid_window.show()
self.log_debug(f"EPG data parsed successfully, last update: {self.epg_last_update}")
# Update current channel's EPG if available
if hasattr(self, 'current_channel_id') and self.current_channel_id:
print("[EPG] Initializing current channel EPG immediately after download")
self.update_current_epg()
self.update_epg_info(self.current_channel_name, self.current_channel_resolution)
# Setup auto-refresh aligned to clock
self.setup_epg_auto_refresh()
except Exception as e:
self.log_debug(f"Error processing EPG data: {str(e)}")
def refresh_epg_time(self):
if not hasattr(self, 'current_channel_name') or not hasattr(self, 'current_channel_resolution'):
return
text = self.epg_display.text()
parts = text.split("\n", 1)
if len(parts) == 2:
first_line, epg_line = parts
if "|" in first_line:
name_resolution = first_line.split("|")
if len(name_resolution) >= 2:
name = name_resolution[0].replace("", "").replace("", "").strip()
resolution = name_resolution[1].strip()
self.update_epg_info(name, resolution, self.current_channel_epg)
def retry_detect_resolution(self, item, base, retries=6):
def try_fps_only():
info = self.player.get_stream_info()
if info and info != "NA":
self.current_channel_resolution = info
self.update_epg_info(self.current_channel_name or base, info)
return
else:
self.current_channel_resolution = "Stream not available"
self.update_epg_info(self.current_channel_name or base, "Stream not available")
if retries > 0:
QTimer.singleShot(10000, lambda: self.retry_detect_resolution(item, base, retries - 1))
try_fps_only()
def set_volume(self, value):
self.player.media_player.audio_set_volume(value)
if value == 0:
self.player.media_player.audio_set_mute(True)
self.mute_button.setChecked(True)
self.mute_button.setText("๐")
else:
self.player.media_player.audio_set_mute(False)
self.mute_button.setChecked(False)
self.mute_button.setText("๐")
def setup_epg_auto_refresh(self):
now = datetime.now()
# Align to next :00 or :30
if now.minute < 30:
next_run = now.replace(minute=30, second=0, microsecond=0)
else:
next_run = now.replace(hour=now.hour + 1 if now.hour < 23 else 0, minute=0, second=0, microsecond=0)
delay_ms = int((next_run - now).total_seconds() * 1000)
QTimer.singleShot(delay_ms, self.start_half_hour_timer)
def show_status_message(self, msg):
self.epg_display.setText(f"{msg}
")
def start_half_hour_timer(self):
self.update_current_epg() # Update EPG data
self.update_epg_info(self.current_channel_name, self.current_channel_resolution)
# Start timer for next 30-minute interval
self._epg_timer = QTimer()
self._epg_timer.timeout.connect(self.trigger_epg_update)
self._epg_timer.start(30 * 60 * 1000) # every 30 mins
def toggle_epg_grid(self):
if self.epg_grid_window is None:
# Use dummy XML if EPG not ready yet
dummy_epg = self.epg_data if self.epg_data is not None else ET.Element("tv")
self.epg_grid_window = EPGGridWindow(dummy_epg, self.channel_order, self.channel_video_info, self)
if self.epg_grid_window.isVisible():
self.epg_grid_window.hide()
else:
self.epg_grid_window.show()
def toggle_fullscreen(self, event=None):
if not self._is_fullscreen:
self._normal_geometry = self.geometry()
self._normal_flags = self.windowFlags()
self.channel_list.hide()
self.epg_display.hide()
self.controls.hide()
self.setWindowFlags(self._normal_flags | Qt.FramelessWindowHint)
self.showFullScreen()
else:
self.setWindowFlags(self._normal_flags)
self.showNormal()
if self._normal_geometry:
self.setGeometry(self._normal_geometry)
self.channel_list.show()
self.epg_display.show()
self.controls.show()
self._is_fullscreen = not self._is_fullscreen
def toggle_mute(self):
is_muted = self.player.media_player.audio_get_mute()
self.player.media_player.audio_set_mute(not is_muted)
self.mute_button.setText("๐" if not is_muted else "๐")
self.mute_button.setChecked(not is_muted)
def toggle_play_pause(self):
mp = self.player.media_player
if mp.is_playing():
mp.pause()
else:
mp.play()
def trigger_epg_update(self):
print("[EPG] Auto-updating at 30-minute boundary")
self.update_current_epg()
self.update_epg_info(self.current_channel_name, self.current_channel_resolution)
self.scroll_to_current_time() # If you want to auto-scroll like the grid does
def update_current_epg(self, epg_data=None):
if epg_data is not None:
self.epg_data = epg_data
if self.epg_data is None or not hasattr(self, 'current_channel_id'):
return
try:
# Get current LOCAL time in Asia/Karachi
local_tz = pytz.timezone("Asia/Karachi")
now_local = datetime.now(local_tz)
current_prog = None
next_prog = None
for programme in EPGStore.get().findall(".//programme"):
if programme.get("channel") != self.current_channel_id:
continue
try:
# Parse UTC times from EPG
start_utc = datetime.strptime(programme.get("start")[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
stop_utc = datetime.strptime(programme.get("stop")[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
# Convert to local time
start_local = start_utc.astimezone(local_tz)
stop_local = stop_utc.astimezone(local_tz)
if start_local <= now_local < stop_local:
current_prog = programme
elif start_local > now_local:
next_prog = programme
break # EPG is sorted by time, so first match is sufficient
except Exception as e:
print(f"[EPG Parse Error] Failed to parse start/stop: {e}")
continue
# Set current program
if current_prog is not None:
self.current_channel_epg = {
'title': unquote(current_prog.findtext('title', 'No Programme Information')),
'start': current_prog.get('start'),
'stop': current_prog.get('stop'),
'desc': unquote(current_prog.findtext('desc', ''))
}
else:
self.current_channel_epg = None
# Set next program
if next_prog is not None:
self.next_program_epg = {
'title': unquote(next_prog.findtext('title', 'No information')),
'start': next_prog.get('start'),
'stop': next_prog.get('stop'),
'desc': unquote(next_prog.findtext('desc', ''))
}
else:
self.next_program_epg = None
# โ
If current program expired, force immediate display update
if self.current_channel_epg:
stop_time_str = self.current_channel_epg.get('stop')
try:
stop_time_utc = datetime.strptime(stop_time_str[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
stop_time_local = stop_time_utc.astimezone(local_tz)
if now_local >= stop_time_local:
if hasattr(self, 'current_channel_name') and hasattr(self, 'current_channel_resolution'):
self.update_epg_info(self.current_channel_name, self.current_channel_resolution)
except Exception as e:
print(f"[EPG Stop Check] Failed to parse stop time: {e}")
except Exception as e:
print(f"Error updating current EPG: {str(e)}")
def update_epg_data(self):
if not self.default_epg_urls:
self.log_debug("No default EPG URLs available")
return
# Clear previous EPG before bulk reload
EPGStore.clear()
for url in self.default_epg_urls:
self.log_debug(f"Downloading EPG from: {url}")
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
self.process_epg_data(response.content)
except Exception as e:
self.log_debug(f"[EPG] Failed to download from {url}: {e}")
self.setup_epg_auto_refresh()
def update_epg_info(self, name=None, resolution=None):
if not hasattr(self, 'current_channel_name'):
return
# Create formatted EPG text
epg_text = ""
# Channel name (top left) and resolution (top right)
channel_font = QFont()
channel_font.setPointSize(14)
channel_font.setBold(True)
epg_text += ""
epg_text += f"| {name} | {resolution} |
"
epg_text += "
"
# Current program
if self.current_channel_epg:
# Title (big and bold)
epg_text += f""
epg_text += f"{self.current_channel_epg.get('title', 'No Programme Information')}"
epg_text += "
"
# Time and duration
if 'start' in self.current_channel_epg and 'stop' in self.current_channel_epg:
start = self.current_channel_epg['start']
stop = self.current_channel_epg['stop']
try:
# Convert GMT times to aware datetime objects
gmt_start = datetime.strptime(start[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
gmt_stop = datetime.strptime(stop[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
# Convert to local time (Pakistan time here, change if needed)
local_tz = pytz.timezone("Asia/Karachi")
start_local = gmt_start.astimezone(local_tz)
stop_local = gmt_stop.astimezone(local_tz)
now_local = datetime.now(local_tz)
# Format time display
start_time_str = start_local.strftime("%I:%M%p").lstrip('0')
stop_time_str = stop_local.strftime("%I:%M%p").lstrip('0')
duration = int((stop_local - start_local).total_seconds() // 60)
# Progress percentage
total_seconds = (stop_local - start_local).total_seconds()
elapsed_seconds = (now_local - start_local).total_seconds()
progress_percent = max(0, min(100, int((elapsed_seconds / total_seconds) * 100))) if total_seconds > 0 else 0
# Build HTML progress bar
progress_bar_html = f"""
"""
# Add everything to epg_text
epg_text += f""
epg_text += f"{start_time_str} - {stop_time_str} ({duration} mins)"
epg_text += progress_bar_html
epg_text += "
"
except:
pass
# Description
if 'desc' in self.current_channel_epg and self.current_channel_epg['desc']:
epg_text += f""
epg_text += f"{self.current_channel_epg['desc']}"
epg_text += "
"
# Next program
if self.next_program_epg:
epg_text += f""
epg_text += f"Next: {self.next_program_epg.get('title', 'No information')}"
epg_text += "
"
if 'desc' in self.next_program_epg and self.next_program_epg['desc']:
epg_text += f""
epg_text += f"{self.next_program_epg['desc']}"
epg_text += "
"
if 'start' in self.next_program_epg and 'stop' in self.next_program_epg:
start = self.next_program_epg['start']
stop = self.next_program_epg['stop']
try:
# Convert GMT times to aware datetime objects
gmt_start = datetime.strptime(start[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
gmt_stop = datetime.strptime(stop[:14], "%Y%m%d%H%M%S").replace(tzinfo=timezone.utc)
# Convert to local time (Pakistan time here, change if needed)
local_tz = pytz.timezone("Asia/Karachi")
start_local = gmt_start.astimezone(local_tz)
stop_local = gmt_stop.astimezone(local_tz)
# Format time display
start_time_str = start_local.strftime("%I:%M%p").lstrip('0')
stop_time_str = stop_local.strftime("%I:%M%p").lstrip('0')
epg_text += f""
epg_text += f"{start_time_str} - {stop_time_str}"
epg_text += "
"
except:
pass
else:
epg_text += ""
epg_text += "No Programme Information"
epg_text += "
"
self.epg_display.setText(epg_text)
class NowLineOverlay(QWidget):
def __init__(self, parent, grid_window):
super().__init__(parent)
self.grid_window = grid_window
self.slot_width = grid_window.timeline_slot_width # โ
store slot width
self.setAttribute(Qt.WA_TransparentForMouseEvents)
self.setStyleSheet("background: transparent;")
self.setVisible(True) # โ
Ensure it's visible
def paintEvent(self, event):
from PyQt5.QtGui import QPainter, QPen
painter = QPainter(self)
pen = QPen(Qt.red, 2)
painter.setPen(pen)
now = datetime.now(pytz.timezone("Asia/Karachi"))
delta = now - self.grid_window.timeline_start
adjusted_seconds = delta.total_seconds() + RED_LINE_SHIFT_SECONDS
block_width = self.grid_window.timeline_slot_width # should be 200 if set correctly
# Total X position of the red line
x = (adjusted_seconds / 1800) * block_width
painter.drawLine(int(x), 0, int(x), self.height())
class TimelineLabel(QLabel):
def __init__(self, text, start_offset, duration, parent_grid, timeline_slot_width):
super().__init__()
self.parent_grid = parent_grid
self.full_text = text
self.start_offset = start_offset
self.duration = duration
self.parent_grid = parent_grid
self.timeline_slot_width = timeline_slot_width
self.setStyleSheet("""
border: 1px solid #555;
color: white;
""")
self.setWordWrap(True) # Allow multi-line
self.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.setText(self.full_text) # Let QLabel handle the text
self.setStyleSheet("""
border: 1px solid #555;
color: white;
font-size: 10pt;
padding: 2px; /* reduced padding */
background-color: #333;
""")
self.setStyleSheet("border: 1px solid #444; padding: 2px; background-color: #333;")
def mouseDoubleClickEvent(self, event):
print("[DEBUG] Program grid double-click detected")
if self.parent_grid and self.parent_grid.main_window:
layout = self.parent_grid.epg_layout
index = layout.indexOf(self)
if index != -1:
row, _, _, _ = layout.getItemPosition(index)
print(f"[DEBUG] EPG row double-clicked: {row}")
if 0 <= row < self.parent_grid.main_window.channel_list.count():
item = self.parent_grid.main_window.channel_list.item(row)
print(f"[DEBUG] Attempting to play channel from program grid: {item.text()}")
self.parent_grid.main_window.channel_list.setCurrentItem(item)
self.parent_grid.main_window.on_channel_clicked(item)
event.accept()
class VideoFrame(QFrame):
def __init__(self, parent=None):
super().__init__(parent)
self.setStyleSheet("background-color: black;")
class VLCPlayer:
def __init__(self, video_frame):
self.instance = vlc.Instance()
self.media_player = self.instance.media_player_new()
if sys.platform == "win32":
self.media_player.set_hwnd(int(video_frame.winId()))
else:
self.media_player.set_xwindow(int(video_frame.winId()))
self.playback_timeout_timer = QTimer()
self.playback_timeout_timer.setSingleShot(True)
self.playback_timeout_timer.timeout.connect(self.check_playback_timeout)
def check_playback_timeout(self):
if not self.media_player.is_playing():
print("[VLC] Stream timeout โ stopping playback")
self.media_player.stop()
def play_stream(self, url):
try:
self.media_player.stop() # Make sure it's fully stopped before new media
except Exception:
pass
media = self.instance.media_new(url)
# โ
Optimized for faster startup
media.add_option(":network-caching=1000")
media.add_option(":file-caching=500")
media.add_option(":tcp-caching=500")
media.add_option(":avcodec-hw=any") # โ
Try hardware decoding when possible
media.add_option("--no-drop-late-frames")
media.add_option("--no-skip-frames")
media.add_option(":http-reconnect")
#media.add_option(":http-continuous")
media.add_option(":ffmpeg-skiploopfilter=all")
media.add_option(":ffmpeg-skipframe=default")
self.media_player.set_media(media)
self.media_player.audio_set_volume(70)
self.media_player.play()
# Start timeout safety
self.playback_timeout_timer.start(10000)
def release(self):
try:
self.media_player.release()
except Exception: pass
try:
self.instance.release()
except Exception: pass
def stop(self):
self.media_player.stop()
def get_logo_path(url):
"""Returns local cache path for a logo URL"""
hashname = hashlib.md5(url.encode('utf-8')).hexdigest()
return os.path.join(CACHE_DIR, f"{hashname}.png")
def load_logo(url):
"""Returns QPixmap (cached or downloaded later)"""
path = get_logo_path(url)
if os.path.exists(path):
return QPixmap(path)
# Instead of blocking download, skip if not in cache
return None # Later, we can trigger background fetch
def monitor_php_server(php_dir):
global php_proc
if php_proc and php_proc.poll() is not None:
print("[WARNING] PHP server exited. Restarting...")
start_php_server(php_dir)
def restart_php_server(php_dir):
global php_proc
try:
if php_proc:
print("[PHP] Restarting embedded PHP server...")
php_proc.terminate()
php_proc.wait(timeout=2)
except Exception as e:
print(f"[PHP] Error terminating PHP server: {e}")
# Start again
php_proc = start_php_server(php_dir)
def start_php_server(php_dir, port=8888):
global php_proc
try:
os.chdir(php_dir)
php_proc = subprocess.Popen(
["php", "-S", f"localhost:{port}"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
)
return php_proc
except Exception as e:
print(f"Failed to start PHP server: {e}")
return None
def wait_for_php_server(host='localhost', port=8888, timeout=3):
"""Wait until PHP server is accepting connections."""
start_time = time.time()
while time.time() - start_time < timeout:
try:
with socket.create_connection((host, port), timeout=0.5):
return True
except (ConnectionRefusedError, socket.timeout, OSError):
time.sleep(0.1)
return False
if __name__ == "__main__":
php_dir = os.path.join(os.path.dirname(__file__), "php_files")
os.makedirs(php_dir, exist_ok=True)
php_proc = start_php_server(php_dir)
app = QApplication(sys.argv)
window = IPTVWindow(php_dir)
# โ
Add this backup cleanup line
app.aboutToQuit.connect(window.close)
window.show()
# Now create and show EPG window AFTER main window is visible
if window.epg_grid_window is None:
window.epg_grid_window = EPGGridWindow(
window.epg_data,
window.channel_order,
window.channel_video_info,
window,
channel_logos=window.channel_logos
)
window.epg_grid_window.show()
# โ
Move PHP monitor timer here BEFORE app.exec_()
php_timer = QTimer()
php_timer.timeout.connect(lambda: monitor_php_server(php_dir))
php_timer.start(15000)
# โ
Debug active threads after 1 second
QTimer.singleShot(1000, lambda: print(f"[DEBUG] Threads at 1s: {[t.name for t in threading.enumerate()]}"))
if platform.system() == "Windows":
try:
import ctypes
# Set Per Monitor v2 DPI Awareness
awareness_context = 3 # DPI_AWARENESS_CONTEXT_PER_MONITOR_AWARE_V2
ctypes.windll.user32.SetProcessDpiAwarenessContext(ctypes.c_void_p(awareness_context))
print("[DPI] Enabled Per-Monitor V2 DPI Awareness")
except Exception as e:
print("[DPI] DPI awareness setting failed:", e)
exit_code = app.exec_()
# Stop PHP if still running
if php_proc:
php_proc.terminate()
php_proc.wait(2)
sys.exit(exit_code)