Task2_python_20260420.py - master - svs_win/demo_integration

Task2_python_20260420.py
238 строк9 KB
svs_win
create: Task2_python_20260420.py
20 апр 2026, 19:53
20 апр 2026, 19:53160ebee
import json
import math
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import datetime
import re

EPS = 1e-7
MAX_WAIT = 1e9

def normalize(vec):
    mag = math.hypot(vec['x'], vec['y'])
    if mag < EPS:
        return None
    return {'x': vec['x'] / mag, 'y': vec['y'] / mag}

def validate_input(data):
    try:
        if not isinstance(data, dict):
            return "Root must be JSON object"
        
        # target_star_vector
        tv = data.get('target_star_vector')
        if not isinstance(tv, dict) or 'x' not in tv or 'y' not in tv:
            return "Invalid target_star_vector"
        if not isinstance(tv['x'], (int, float)) or not isinstance(tv['y'], (int, float)):
            return "target_star_vector coords must be numbers"
        if math.hypot(tv['x'], tv['y']) < EPS:
            return "target_star_vector cannot be zero"

        # celestial_bodies
        bodies = data.get('celestial_bodies')
        if not isinstance(bodies, list) or len(bodies) > 100:
            return "celestial_bodies must be list of max length 100"
        
        ids = set()
        for i, b in enumerate(bodies):
            if not isinstance(b, dict):
                return f"Body {i} invalid"
            bid = b.get('id')
            btype = b.get('type')
            if not isinstance(bid, str) or not bid:
                return f"Body {i} missing id"
            if bid in ids:
                return f"Duplicate id {bid}"
            ids.add(bid)
            if btype not in ('star', 'planet', 'moon'):
                return f"Invalid type for {bid}"
            
            r = b.get('radius')
            if not isinstance(r, (int, float)) or r <= 0:
                return f"Invalid radius for {bid}"

            if btype == 'star':
                pos = b.get('position')
                if not isinstance(pos, dict) or 'x' not in pos or 'y' not in pos:
                    return f"Star {bid} missing position"
                if not isinstance(pos['x'], (int, float)) or not isinstance(pos['y'], (int, float)):
                    return f"Star {bid} position coords invalid"
            else:
                pid = b.get('parent_id')
                if not isinstance(pid, str) or not pid:
                    return f"{btype} {bid} missing parent_id"
                if pid == bid:
                    return f"{btype} {bid} cannot parent itself"
                
                orb = b.get('orbit_radius')
                if not isinstance(orb, (int, float)) or orb <= 0:
                    return f"Invalid orbit_radius for {bid}"
                
                av = b.get('angular_velocity')
                if not isinstance(av, (int, float)) or av <= 0:
                    return f"Invalid angular_velocity for {bid}"
                
                ia = b.get('initial_angle')
                if not isinstance(ia, (int, float)) or not (0 <= ia < 360):
                    return f"Invalid initial_angle for {bid}"
                
                rc = b.get('rotation_clockwise')
                if not isinstance(rc, bool):
                    return f"Invalid rotation_clockwise for {bid}"

        # observation_params
        op = data.get('observation_params')
        if not isinstance(op, dict):
            return "Invalid observation_params"
        st = op.get('start_time')
        rtt = op.get('required_transmission_time')
        if not isinstance(st, str) or not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', st):
            return "Invalid start_time format"
        try:
            datetime.fromisoformat(st.replace('Z', '+00:00'))
        except ValueError:
            return "Invalid start_time value"
        if not isinstance(rtt, int) or not (1 <= rtt <= 1000000):
            return "Invalid required_transmission_time"
            
        return None
    except Exception as e:
        return str(e)

