import os import asyncio import base64 import json from datetime import datetime, timezone from typing import Dict, Optional, List TCP_HOST = os.getenv("TCP_HOST", "0.0.0.0") TCP_PORT = int(os.getenv("TCP_PORT", "8181")) MODE = os.getenv("MODE", "stream").lower() LENGTH_BYTES = int(os.getenv("LENGTH_BYTES", "2")) READ_CHUNK = int(os.getenv("READ_CHUNK", "65536")) MAX_MSG = int(os.getenv("MAX_MSG", "1048576")) IDLE_TIMEOUT = int(os.getenv("IDLE_TIMEOUT", "90")) ACK_JSON = b'{"ok":true}\n' LOG_MODE = os.getenv("LOG_MODE", "both").lower() LOG_MAX_BYTES = int(os.getenv("LOG_MAX_BYTES", "4096")) CTRL_HOST = os.getenv("CTRL_HOST", "0.0.0.0") CTRL_PORT = int(os.getenv("CTRL_PORT", "8182")) CLIENTS: Dict[str, asyncio.StreamWriter] = {} META: Dict[str, Dict] = {} DEV_INDEX: Dict[str, str] = {} INBOX: Dict[str, asyncio.Queue] = {} def bytes_to_hex(b: bytes) -> str: return "".join(f"{x:02x}" for x in b) def hexdump(b: bytes, width=16): def p(x): return chr(x) if 32 <= x < 127 else "." lines = [] for i in range(0, len(b), width): chunk = b[i:i+width] hexpart = " ".join(f"{c:02x}" for c in chunk) lines.append(f"{i:04x} {hexpart:<{width*3}} |{''.join(p(c) for c in chunk)}|") return "\n".join(lines) def print_payload(peer, data: bytes): ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%fZ") size = len(data) print(f"[tcp] {ts} {peer} -> {size} bytes", flush=True) sample = data[:LOG_MAX_BYTES] if LOG_MODE in ("text", "both"): print("----- TEXT (utf-8, truncated) -----", flush=True) print(sample.decode("utf-8", errors="replace"), flush=True) if LOG_MODE in ("hex", "both"): print("----- HEXDUMP (truncated) -----", flush=True) print(hexdump(sample), flush=True) if size > LOG_MAX_BYTES: print(f"... (truncated, showed first {LOG_MAX_BYTES} of {size} bytes)", flush=True) async def read_stream(r: asyncio.StreamReader): return (await r.read(READ_CHUNK)) or None async def read_line(r: asyncio.StreamReader): return (await r.readline()) or None async def read_length_prefixed(r: asyncio.StreamReader): hdr = await r.readexactly(LENGTH_BYTES) n = int.from_bytes(hdr, "big") if n < 0 or n > MAX_MSG: return None return await r.readexactly(n) async def handle_conn(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): addr = writer.get_extra_info("peername") key = f"{addr[0]}:{addr[1]}" CLIENTS[key] = writer META[key] = {"addr": addr, "device_id": None, "last": datetime.now(timezone.utc)} INBOX[key] = asyncio.Queue() print(f"[tcp] connected: {key} (mode={MODE})", flush=True) read_once = read_stream if MODE == "stream" else (read_line if MODE == "line" else read_length_prefixed) try: while True: try: chunk = await asyncio.wait_for(read_once(reader), timeout=1.0) except asyncio.TimeoutError: if (datetime.now(timezone.utc) - META[key]["last"]).total_seconds() > IDLE_TIMEOUT: print(f"[tcp] idle timeout: {key}", flush=True) break continue except asyncio.IncompleteReadError: break if not chunk: break META[key]["last"] = datetime.now(timezone.utc) print_payload(key, chunk) try: INBOX[key].put_nowait(chunk) except Exception: pass try: obj = json.loads(chunk.decode("utf-8", errors="ignore")) if isinstance(obj, dict) and obj.get("device_id"): META[key]["device_id"] = str(obj["device_id"]) DEV_INDEX[str(obj["device_id"]) ] = key except Exception: pass try: writer.write(ACK_JSON) await writer.drain() except ConnectionResetError: break finally: try: writer.close() await writer.wait_closed() if key in INBOX: q = INBOX.pop(key) while not q.empty(): try: q.get_nowait() except Exception: break except Exception: pass CLIENTS.pop(key, None) did = META.get(key, {}).get("device_id") if did and DEV_INDEX.get(did) == key: DEV_INDEX.pop(did, None) META.pop(key, None) print(f"[tcp] disconnected: {key}", flush=True) from fastapi import FastAPI, HTTPException, Body, Form from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field try: from fastapi.middleware.cors import CORSMiddleware except Exception: CORSMiddleware = None app = FastAPI(title="TCP Push Control") if CORSMiddleware: app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) class PushReq(BaseModel): target: str = Field(..., description='"*" | "ip:port" | "dev:"') data: str = Field(..., description='payload 문자열 (encoding에 따라 해석)') encoding: str = Field("text", description='"text" | "hex" | "base64"') append_newline: bool = Field(True, description='text일 때 끝에 \n 추가 여부') await_response: bool = Field(False, description="보낸 뒤 TCP 응답을 기다려 HTTP로 반환할지") response_timeout_ms: int = Field(3000, ge=1, le=30000, description="응답 대기 타임아웃(ms)") def resolve_targets(target: str) -> List[str]: if target == "*": return list(CLIENTS.keys()) if target.startswith("dev:"): key = DEV_INDEX.get(target[4:]) return [key] if key else [] return [target] if target in CLIENTS else [] def decode_payload(data: str, enc: str, append_newline: bool) -> bytes: enc = enc.lower() if enc == "text": b = data.encode("utf-8") return b + (b"\n" if append_newline else b"") if enc == "hex": h = "".join(ch for ch in data if ch in "0123456789abcdefABCDEF") return bytes.fromhex(h) if enc == "base64": return base64.b64decode(data) raise HTTPException(status_code=400, detail="encoding must be text|hex|base64") @app.get("/clients") def list_clients(): out = [{"key": k, "device_id": m.get("device_id"), "last": m.get("last").isoformat()} for k,m in META.items()] return {"count": len(out), "items": out} # ★ FIX: 경로를 '/push' 로 명시 (기존의 "/" 결합 실수 수정) @app.post("/push") async def push(req: PushReq): keys = resolve_targets(req.target) if not keys: raise HTTPException(status_code=404, detail="target not found or not connected") if req.await_response and len(keys) != 1: raise HTTPException(status_code=400, detail="await_response는 단일 대상일 때만 지원한다") payload = decode_payload(req.data, req.encoding, req.append_newline) sent, failed, responses = 0, 0, [] for k in keys: w = CLIENTS.get(k) if not w: failed += 1; continue if req.await_response: q = INBOX.get(k) if q is None: raise HTTPException(status_code=500, detail="internal inbox missing") while not q.empty(): try: q.get_nowait() except Exception: break try: w.write(payload); await w.drain(); sent += 1 except Exception: failed += 1; continue if req.await_response: try: chunk = await asyncio.wait_for(INBOX[k].get(), timeout=req.response_timeout_ms/1000.0) responses.append({"key": k, "size": len(chunk), "hex": bytes_to_hex(chunk), "text_preview": chunk.decode("utf-8", errors="replace")}) except asyncio.TimeoutError: responses.append({"key": k, "timeout": True, "hex": None, "text_preview": None}) return {"ok": True, "sent": sent, "failed": failed, "await_response": req.await_response, "responses": responses if req.await_response else None} # ---- Parser endpoints ---- import struct, re def crc16_modbus(data: bytes) -> int: crc = 0xFFFF for b in data: crc ^= b for _ in range(8): if crc & 0x0001: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc & 0xFFFF def _clean_hex(s: str) -> str: s = s.strip().replace("0x","").replace("0X","") s = re.sub(r"[^0-9A-Fa-f]", "", s) if len(s) % 2 == 1: s = "0" + s return s.lower() def parse_modbus_response(hex_str: str): raw_hex = _clean_hex(hex_str) if not raw_hex: return {"error": "no hex"} buf = bytes.fromhex(raw_hex) if buf.endswith(b"\r\n"): buf = buf[:-2] if len(buf) < 5: return {"error": "frame too short", "length": len(buf), "raw_hex": raw_hex} slave_id, function = buf[0], buf[1] if function in (1,2,3,4): byte_count = buf[2] data_start, data_end = 3, 3 + byte_count if data_end + 2 > len(buf): return {"error": "byte_count mismatch", "length": len(buf), "raw_hex": raw_hex} data = buf[data_start:data_end] provided_crc_lo, provided_crc_hi = buf[data_end], buf[data_end+1] provided_crc = (provided_crc_hi << 8) | provided_crc_lo calc_crc = crc16_modbus(buf[:data_end]) crc_ok = (calc_crc == provided_crc) regs = [int.from_bytes(data[i:i+2], "big") for i in range(0, len(data), 2) if i+2<=len(data)] floats = [] if len(data) >= 4 and len(data) % 4 == 0: for i in range(0, len(data), 4): w1, w2 = data[i:i+2], data[i+2:i+4] floats.append(struct.unpack(">f", w2 + w1)[0]) return {"id": slave_id, "function": function, "byte_count": byte_count, "data_hex": data.hex(), "registers_be": regs, "floats_wordswap": floats, "crc_provided_hex": f"{provided_crc:04x}", "crc_calculated_hex": f"{calc_crc:04x}", "crc_ok": crc_ok, "frame_hex": buf.hex()} return {"id": slave_id, "function": function, "frame_hex": buf.hex(), "note": "non-standard or write-response"} from fastapi import HTTPException @app.post("/test") def test_parse(hex_form: Optional[str] = Form(None), body: Optional[dict] = Body(None)): hex_input = hex_form or (body.get("hex") if body else None) or (body.get("data") if body else None) if not hex_input: raise HTTPException(status_code=400, detail="hex (또는 data) 가 필요하다") return {"ok": True, "input": hex_input, "result": parse_modbus_response(hex_input)} @app.post("/test_auto") def test_auto(push_json: dict = Body(...)): """ /push 의 JSON 응답을 그대로 넣으면 내부에서 hex를 뽑아 파싱한다. """ hexv = None try: if push_json and push_json.get("responses"): for r in push_json["responses"]: if isinstance(r, dict) and r.get("hex"): hexv = r["hex"] break except Exception: pass if not hexv: raise HTTPException(status_code=400, detail="push JSON에서 hex 응답을 찾지 못함 (target='*'이거나 timeout)") return {"ok": True, "input": {"from":"test_auto","hex":hexv}, "result": parse_modbus_response(hexv)} @app.get("/ui", response_class=HTMLResponse) def ui(): return """ /push 제어 + 데이터 읽기 (auto-parse)

/push 제어 + 데이터 읽기

특정 ip:port를 선택해야 응답을 기다려 파싱 가능. '*'는 방송용.
연결 조회 중…

제어

데이터 읽기

/push 응답

대기 중…

파싱 결과

CRC 유효성
-
레지스터(2B)
-
Float(워드스왑)
-
파생값
-
대기 중…
""" async def run_tcp(): server = await asyncio.start_server(handle_conn, TCP_HOST, TCP_PORT, reuse_port=True) addrs = ", ".join(str(s.getsockname()) for s in server.sockets) print(f"[tcp] listening on {addrs} (mode={MODE}, log={LOG_MODE})", flush=True) async with server: await server.serve_forever() async def run_http(): import uvicorn server = uvicorn.Server(uvicorn.Config(app, host=CTRL_HOST, port=CTRL_PORT, log_level="info", reload=False)) await server.serve() async def main(): await asyncio.gather(run_tcp(), run_http()) if __name__ == "__main__": asyncio.run(main())