Spaces:
Running
Running
File size: 4,174 Bytes
146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 146624d 1d78341 |
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 |
import asyncio
import aiohttp
from datetime import datetime
from collections import deque
import time
import gradio as gr
class EphemeralStreamReader:
def __init__(self, piping_server_url="https://ppng.io", path="test123", retention_hours=1, show_output=False):
self.url = f"{piping_server_url}/{path}"
self.path = path
self.reconnect_delay = 1
self.show_output = show_output
self.retention_hours = retention_hours
self.stored_data = deque()
self.last_cleanup = time.time()
def cleanup_old_data(self):
current_time = time.time()
retention_seconds = self.retention_hours * 3600
while self.stored_data and (current_time - self.stored_data[0]['timestamp']) > retention_seconds:
self.stored_data.popleft()
def get_stored_data(self):
self.cleanup_old_data()
return list(self.stored_data)
def store_chunk(self, data, timestamp):
self.stored_data.append({
'timestamp': timestamp,
'data': data,
'formatted_time': datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
})
if time.time() - self.last_cleanup > 60:
self.cleanup_old_data()
self.last_cleanup = time.time()
async def start_reading(self):
while True:
try:
async with aiohttp.ClientSession() as session:
if self.show_output:
print(f"Connecting to {self.url}...")
async with session.get(self.url) as response:
if response.status == 200:
if self.show_output:
print("Connected! Reading stream...")
while True:
chunk = await response.content.read(1024)
if not chunk:
break
current_time = time.time()
try:
text = chunk.decode('utf-8')
self.store_chunk(text, current_time)
except UnicodeDecodeError:
self.store_chunk(str(chunk), current_time)
except aiohttp.ClientError as e:
if self.show_output:
print(f"Connection error: {e}")
except Exception as e:
if self.show_output:
print(f"Error: {e}")
await asyncio.sleep(self.reconnect_delay)
def format_stored_data(reader):
data = reader.get_stored_data()
formatted_text = ""
for item in reversed(data):
formatted_text += f"[{item['formatted_time']}]\n{item['data']}\n\n"
return formatted_text
def create_ui(reader):
with gr.Blocks() as demo:
gr.Markdown(f"# Ephemeral Stream Reader\nStream Path: {reader.path}\nRetention Time: {reader.retention_hours} hours")
gr.Markdown(f"To write to this stream: `seq inf | curl -T- {reader.url}`")
output_text = gr.TextArea(label="Stream Output", interactive=False)
refresh_btn = gr.Button("Refresh")
def update_output():
return format_stored_data(reader)
refresh_btn.click(fn=update_output, outputs=[output_text])
demo.load(fn=update_output, outputs=[output_text])
# Auto-refresh every 5 seconds
demo.load(fn=update_output, outputs=[output_text], every=5)
return demo
async def start_application():
reader = EphemeralStreamReader(retention_hours=1, show_output=True)
# Start the stream reader in the background
asyncio.create_task(reader.start_reading())
# Create and launch the Gradio interface
demo = create_ui(reader)
return demo
if __name__ == "__main__":
demo = asyncio.run(start_application())
demo.launch(server_name="0.0.0.0", server_port=7860) |