class StarSolver:
    def __init__(self, data):
        self.target_dir = normalize(data['target_star_vector'])
        self.req_time = data['observation_params']['required_transmission_time']
        self.bodies = {}
        
        for b in data['celestial_bodies']:
            self.bodies[b['id']] = b
            
        # Precompute radians
        for b in self.bodies.values():
            if b['type'] != 'star':
                b['_omega'] = b['angular_velocity'] * math.pi / 180.0
                b['_theta0'] = b['initial_angle'] * math.pi / 180.0
                b['_clock'] = -1 if b['rotation_clockwise'] else 1

    def get_pos(self, bid, t):
        b = self.bodies[bid]
        if b['type'] == 'star':
            return (b['position']['x'], b['position']['y'])
        p_pos = self.get_pos(b['parent_id'], t) if b['parent_id'] != 'Atlas' else (0.0, 0.0)
        theta = b['_theta0'] + b['_clock'] * b['_omega'] * t
        return (p_pos[0] + b['orbit_radius'] * math.cos(theta),
                p_pos[1] + b['orbit_radius'] * math.sin(theta))

    def is_occluded(self, t):
        dx, dy = self.target_dir['x'], self.target_dir['y']
        for b in self.bodies.values():
            cx, cy = self.get_pos(b['id'], t)
            proj = cx * dx + cy * dy
            if proj <= 0: continue
            perp = abs(cx * dy - cy * dx)
            if perp <= b['radius'] + EPS:
                return True
        return False

    def find_event(self, t_start, become_occ, limit):
        """Находит время смены состояния. become_occ=True: ищем затмение, False: ищем просвет."""
        t_low = t_start
        t_high = t_start + 0.5
        state_start = self.is_occluded(t_start)
        
        # Экспоненциальный поиск интервала смены состояния
        while t_high <= limit:
            if self.is_occluded(t_high) != state_start:
                break
            t_low = t_high
            t_high = min(t_high * 2, limit + 1)
        else:
            return limit

        if self.is_occluded(t_high) == state_start:
            return limit # Не нашли в пределах лимита

        # Бисекция для точности
        for _ in range(60):
            t_mid = (t_low + t_high) / 2
            if self.is_occluded(t_mid) == state_start:
                t_low = t_mid
            else:
                t_high = t_mid
        return t_high

    def solve(self):
        t = 0.0
        # Защита от зацикливания на границах
        last_t = -1.0
        
        while t <= MAX_WAIT + self.req_time + 1:
            if abs(t - last_t) < 1e-4:
                t += 0.5
            last_t = t

            if not self.is_occluded(t):
                # Видимо. Пробуем начать в ближайшую целую секунду
                start_cand = math.ceil(t - EPS)
                if start_cand < 0: start_cand = 0
                
                if not self.is_occluded(start_cand):
                    # Ищем следующее затмение после start_cand
                    t_next_occ = self.find_event(start_cand, True, MAX_WAIT + self.req_time + 2)
                    duration = t_next_occ - start_cand
                    
                    if duration >= self.req_time:
                        wait = int(start_cand)
                        dur = int(math.floor(duration))
                        return {"found": True, "next_fitting_interval_in": wait, "interval_duration": dur}
                    else:
                        # Окно слишком короткое, прыгаем за него
                        t = t_next_occ
                        continue
                else:
                    t = self.find_event(t, False, MAX_WAIT + self.req_time)
            else:
                # Затмение, ищем выход
                t = self.find_event(t, False, MAX_WAIT + self.req_time)
                
        return {"found": False}

class RequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != '/api/v1/star_visibility':
            self.send_response(404)
            self.end_headers()
            return

        content_length = int(self.headers.get('Content-Length', 0))
        try:
            body = json.loads(self.rfile.read(content_length))
        except json.JSONDecodeError:
            self._respond(400, {"status": "incorrect_input"})
            return

        err = validate_input(body)
        if err:
            self._respond(400, {"status": "incorrect_input"})
            return

        try:
            solver = StarSolver(body)
            result = solver.solve()
            self._respond(200, result)
        except Exception:
            self._respond(400, {"status": "incorrect_input"})

    def _respond(self, code, data):
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

    def log_message(self, format, *args):
        pass  # Отключаем логи запросов для чистоты вывода

if __name__ == '__main__':
    server = HTTPServer(('0.0.0.0', 8000), RequestHandler)
    print("Server running on port 8000...")
    server.serve_forever()