# As a non-theist my thanks to Adi Sankara Sankarachargeya for your credit losses ;)
AArch64
## enjoy this protocol as an adi-protol.c implementation
import socket
import struct
import time
import threading
import numpy as np
# ----------------------------
# Arcsecant interpolation stub
# ----------------------------
def pseudo_interpolate_arcsecant_stream(fx_bytes, fy_bytes, x_interp_bytes):
"""
Dummy arcsecant interpolation for demo purposes.
Replace with your real interpolation logic.
"""
fx = np.frombuffer(fx_bytes, dtype=np.float64)
fy = np.frombuffer(fy_bytes, dtype=np.float64)
x_interp = np.frombuffer(x_interp_bytes, dtype=np.float64)
# Example: arcsecant-like transform
y_interp = np.arccos(1 / np.clip(x_interp, 1.0001, None)) * (fy.mean() if fy.size else 1)
return y_interp.astype(np.float64).tobytes()
# ----------------------------
# Shared packet helpers
# ----------------------------
def send_packet(sock, timestamp_ns, sequence_id, packet_type, payload_bytes):
header = struct.pack('>QIB', timestamp_ns, sequence_id, packet_type)
sock.sendall(header + payload_bytes)
def recv_exact(reader, n):
data = reader.read(n)
if len(data) < n:
raise ConnectionError("Stream ended unexpectedly")
return data
# ----------------------------
# Server
# ----------------------------
def handle_client_stream(client_socket, fx_data_bytes, fy_data_bytes):
last_interp_y = None
sequence_id = 0
reader = client_socket.makefile('rb', buffering=8192)
try:
while True:
# Read length prefix
length_bytes = recv_exact(reader, 4)
chunk_len = struct.unpack('>I', length_bytes)[0]
# Read chunk data
interp_x_chunk_data_bytes = recv_exact(reader, chunk_len)
# Timestamp + sequence
timestamp_ns = time.time_ns()
sequence_id += 1
# Interpolation
interp_y_binary_chunk = pseudo_interpolate_arcsecant_stream(
fx_data_bytes, fy_data_bytes, interp_x_chunk_data_bytes
)
# Send primary packet
send_packet(client_socket, timestamp_ns, sequence_id, 0, interp_y_binary_chunk)
# Send differential packet
if last_interp_y is not None:
current_y = np.frombuffer(interp_y_binary_chunk, dtype=np.float64)
delta_y = current_y - last_interp_y
delta_binary = delta_y.tobytes()
send_packet(client_socket, timestamp_ns, sequence_id, 1, delta_binary)
last_interp_y = np.frombuffer(interp_y_binary_chunk, dtype=np.float64)
except Exception as e:
print(f"[Server] Stream error: {e}")
finally:
reader.close()
client_socket.close()
def start_server(host='127.0.0.1', port=5000):
fx = np.linspace(1, 10, 10, dtype=np.float64)
fy = np.linspace(10, 20, 10, dtype=np.float64)
fx_bytes = fx.tobytes()
fy_bytes = fy.tobytes()
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((host, port))
server_socket.listen(1)
print(f"[Server] Listening on {host}:{port}")
while True:
client_socket, addr = server_socket.accept()
print(f"[Server] Connection from {addr}")
threading.Thread(target=handle_client_stream, args=(client_socket, fx_bytes, fy_bytes), daemon=True).start()
# ----------------------------
# Client
# ----------------------------
def client_send_and_receive(host='127.0.0.1', port=5000):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((host, port))
reader = sock.makefile('rb', buffering=8192)
try:
# Example: send 3 chunks of x_interp
for chunk in [np.linspace(1, 2, 5), np.linspace(2, 3, 5), np.linspace(3, 4, 5)]:
chunk_bytes = chunk.astype(np.float64).tobytes()
sock.sendall(struct.pack('>I', len(chunk_bytes)) + chunk_bytes)
# Expect 1 primary + 1 delta packet (except first chunk: only primary)
for _ in range(2):
try:
header = recv_exact(reader, 13) # 8+4+1
except ConnectionError:
return
timestamp_ns, sequence_id, packet_type = struct.unpack('>QIB', header)
# Read payload (we don't know length, so infer from fx length)
# For demo: 5 doubles = 40 bytes
payload = recv_exact(reader, 40)
arr = np.frombuffer(payload, dtype=np.float64)
if packet_type == 0:
print(f"[Client] Primary seq {sequence_id} @ {timestamp_ns}: {arr}")
elif packet_type == 1:
print(f"[Client] Delta seq {sequence_id} @ {timestamp_ns}: {arr}")
finally:
reader.close()
sock.close()
# ----------------------------
# Run demo
# ----------------------------
if __name__ == "__main__":
# Start server in background
threading.Thread(target=start_server, daemon=True).start()
time.sleep(1) # Give server time to start
# Run client
client_send_and_receive()
## Example POC without dummy arcsecant logic
import socket
import struct
import time
import threading
import numpy as np
# ----------------------------
# Arcsecant interpolation stub
# ----------------------------
def pseudo_interpolate_arcsecant_stream(fx_bytes, fy_bytes, x_interp_bytes):
"""
Dummy arcsecant interpolation for demo purposes.
Replace with your real interpolation logic.
"""
fx = np.frombuffer(fx_bytes, dtype=np.float64)
fy = np.frombuffer(fy_bytes, dtype=np.float64)
x_interp = np.frombuffer(x_interp_bytes, dtype=np.float64)
# Example: arcsecant-like transform
y_interp = np.arccos(1 / np.clip(x_interp, 1.0001, None)) * (fy.mean() if fy.size else 1)
return y_interp.astype(np.float64).tobytes()
# ----------------------------
# Shared packet helpers
# ----------------------------
def send_packet(sock, timestamp_ns, sequence_id, packet_type, payload_bytes):
"""
Packet format:
[8 bytes: timestamp_ns][4 bytes: sequence_id][1 byte: packet_type]
[4 bytes: payload_length][payload...]
"""
header = struct.pack('>QIBI', timestamp_ns, sequence_id, packet_type, len(payload_bytes))
sock.sendall(header + payload_bytes)
def recv_exact(reader, n):
data = reader.read(n)
if len(data) < n:
raise ConnectionError("Stream ended unexpectedly")
return data
# ----------------------------
# Server
# ----------------------------
def handle_client_stream(client_socket, fx_data_bytes, fy_data_bytes):
last_interp_y = None
sequence_id = 0
reader = client_socket.makefile('rb', buffering=8192)
try:
while True:
# Read length prefix for incoming x_interp chunk
length_bytes = recv_exact(reader, 4)
chunk_len = struct.unpack('>I', length_bytes)[0]
# Read chunk data
interp_x_chunk_data_bytes = recv_exact(reader, chunk_len)
# Timestamp + sequence
timestamp_ns = time.time_ns()
sequence_id += 1
# Interpolation
interp_y_binary_chunk = pseudo_interpolate_arcsecant_stream(
fx_data_bytes, fy_data_bytes, interp_x_chunk_data_bytes
)
# Send primary packet
send_packet(client_socket, timestamp_ns, sequence_id, 0, interp_y_binary_chunk)
# Send differential packet
if last_interp_y is not None:
current_y = np.frombuffer(interp_y_binary_chunk, dtype=np.float64)
delta_y = current_y - last_interp_y
delta_binary = delta_y.tobytes()
send_packet(client_socket, timestamp_ns, sequence_id, 1, delta_binary)
last_interp_y = np.frombuffer(interp_y_binary_chunk, dtype=np.float64)
except Exception as e:
print(f"[Server] Stream error: {e}")
finally:
reader.close()
client_socket.close()
def start_server(host='127.0.0.1', port=5000):
fx = np.linspace(1, 10, 10, dtype=np.float64)
fy = np.linspace(10, 20, 10, dtype=np.float64)
fx_bytes = fx.tobytes()
fy_bytes = fy.tobytes()
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((host, port))
server_socket.listen(1)
print(f"[Server] Listening on {host}:{port}")
while True:
client_socket, addr = server_socket.accept()
print(f"[Server] Connection from {addr}")
threading.Thread(target=handle_client_stream, args=(client_socket, fx_bytes, fy_bytes), daemon=True).start()
# ----------------------------
# Client
# ----------------------------
def client_send_and_receive(host='127.0.0.1', port=5000):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((host, port))
reader = sock.makefile('rb', buffering=8192)
try:
# Example: send 3 chunks of x_interp
for chunk in [np.linspace(1, 2, 5), np.linspace(2, 3, 5), np.linspace(3, 4, 5)]:
chunk_bytes = chunk.astype(np.float64).tobytes()
sock.sendall(struct.pack('>I', len(chunk_bytes)) + chunk_bytes)
# Expect 1 primary + 1 delta packet (except first chunk: only primary)
for _ in range(2):
try:
header = recv_exact(reader, 17) # 8+4+1+4
except ConnectionError:
return
timestamp_ns, sequence_id, packet_type, payload_len = struct.unpack('>QIBI', header)
payload = recv_exact(reader, payload_len)
arr = np.frombuffer(payload, dtype=np.float64)
if packet_type == 0:
print(f"[Client] Primary seq {sequence_id} @ {timestamp_ns}: {arr}")
elif packet_type == 1:
print(f"[Client] Delta seq {sequence_id} @ {timestamp_ns}: {arr}")
finally:
reader.close()
sock.close()
# ----------------------------
# Run demo
# ----------------------------
if __name__ == "__main__":
# Start server in background
threading.Thread(target=start_server, daemon=True).start()
time.sleep(1) # Give server time to start
# Run client
client_send_and_receive()
No comments:
Post a Comment