import numpy as np
import matplotlib.pyplot as plt
import pubchempy as pcp
import requests
import json
import time
import re
import warnings
import socket
import struct
import threading
import traceback
import logging
import argparse
# Suppress Astroquery's InsecureRequestWarning when connecting to some services
from requests.packages.urllib3.exceptions import InsecureRequestWarning
warnings.simplefilter('ignore', InsecureRequestWarning)
# --- Library Imports for Astrophysics and Data ---
# This block handles imports for the client-side astrophysics application.
from pystac_client import Client
from datetime import datetime
from astroquery.gaia import Gaia
from astroquery.cadc import Cadc
from astroquery.utils.tap import TapPlus
from astroquery.ned import Ned
from astropy import units as u
from astropy.coordinates import SkyCoord
from astropy.table import Table
import nmrglue as ng # Used by the server for NMR data handling
# --- Core Scientific & Mathematical Constants ---
# This section consolidates constants from all programs.
# Physics Constants (from astro.py)
CURIE_TO_BQ = 3.7e10
LN2 = np.log(2)
STEFAN_BOLTZMANN_CONSTANT = 5.670374419e-8
PLANCK_CONSTANT = 6.62607015e-34
BOLTZMANN_CONSTANT = 1.380649e-23
C = 299792458
HBAR = 1.054571817e-34
LAMBDA_C = HBAR / (np.pi * C)
# Networking & Operation Codes (from n-dim.py and NMRBBS.py)
OPERATION_INTERPOLATE = 0
OPERATION_DIFFERENTIATE = 1
OPERATION_CALCULATE_GRADIENT_1D = 2
OPERATION_HYPERBOLIC_INTERCEPT_HANDLER = 3
OPERATION_INTEGRATE = 4
OPERATION_INTEGRATE_ND = 5
OPERATION_WORKFLOW = 6
SERVER_HOST = '127.0.0.1'
SERVER_PORT = 12345
# Server Message Codes (from NMRBBS.py)
MSG_TYPE_INTERPOLATION_RESULT = 5
MSG_TYPE_DIFFERENTIATION_RESULT = 12
MSG_TYPE_CLIENT_DISCONNECT = 254
MSG_TYPE_ID_ASSIGNMENT = 255
# --- Shared Networking & Data Handling Suite ---
# These helper functions are crucial for communication between the client and server.
# They are used by both the `Server` class and the client's `execute_workflow` function.
def _sendall_data(sock, data_array):
"""Helper to send a numpy array's bytes."""
data_bytes = data_array.astype(np.float32).tobytes()
sock.sendall(struct.pack('!I', len(data_bytes)))
sock.sendall(data_bytes)
def _recvall(sock, n):
"""Helper function to reliably receive exactly 'n' bytes from a socket."""
data = b''
while len(data) < n:
packet = sock.recv(n - len(data))
if not packet:
return None
data += packet
return data
def _recvall_data(sock):
"""Helper to receive a numpy array's bytes or an error message."""
data_len_bytes = _recvall(sock, 4)
if data_len_bytes is None:
raise ConnectionResetError("Incomplete data (length) received from server.")
data_len = struct.unpack('!I', data_len_bytes)[0]
data_bytes = _recvall(sock, data_len)
if data_bytes is None:
raise ConnectionResetError("Incomplete data (content) received from server.")
try:
error_message = data_bytes.decode('utf-8')
if "Error" in error_message or "Server internal error" in error_message:
raise RuntimeError(f"Server returned an error: {error_message}")
except UnicodeDecodeError:
pass
return np.array(struct.unpack(f'!{data_len // 4}f', data_bytes), dtype=np.float32)
# --- Server-Side Mathematical & Data Services Suite ---
# This class contains the logic that runs when the program is started in server mode.
# It handles requests for complex mathematical operations and data broadcasting.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class Server:
def __init__(self, host: str, port: int):
self.host = host
self.port = port
self.clients: list[dict] = []
self.clients_lock = threading.Lock()
self.client_data: dict[socket.socket, dict] = {}
self.client_data_lock = threading.Lock()
self.next_client_id = 0
def pseudo_interpolate_arcsecant_1d_triple(self, fx: np.ndarray, fy: np.ndarray, fz: np.ndarray, x_interp: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""
Performs a custom, non-linear interpolation on 1D data with y and z datasets.
This function uses a simple sine-based easing function to approximate the non-linear
interpolation intended in the original code, avoiding common mathematical errors.
"""
if not (len(fx) == len(fy) == len(fz)) or len(fx) < 2:
raise ValueError("X, Y, and Z data must have equal length and at least two points.")
interp_y = np.zeros_like(x_interp, dtype=float)
interp_z = np.zeros_like(x_interp, dtype=float)
for i, x in enumerate(x_interp):
if x <= fx[0]:
idx1, idx2 = 0, 1
elif x >= fx[-1]:
idx1, idx2 = len(fx) - 2, len(fx) - 1
else:
idx1 = np.searchsorted(fx, x, side='right') - 1
idx2 = idx1 + 1
x1, x2 = fx[idx1], fx[idx2]
y1, y2 = fy[idx1], fy[idx2]
z1, z2 = fz[idx1], fz[idx2]
if (x2 - x1) == 0:
interp_y[i] = y1
interp_z[i] = z1
continue
t_linear = (x - x1) / (x2 - x1)
scaled_t = 0.5 * (1 + np.sin(np.pi * (t_linear - 0.5)))
scaled_t = np.clip(scaled_t, 0, 1)
interp_y[i] = y1 + (y2 - y1) * scaled_t
interp_z[i] = z1 + (z2 - z1) * scaled_t
return interp_y, interp_z
def pseudo_interpolate_arcsecant_nd_triple(self, all_fy_data: list[np.ndarray], all_fz_data: list[np.ndarray], all_fx_data: list[np.ndarray], x_interp: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""
Performs the non-linear interpolation for n-dimensional data by calling
the 1D interpolation function for each dimension.
"""
all_interp_y = []
all_interp_z = []
num_dimensions = len(all_fy_data)
if not (len(all_fx_data) == num_dimensions == len(all_fz_data)):
raise ValueError("The number of x, y, and z data arrays must match across dimensions.")
for i, (fx, fy, fz) in enumerate(zip(all_fx_data, all_fy_data, all_fz_data)):
interp_y, interp_z = self.pseudo_interpolate_arcsecant_1d_triple(fx, fy, fz, x_interp)
all_interp_y.extend(interp_y)
all_interp_z.extend(interp_z)
return np.array(all_interp_y), np.array(all_interp_z)
def numerical_derivative_1d(self, y_values: np.ndarray, x_values: np.ndarray) -> np.ndarray:
"""
Calculates the numerical derivative of a 1D array using a robust gradient method.
"""
if len(y_values) != len(x_values) or len(y_values) < 2:
raise ValueError("Y and X data must have equal length and at least two points for derivative calculation.")
return np.gradient(y_values, x_values)
def differentiate_arcsecant_nd_triple(self, all_fy_data: list[np.ndarray], all_fz_data: list[np.ndarray], all_fx_data: list[np.ndarray], x_eval: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""
Calculates the numerical derivative for n-dimensional data by differentiating
the interpolated data for each dimension.
"""
num_dimensions = len(all_fy_data)
if not (len(all_fx_data) == num_dimensions == len(all_fz_data)):
raise ValueError("The number of x, y, and z data arrays must match.")
if len(x_eval) < 2:
raise ValueError("At least two evaluation points are needed for differentiation.")
all_derivatives_y = []
all_derivatives_z = []
for i, (fx, fy, fz) in enumerate(zip(all_fx_data, all_fy_data, all_fz_data)):
interpolated_y, interpolated_z = self.pseudo_interpolate_arcsecant_1d_triple(fx, fy, fz, x_eval)
derivatives_y = self.numerical_derivative_1d(interpolated_y, x_eval)
derivatives_z = self.numerical_derivative_1d(interpolated_z, x_eval)
all_derivatives_y.extend(derivatives_y)
all_derivatives_z.extend(derivatives_z)
return np.array(all_derivatives_y), np.array(all_derivatives_z)
def handle_client(self, client_socket: socket.socket, client_address: tuple):
"""Main loop for handling client connections."""
logging.info(f"Connection from {client_address}")
try:
# We only expect a single workflow request in this simplified example
op_code_byte = _recvall(client_socket, 1)
if not op_code_byte: return
op_code = struct.unpack('!B', op_code_byte)[0]
if op_code == OPERATION_WORKFLOW:
self.handle_workflow_request(client_socket)
except Exception as e:
logging.error(f"Error handling client {client_address}: {e}", exc_info=True)
finally:
client_socket.close()
def handle_workflow_request(self, client_socket):
"""Processes a workflow request from a client."""
logging.info("Processing workflow request...")
data_len_bytes = _recvall(client_socket, 4)
if not data_len_bytes: return
data_len = struct.unpack('!I', data_len_bytes)[0]
data_bytes = _recvall(client_socket, data_len)
if not data_bytes: return
workflow_steps = json.loads(data_bytes.decode('utf-8'))
# This is where the magic happens: the server processes the client's request.
# It's a simplified version for demonstration.
try:
# For this integrated version, we assume the workflow only contains the
# interpolation and differentiation steps, as defined in the client code.
interpolated_data = None
for step in workflow_steps:
if step["operation_type"] == "INTERPOLATE":
input_data = step["input_data"]
fx = np.array(input_data["fx_data"][0])
fy = np.array(input_data["fy_data"][0])
fz = np.array(input_data["fz_data"][0])
x_interp = np.array(step["parameters"]["x_interp_points"])
interp_y, interp_z = self.pseudo_interpolate_arcsecant_nd_triple([fy], [fz], [fx], x_interp)
interpolated_data = np.concatenate([interp_y, interp_z])
elif step["operation_type"] == "DIFFERENTIATE":
# The client's workflow references the previous step's output
input_data = step["input_data"]
if input_data.get("source_id") == "interpolated_data" and interpolated_data is not None:
# Split the interpolated data back into y and z for differentiation
mid_point = len(interpolated_data) // 2
interp_y = interpolated_data[:mid_point]
interp_z = interpolated_data[mid_point:]
x_eval = np.array(step["parameters"]["x_eval_points"])
# Note: The server's differentiate function handles the rest of the logic
# by re-interpolating and differentiating.
deriv_y, deriv_z = self.differentiate_arcsecant_nd_triple([interp_y], [interp_z], [x_eval, x_eval], x_eval)
# Send back the result
result_array = np.concatenate([deriv_y, deriv_z])
_sendall_data(client_socket, result_array)
logging.info("Successfully processed and sent back differentiation result.")
return
except Exception as e:
error_msg = f"Server internal error: {e}"
logging.error(error_msg, exc_info=True)
try:
client_socket.sendall(struct.pack('!I', len(error_msg.encode('utf-8'))))
client_socket.sendall(error_msg.encode('utf-8'))
except: pass
def run(self):
"""Starts the server and listens for connections."""
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((self.host, self.port))
server_socket.listen(5)
logging.info(f"Server listening on {self.host}:{self.port}")
while True:
try:
client_socket, client_address = server_socket.accept()
client_thread = threading.Thread(target=self.handle_client, args=(client_socket, client_address))
client_thread.daemon = True
client_thread.start()
except KeyboardInterrupt:
logging.info("Server shutting down.")
break
except Exception as e:
logging.error(f"Server accept loop error: {e}")
server_socket.close()
# --- Client-Side Application & Physics Exploration Suite ---
# This section contains the user-facing application logic. When run in client mode,
# it explores various physics concepts and makes requests to the server for calculations.
def execute_workflow(workflow_steps):
"""
Executes a sequence of operations on the server as a single workflow.
This is the bridge between the client's physics exploration and the server's math services.
"""
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
client_socket.connect((SERVER_HOST, SERVER_PORT))
print(f"Connected to server at {SERVER_HOST}:{SERVER_PORT}")
client_socket.sendall(struct.pack('!B', OPERATION_WORKFLOW))
workflow_json = json.dumps(workflow_steps)
workflow_bytes = workflow_json.encode('utf-8')
client_socket.sendall(struct.pack('!I', len(workflow_bytes)))
client_socket.sendall(workflow_bytes)
print("Workflow sent to server. Waiting for result...")
result = _recvall_data(client_socket)
return result
except Exception as e:
print(f"Client error during workflow execution: {e}")
return None
finally:
client_socket.close()
print("Connection closed.")
def _encode_output_with_math(x_data, y_data):
"""
This function sends data to the server for processing.
It takes our data points, smooths them out with a line, and then asks the server
to calculate the slope at different points, which helps us understand how the data is changing.
"""
print("\n--- Applying advanced math to analyze our data via the server ---")
x_eval_points = np.linspace(min(x_data), max(x_data), 10, dtype=np.float32)
workflow = [
{
"operation_type": "INTERPOLATE",
"input_data": {
"type": "direct",
"fx_data": [x_data.tolist()],
"fy_data": [y_data.tolist()],
"fz_data": [[]]
},
"parameters": {
"x_interp_points": x_eval_points.tolist()
},
"output_id": "interpolated_data"
},
{
"operation_type": "DIFFERENTIATE",
"input_data": {
"type": "reference",
"source_id": "interpolated_data"
},
"parameters": {
"x_eval_points": x_eval_points.tolist()
}
}
]
result = execute_workflow(workflow)
if result is not None:
print("Encoded Output (The 'rate of change' of our data from the server):")
print(result)
return result
return None
def analyze_fundamental_physics():
"""Explores fundamental physics concepts and uses the math server for analysis."""
print("\n--- Exploring Foundational Physics Concepts ---")
mass_kg = 1.0
mass_energy = mass_kg * C**2
print(f"A single kilogram of matter contains an immense amount of energy: {mass_energy:.4e} Joules.")
T_surface = 5778
solar_radiation = STEFAN_BOLTZMANN_CONSTANT * T_surface**4
print(f"The Sun's surface at {T_surface} Kelvin emits energy at a rate of: {solar_radiation:.4e} Watts per square meter.")
x_data = np.array([1, 2, 3, 4, 5], dtype=np.float32)
y_data = np.array([10, 12, 15, 19, 25], dtype=np.float32)
_encode_output_with_math(x_data, y_data)
def explore_cosmic_stability():
"""Analyzes the stability of cosmic objects."""
print("\n--- Analyzing the Stability of Cosmic Objects ---")
def _check_stability(object_type, stability_data):
"""Checks if a cosmic object is stable based on its data."""
print(f"Checking the stability of a '{object_type}'...")
for point, value in stability_data.items():
if value < 0:
print(f" Alert: Instability detected! A negative value was found at {point}.")
else:
print(f" Status: Stable. Everything looks good at {point}.")
neutron_star_data = {'pressure_point_A': 10, 'pressure_point_B': -5}
_check_stability("Neutron Star", neutron_star_data)
white_dwarf_data = {'pressure_point_A': 5, 'pressure_point_B': 8}
_check_stability("White Dwarf", white_dwarf_data)
def find_celestial_objects():
"""Queries various astronomical databases."""
print("\n--- Searching International Astronomical Databases ---")
ra, dec = 180.0, -60.0
target_name = "M104"
start_date = "2023-01-01"
end_date = "2023-12-31"
print(f"Searching for information around coordinates RA:{ra}, DEC:{dec}...")
try:
coord = SkyCoord(ra=ra, dec=dec, unit=(u.degree, u.degree))
gaia_data = Gaia.query_object_async(coord, radius=u.Quantity(1.0, u.deg))
if gaia_data and isinstance(gaia_data, Table):
print(f" ✅ Found {len(gaia_data)} stars and objects in the Gaia catalog.")
else:
print(" ❌ No Gaia data found or query failed.")
except Exception as e:
print(f" ❌ Gaia search failed: {e}")
try:
cadc = Cadc()
cadc_result = cadc.query_object(target_name, collection='CFHT')
if cadc_result and len(cadc_result) > 0:
print(f" ✅ Found {len(cadc_result)} observations of '{target_name}' in the CADC database.")
else:
print(f" ❌ No CADC data found for '{target_name}'.")
except Exception as e:
print(f" ❌ CADC search failed: {e}")
try:
ned_result = Ned.query_object(target_name)
if ned_result and len(ned_result) > 0:
print(f" ✅ Found {len(ned_result)} entries for '{target_name}' in the NED database.")
else:
print(f" ❌ No NED data found for '{target_name}'.")
except Exception as e:
print(f" ❌ NED search failed: {e}")
try:
catalog = Client.open("https://earth-search.aws.element84.com/v1")
search = catalog.search(
intersects={"type": "Point", "coordinates": [ra, dec]},
datetime=f"{start_date}T00:00:00Z/{end_date}T23:59:59Z",
)
item_collection = search.item_collection()
print(f" ✅ Found {len(item_collection)} catalog items in the STAC search.")
except Exception as e:
print(f" ❌ STAC search failed: {e}")
try:
isro_data_url = "https://isro.gov.in/launcher"
requests.get(isro_data_url)
print(" ✅ Successfully connected to the ISRO data endpoint.")
except requests.exceptions.RequestException as e:
print(f" ❌ ISRO data query failed: {e}")
# --- Main Program Entry Point ---
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Integrated Client/Server for Astrophysics and Data Analysis.")
parser.add_argument('--mode', type=str, choices=['server', 'client'], required=True,
help='Specify the program mode: "server" to run the mathematical service, or "client" to run the astrophysics explorer.')
args = parser.parse_args()
if args.mode == 'server':
print("--- Starting Mathematical Service Server ---")
server = Server(SERVER_HOST, SERVER_PORT)
server.run()
elif args.mode == 'client':
print("--- Starting Simplified Astrophysics Explorer (Client) ---")
print("This program runs a full suite of astronomical analyses and database queries for you.")
print("It will attempt to connect to the math service server if one is running.")
analyze_fundamental_physics()
explore_cosmic_stability()
find_celestial_objects()
print("\n--- All client explorations and searches are complete. ---")
Tuesday, August 19, 2025
ASS.py Simplified Cosmos Explorer 0.2A
Subscribe to:
Post Comments (Atom)
No comments:
Post a Comment