Spaces:

Greg-House
/

Simple-piping

Paused

App Files Files Community

Greg-House commited on Nov 14, 2024

Commit

146624d

verified ·

1 Parent(s): 76f3cde

Create app.py

Browse files

Files changed (1) hide show

app.py +202 -0

app.py ADDED Viewed

	@@ -0,0 +1,202 @@

+import asyncio
+import aiohttp
+import time
+import ssl
+from datetime import datetime
+from collections import deque
+import json
+from pathlib import Path
+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)
+                                    if self.show_output:
+                                        timestamp = datetime.fromtimestamp(current_time).strftime('%Y-%m-%d %H:%M:%S')
+                                        print(f"[{timestamp}] Received: {text}", end='', flush=True)
+                                except UnicodeDecodeError:
+                                    self.store_chunk(str(chunk), current_time)
+                                    if self.show_output:
+                                        print(f"Received raw bytes: {chunk}")
+            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}")
+            if self.show_output:
+                print(f"Connection closed. Reconnecting in {self.reconnect_delay} seconds...")
+            await asyncio.sleep(self.reconnect_delay)
+    async def start_http_server(self, host='localhost', port=8000, ssl_context=None):
+        from aiohttp import web
+        async def handle_get(request):
+            data = self.get_stored_data()
+            html = f"""
+            <!DOCTYPE html>
+            <html>
+            <head>
+                <title>Ephemeral Stream Data</title>
+                <style>
+                    body {{
+                        font-family: Arial, sans-serif;
+                        margin: 20px;
+                    }}
+                    .data-container {{
+                        margin: 20px 0;
+                    }}
+                    .timestamp {{
+                        color: #666;
+                        font-size: 0.9em;
+                    }}
+                    .data-item {{
+                        margin: 10px 0;
+                        padding: 10px;
+                        background: #f5f5f5;
+                        border-radius: 4px;
+                    }}
+                    .refresh-btn {{
+                        padding: 10px 20px;
+                        background: #4CAF50;
+                        color: white;
+                        border: none;
+                        border-radius: 4px;
+                        cursor: pointer;
+                    }}
+                    .info {{
+                        margin-bottom: 20px;
+                        color: #666;
+                    }}
+                </style>
+                <script>
+                    function refreshData() {{
+                        window.location.reload();
+                    }}
+                </script>
+            </head>
+            <body>
+                <h1>Ephemeral Stream Data</h1>
+                <div class="info">
+                    <p>Stream Path: {self.path}</p>
+                    <p>Retention Time: {self.retention_hours} hours</p>
+                    <p>To write to this stream: <code>seq inf | curl -T- {self.url}</code></p>
+                </div>
+                <button class="refresh-btn" onclick="refreshData()">Refresh Data</button>
+                <div class="data-container">
+            """
+            for item in reversed(data):
+                html += f"""
+                    <div class="data-item">
+                        <div class="timestamp">{item['formatted_time']}</div>
+                        <pre>{item['data']}</pre>
+                    </div>
+                """
+            html += """
+                </div>
+            </body>
+            </html>
+            """
+            return web.Response(text=html, content_type='text/html')
+        app = web.Application()
+        app.router.add_get('/', handle_get)
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.TCPSite(runner, host, port, ssl_context=ssl_context)
+        await site.start()
+        protocol = "https" if ssl_context else "http"
+        print(f"\nWeb interface available at {protocol}://{host}:{port}")
+        print(f"To write to this stream: seq inf | curl -T- {self.url}")
+def create_ssl_context():
+    """Create SSL context from certificate and key files"""
+    cert_path = Path("./localhost.pem")
+    key_path = Path("./localhost-key.pem")
+    if not cert_path.exists() or not key_path.exists():
+        print("Warning: SSL certificate or key not found. Running in HTTP mode.")
+        return None
+    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    ssl_context.load_cert_chain(str(cert_path), str(key_path))
+    return ssl_context
+async def main():
+    ssl_context = create_ssl_context()
+    reader = EphemeralStreamReader(retention_hours=1, show_output=False)
+    await asyncio.gather(
+        reader.start_reading(),
+        reader.start_http_server(host='localhost', port=8000, ssl_context=ssl_context)
+    )
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description='Ephemeral Stream Reader with HTTPS support')
+    parser.add_argument('--host', default='localhost', help='Host to bind to')
+    parser.add_argument('--port', type=int, default=8000, help='Port to bind to')
+    parser.add_argument('--path', default='test123', help='Path for the stream')
+    args = parser.parse_args()
+    print("Starting Ephemeral Stream Reader with HTTPS support...")
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("\nStopping server...")