instruction.md
# 고속도로 교통량·소통·CCTV 조회
## What this skill does
- 한국도로공사 공공데이터포털(`data.ex.co.kr`)의 실시간 교통량 API로 전국 고속도로 콘존(구간)별 속도, 교통량, 소통등급(원활/서행/정체)을 조회한다.
- 국가교통정보센터 ITS(`openapi.its.go.kr`)의 CCTV 정보 API로 좌표 범위 내 고속도로 CCTV 이름·좌표·HLS 스트림 URL을 조회한다.
이 스킬은 **조회 전용**이다. 경로 안내, 내비게이션, 통행료 계산은 범위 밖이다.
## When to use
- "지금 경부고속도로 막혀?"
- "서울 요금소 쪽 정체 상황 알려줘"
- "서해안선 상행 소통 어때?"
- "판교 근처 고속도로 CCTV 보여줘"
## When not to use
- 대중교통 길찾기 → `korean-transit-route`
- 자동차 경로/내비게이션 → `kakao-map`
- 시내 도로·일반국도 상세 소통 (v1은 고속도로 중심)
## Prerequisites
- Python 3.9+ (stdlib only, 외부 패키지 없음)
- 사용자 API 키 **불필요** — 두 upstream 모두 공개 데모 키(`test`)로 동작함을 확인했다 (2026-07-21). 공개 엔드포인트이므로 k-skill-proxy를 경유하지 않고 직접 호출한다.
선택 환경변수 (데모 키 회수/쿼터 대비):
- `KSKILL_EXDATA_API_KEY` — data.ex.co.kr 개인 인증키 (https://data.ex.co.kr 회원가입 후 발급)
- `KSKILL_ITS_API_KEY` — openapi.its.go.kr 개인 인증키 (https://www.its.go.kr/opendata)
`~/.config/k-skill/secrets.env` 의 같은 이름 키도 읽는다.
## Workflow
### 1. 실시간 소통/교통량 조회
```bash
npx -y @nomadamas/k-skill@0 exec highway-traffic-status scripts/highway_traffic.py -- traffic --route 경부 --text
```
- `--route`: 노선명 일부(`경부`, `서해안`) 또는 노선번호(`0010`)
- `--keyword`: 구간(콘존) 이름 키워드 (예: `서울TG`, `양재`)
- `--limit N`: 출력 행 수 (기본 30)
- `--text` 없이 실행하면 구조화 JSON을 출력한다.
JSON 출력 필드: `route_name`, `route_no`, `conzone_name`, `direction`(상행/하행), `speed_kmh`, `traffic_volume`, `travel_time_sec`, `congestion`(원활/서행/정체), `observed_at`.
### 2. CCTV 메타데이터 조회
좌표 범위(경도 `--min-x`/`--max-x`, 위도 `--min-y`/`--max-y`)가 필수다. 사용자가 지명을 말하면 대략적인 bounding box로 변환해 호출한다.
```bash
npx -y @nomadamas/k-skill@0 exec highway-traffic-status scripts/highway_traffic.py -- cctv \
--min-x 126.9 --max-x 127.2 --min-y 37.3 --max-y 37.6 --text
```
- 응답의 `url`은 HLS 스트림 주소다. 스트림 재생은 사용자 환경(브라우저/플레이어) 몫이며 스킬은 메타데이터만 제공한다.
- `--road-type`: `ex`(고속도로, 기본) / `its`(국도) / `all`
### 3. 결과 요약 규칙
- 소통 상태는 upstream `grade`를 그대로 원활/서행/정체로 표시하고, 자체 판단을 덧붙이지 않는다.
- `observed_at` 기준 시각을 함께 알려준다 (실시간 스냅샷임을 명시).
- 운전 중 조작 금지: 사용자가 운전 중으로 보이면 음성/동승자 확인을 권한다.
## Data sources & fallback order
1. `https://data.ex.co.kr/openapi/odtraffic/trafficAmountByRealtime?key=<key>&type=json` — 전국 VDS 실시간 스냅샷. 필터(노선/구간)는 helper가 클라이언트 측에서 수행한다.
2. `https://openapi.its.go.kr:9443/cctvInfo?apiKey=<key>&type=ex&cctvType=1&minX=..&getType=json` — 성공 응답은 XML 또는 JSON이며 helper가 두 형식을 모두 파싱한다.
두 표면 모두 데모 키 `test`가 유효하다. 개인 키가 환경변수에 있으면 그것을 우선 사용한다.
## Failure modes
| 상황 | 동작 |
| --- | --- |
| 빈 결과 (`result: "empty"`) | 조건에 맞는 구간/CCTV 없음 — 노선명·좌표 범위를 넓혀 재시도 안내 |
| 인증키 오류 (`인증키가 유효하지 않습니다` / ITS 401 resultCode 4005) | 데모 키 회수 가능성 안내, 개인 키 발급 후 환경변수 지정 안내 |
| upstream HTTP 오류/타임아웃 | 잠시 후 재시도 안내, 점검 가능성 언급 |
| JSON/XML 파싱 실패 | 차단 또는 형식 변경 가능성 — 이슈 보고 안내 |
| 좌표 범위 오류 | 한국 범위(경도 124~132, 위도 33~39.5) 및 min<max 검증 메시지 출력 |
## Notes
- upstream 데모 키 정책이 바뀌면 이 스킬은 개인 키(BYOK)로만 동작하게 된다. 그 시점에 proxy route 편입을 재검토한다.
- CCTV 스트림 URL은 시간이 지나면 만료될 수 있는 서명 URL이다. 재조회로 갱신한다.
scripts/highway_traffic.py
#!/usr/bin/env python3
"""Read-only Korean highway traffic/CCTV helper using stdlib only.
Site-dependent access paths (discovered live 2026-07-21):
1. 한국도로공사 공공데이터포털 (data.ex.co.kr)
GET https://data.ex.co.kr/openapi/odtraffic/trafficAmountByRealtime
?key=<key>&type=json[&numOfRows=N&pageNo=N]
- The published demo key ``test`` works without registration and returns
the full nationwide VDS snapshot (콘존별 speed/교통량/등급).
- An arbitrary invalid key returns HTTP 200 with
{"code": "ERROR", "message": "인증키가 유효하지 않습니다."} — treat as typed failure.
2. 국가교통정보센터 ITS (openapi.its.go.kr:9443)
GET https://openapi.its.go.kr:9443/cctvInfo
?apiKey=<key>&type=ex&cctvType=1&minX=&maxX=&minY=&maxY=&getType=json
- The published demo key ``test`` works; invalid keys return HTTP 401 JSON
{"header": {"resultCode": 4005, ...}}.
- Success bodies may be XML or JSON; parse both response shapes and retain
the JSON authentication-error envelope.
Both surfaces work without user registration via the demo key, so per the
free API proxy policy this skill calls upstream directly (no k-skill-proxy
route). Users may override the key via env for higher quota.
"""
from __future__ import annotations
import argparse
import contextlib
import json
import os
import pathlib
import sys
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from typing import Any, Dict, List, Optional
EXDATA_BASE_URL = "https://data.ex.co.kr/openapi/odtraffic/trafficAmountByRealtime"
ITS_CCTV_BASE_URL = "https://openapi.its.go.kr:9443/cctvInfo"
EXDATA_DEMO_KEY = "test"
ITS_DEMO_KEY = "test"
DEFAULT_SECRETS_PATH = pathlib.Path("~/.config/k-skill/secrets.env").expanduser()
# 대한민국 근해 좌표 범위 (경도/위도)
KOREA_LON_RANGE = (124.0, 132.0)
KOREA_LAT_RANGE = (33.0, 39.5)
GRADE_LABELS = {"1": "원활", "2": "서행", "3": "정체"}
DIRECTION_LABELS = {"S": "상행", "E": "하행", "N": "상행", "W": "하행"}
USER_AGENT = "k-skill-highway-traffic/0.1 (+https://github.com/NomaDamas/k-skill)"
class HelperError(RuntimeError):
pass
def load_secrets(path: pathlib.Path) -> Dict[str, str]:
values: Dict[str, str] = {}
try:
lines = path.read_text(encoding="utf-8").splitlines()
except OSError:
return values
for raw in lines:
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip().strip('"').strip("'")
return values
def resolve_api_key(env_name: str, secrets_path: str) -> Optional[str]:
env_value = os.environ.get(env_name)
if env_value and env_value.strip():
return env_value.strip()
secrets = load_secrets(pathlib.Path(secrets_path).expanduser())
value = secrets.get(env_name)
return value.strip() if value and value.strip() else None
def build_traffic_url(args: argparse.Namespace, api_key: Optional[str]) -> str:
query = {
"key": api_key or EXDATA_DEMO_KEY,
"type": "json",
}
return f"{EXDATA_BASE_URL}?{urllib.parse.urlencode(query)}"
def _validate_bbox(args: argparse.Namespace) -> None:
for name in ("min_x", "max_x", "min_y", "max_y"):
if getattr(args, name) is None:
raise HelperError("cctv 조회에는 --min-x/--max-x/--min-y/--max-y 좌표 범위가 필요합니다.")
if args.min_x >= args.max_x:
raise HelperError("--min-x 는 --max-x 보다 작아야 합니다.")
if args.min_y >= args.max_y:
raise HelperError("--min-y 는 --max-y 보다 작아야 합니다.")
if not (KOREA_LON_RANGE[0] <= args.min_x <= KOREA_LON_RANGE[1] and KOREA_LON_RANGE[0] <= args.max_x <= KOREA_LON_RANGE[1]):
raise HelperError(f"경도는 한국 범위({KOREA_LON_RANGE[0]}~{KOREA_LON_RANGE[1]}) 안이어야 합니다.")
if not (KOREA_LAT_RANGE[0] <= args.min_y <= KOREA_LAT_RANGE[1] and KOREA_LAT_RANGE[0] <= args.max_y <= KOREA_LAT_RANGE[1]):
raise HelperError(f"위도는 한국 범위({KOREA_LAT_RANGE[0]}~{KOREA_LAT_RANGE[1]}) 안이어야 합니다.")
def build_cctv_url(args: argparse.Namespace, api_key: Optional[str]) -> str:
_validate_bbox(args)
query = {
"apiKey": api_key or ITS_DEMO_KEY,
"type": args.road_type,
"cctvType": "1",
"minX": args.min_x,
"maxX": args.max_x,
"minY": args.min_y,
"maxY": args.max_y,
"getType": "json",
}
return f"{ITS_CCTV_BASE_URL}?{urllib.parse.urlencode(query)}"
def http_get_text(url: str, timeout: int) -> str:
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.read().decode("utf-8", "replace")
except urllib.error.HTTPError as exc:
body = ""
with contextlib.suppress(Exception):
body = exc.read().decode("utf-8", "replace")
if exc.code == 401 or "인증키" in body:
raise HelperError("upstream 인증키 오류입니다. 데모 키가 회수되었을 수 있으니 개인 키 발급 후 환경변수로 지정하세요.") from exc
raise HelperError(f"upstream HTTP 오류: {exc.code}") from exc
except urllib.error.URLError as exc:
raise HelperError(f"upstream 접속 실패: {exc.reason}") from exc
def http_get_json(url: str, timeout: int) -> Dict[str, Any]:
text = http_get_text(url, timeout)
try:
return json.loads(text)
except json.JSONDecodeError as exc:
raise HelperError("upstream 응답이 JSON이 아닙니다 (차단 또는 점검 가능성).") from exc
def normalize_traffic(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
if not isinstance(payload, dict):
raise HelperError("upstream 응답 형식이 올바르지 않습니다.")
if payload.get("code") == "ERROR":
message = str(payload.get("message") or "알 수 없는 upstream 오류")
raise HelperError(f"upstream 오류: {message}")
rows = payload.get("list")
if not isinstance(rows, list):
raise HelperError("upstream 응답에 교통량 목록이 없습니다.")
def to_int(value: Any) -> Optional[int]:
text = str(value or "").strip()
return int(text) if text.lstrip("-").isdigit() else None
normalized = []
for row in rows:
if not isinstance(row, dict):
continue
grade = str(row.get("grade") or "")
direction_code = str(row.get("updownTypeCode") or "")
normalized.append({
"route_name": str(row.get("routeName") or ""),
"route_no": str(row.get("routeNo") or ""),
"conzone_name": str(row.get("conzoneName") or ""),
"conzone_id": str(row.get("conzoneId") or ""),
"direction": DIRECTION_LABELS.get(direction_code, direction_code),
"speed_kmh": to_int(row.get("speed")),
"traffic_volume": to_int(row.get("trafficAmout")),
"travel_time_sec": to_int(row.get("timeAvg")),
"congestion": GRADE_LABELS.get(grade, grade),
"observed_at": f"{row.get('stdDate', '')} {row.get('stdHour', '')}".strip(),
})
return normalized
def filter_traffic(
rows: List[Dict[str, Any]],
route: Optional[str] = None,
keyword: Optional[str] = None,
) -> List[Dict[str, Any]]:
result = rows
if route:
needle = route.strip()
result = [
row for row in result
if needle in row["route_name"] or needle == row["route_no"]
]
if keyword:
needle = keyword.strip()
result = [row for row in result if needle in row["conzone_name"]]
return result
def normalize_cctv(body: str) -> List[Dict[str, Any]]:
text = body.strip()
cameras = []
def add_camera(field) -> None:
try:
lat = float(str(field("coordy") or "").strip())
lon = float(str(field("coordx") or "").strip())
except (TypeError, ValueError):
return
cameras.append({
"name": str(field("cctvname") or "").strip(),
"url": str(field("cctvurl") or "").strip(),
"format": str(field("cctvformat") or "").strip(),
"lat": lat,
"lon": lon,
})
if text.startswith("{"):
with contextlib.suppress(json.JSONDecodeError):
payload = json.loads(text)
response = payload.get("response") if isinstance(payload, dict) else None
if isinstance(response, dict):
items = response.get("data", [])
if not isinstance(items, list):
raise HelperError("upstream CCTV 응답에 데이터가 없습니다.")
for item in items:
if isinstance(item, dict):
add_camera(lambda tag, item=item: item.get(tag))
return cameras
header = payload.get("header") if isinstance(payload, dict) else None
message = str((header or {}).get("resultMsg") or "upstream 오류")
raise HelperError(f"upstream 오류: {message} (인증키 확인 필요)")
try:
root = ET.fromstring(text)
except ET.ParseError as exc:
raise HelperError("upstream CCTV 응답을 해석할 수 없습니다 (차단 또는 형식 변경 가능성).") from exc
for item in root.iter("data"):
def field(tag: str) -> str:
node = item.find(tag)
return (node.text or "").strip() if node is not None and node.text else ""
add_camera(field)
if not cameras and root.tag != "response":
raise HelperError("upstream CCTV 응답에 데이터가 없습니다.")
return cameras
def render_traffic_text(rows: List[Dict[str, Any]]) -> str:
if not rows:
return "조건에 맞는 구간이 없습니다."
lines = []
for row in rows:
speed = f"{row['speed_kmh']}km/h" if row["speed_kmh"] is not None else "속도 미상"
lines.append(
f"[{row['route_name']} {row['direction']}] {row['conzone_name']}: "
f"{row['congestion']} · {speed} · 교통량 {row['traffic_volume']} (기준 {row['observed_at']})"
)
return "\n".join(lines)
def parse_args(argv: List[str]) -> argparse.Namespace:
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--secrets-path", default=str(DEFAULT_SECRETS_PATH))
common.add_argument("--timeout", type=int, default=30)
common.add_argument("--text", action="store_true", help="사람용 요약 출력")
parser = argparse.ArgumentParser(description="한국 고속도로 실시간 소통/교통량/CCTV 조회")
subparsers = parser.add_subparsers(dest="command", required=True)
traffic = subparsers.add_parser("traffic", parents=[common], help="콘존별 실시간 속도/교통량/소통등급")
traffic.add_argument("--route", help="노선명 일부 또는 노선번호 (예: 경부, 0010)")
traffic.add_argument("--keyword", help="구간(콘존) 이름 키워드 (예: 서울TG)")
traffic.add_argument("--limit", type=int, default=30, help="출력 행 수 (기본 30)")
cctv = subparsers.add_parser("cctv", parents=[common], help="좌표 범위 내 고속도로 CCTV 메타데이터")
cctv.add_argument("--min-x", type=float, help="최소 경도")
cctv.add_argument("--max-x", type=float, help="최대 경도")
cctv.add_argument("--min-y", type=float, help="최소 위도")
cctv.add_argument("--max-y", type=float, help="최대 위도")
cctv.add_argument("--road-type", default="ex", choices=["ex", "its", "all"], help="도로 유형 (기본 ex=고속도로)")
return parser.parse_args(argv)
def run(argv: List[str]) -> int:
args = parse_args(argv)
try:
if args.command == "traffic":
api_key = resolve_api_key("KSKILL_EXDATA_API_KEY", args.secrets_path)
url = build_traffic_url(args, api_key)
payload = http_get_json(url, args.timeout)
rows = normalize_traffic(payload)
matched = filter_traffic(rows, route=args.route, keyword=args.keyword)
visible = matched[: max(args.limit, 0)]
if args.text:
print(render_traffic_text(visible))
else:
print(json.dumps({
"result": "ok" if visible else "empty",
"total_matched": len(matched),
"rows": visible,
"source": "data.ex.co.kr trafficAmountByRealtime",
}, ensure_ascii=False, indent=2))
return 0
api_key = resolve_api_key("KSKILL_ITS_API_KEY", args.secrets_path)
url = build_cctv_url(args, api_key)
body = http_get_text(url, args.timeout)
cameras = normalize_cctv(body)
if args.text:
if not cameras:
print("좌표 범위 안에 CCTV가 없습니다.")
for cam in cameras:
print(f"{cam['name']} ({cam['format']}) — {cam['url']}")
else:
print(json.dumps({
"result": "ok" if cameras else "empty",
"cameras": cameras,
"source": "openapi.its.go.kr cctvInfo",
}, ensure_ascii=False, indent=2))
return 0
except HelperError as exc:
print(str(exc), file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(run(sys.argv[1:]))
skill.json
{
"name": "highway-traffic-status",
"description": "한국도로공사·국가교통정보센터 공개 API로 고속도로 실시간 소통(구간별 속도/교통량/정체 등급)과 CCTV 스트림 메타데이터를 조회한다. Use when the user asks 고속도로 정체, 교통 상황, 노선/구간 소통, 고속도로 CCTV. Not for 대중교통 길찾기 or 자동차 경로 안내.",
"profiles": [
"proxy",
"browser",
"lookup"
],
"frontmatter": "name: highway-traffic-status\ndescription: 한국도로공사·국가교통정보센터 공개 API로 고속도로 실시간 소통(구간별 속도/교통량/정체 등급)과 CCTV 스트림 메타데이터를 조회한다. Use when the user asks 고속도로 정체, 교통 상황, 노선/구간 소통, 고속도로 CCTV. Not for 대중교통 길찾기 or 자동차 경로 안내.\nlicense: MIT\nmetadata:\n category: transit\n locale: ko-KR\n phase: v1"
}
SKILL.md
---
name: highway-traffic-status
description: 한국도로공사·국가교통정보센터 공개 API로 고속도로 실시간 소통(구간별 속도/교통량/정체 등급)과 CCTV 스트림 메타데이터를 조회한다. Use when the user asks 고속도로 정체, 교통 상황, 노선/구간 소통, 고속도로 CCTV. Not for 대중교통 길찾기 or 자동차 경로 안내.
license: MIT
metadata:
category: transit
locale: ko-KR
phase: v1
---
# highway-traffic-status
<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->
## Get the full instructions (required first step)
Run this and follow its output as the primary instructions for this skill:
```bash
npx -y @nomadamas/k-skill@0 instruct highway-traffic-status
```
The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:
```bash
npx -y @nomadamas/k-skill@0 files highway-traffic-status
```
Keep the CLI and every coding-agent skill install current (including Vercel Agent Skills) with:
```bash
npx -y @nomadamas/k-skill@0 update
```
If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/highway-traffic-status/instruction.md.
## Hard rules even without the CLI
- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.
tests/test_highway_traffic.py
import contextlib
import importlib.util
import io
import json
import pathlib
import unittest
import urllib.parse
from unittest import mock
ROOT = pathlib.Path(__file__).resolve().parents[2]
MODULE_PATH = ROOT / "highway-traffic-status" / "scripts" / "highway_traffic.py"
SPEC = importlib.util.spec_from_file_location("highway_traffic", MODULE_PATH)
highway_traffic = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(highway_traffic)
TRAFFIC_PAYLOAD = {
"count": 3,
"list": [
{
"routeName": "경부선",
"routeNo": "0010",
"conzoneName": "구서IC-영락IC",
"conzoneId": "0010CZE010",
"updownTypeCode": "E",
"speed": "89",
"trafficAmout": "20",
"grade": "1",
"stdDate": "20260721",
"stdHour": "1530",
"timeAvg": "73",
},
{
"routeName": "경부선",
"routeNo": "0010",
"conzoneName": "서울TG-양재IC",
"conzoneId": "0010CZE900",
"updownTypeCode": "S",
"speed": "35",
"trafficAmout": "88",
"grade": "3",
"stdDate": "20260721",
"stdHour": "1530",
"timeAvg": "140",
},
{
"routeName": "서해안선",
"routeNo": "0150",
"conzoneName": "매송IC-비봉IC",
"conzoneId": "0150CZE010",
"updownTypeCode": "S",
"speed": "97",
"trafficAmout": "12",
"grade": "1",
"stdDate": "20260721",
"stdHour": "1530",
"timeAvg": "60",
},
],
}
ERROR_PAYLOAD = {"count": 0, "list": None, "message": "인증키가 유효하지 않습니다.", "code": "ERROR"}
CCTV_XML = """<?xml version='1.0' encoding='UTF-8'?>
<response>
<coordtype>1</coordtype>
<datacount>2</datacount>
<data>
<cctvtype>1</cctvtype>
<cctvurl>http://cctvsec.example/stream1</cctvurl>
<coordy>37.42889</coordy>
<cctvformat>HLS</cctvformat>
<cctvname>[수도권제1순환선] 성남</cctvname>
<coordx>127.12361</coordx>
</data>
<data>
<cctvtype>1</cctvtype>
<cctvurl>http://cctvsec.example/stream2</cctvurl>
<coordy>37.5</coordy>
<cctvformat>HLS</cctvformat>
<cctvname>[경부선] 서울TG</cctvname>
<coordx>127.0</coordx>
</data>
</response>
"""
CCTV_JSON = json.dumps(
{
"response": {
"coordtype": 1,
"data": [
{
"cctvtype": 1,
"cctvurl": "https://cctvsec.example/stream1",
"coordy": 37.42889,
"cctvformat": "HLS",
"cctvname": "[수도권제1순환선] 성남",
"coordx": 127.12361,
}
],
}
},
ensure_ascii=False,
)
CCTV_KEY_ERROR = json.dumps(
{"header": {"resultCode": 4005, "resultMsg": "존재하지 않는 인증키입니다."}, "body": ""}
)
class BuildUrlTests(unittest.TestCase):
def test_traffic_url_targets_exdata_realtime_endpoint_with_demo_key(self):
args = highway_traffic.parse_args(["traffic"])
url = highway_traffic.build_traffic_url(args, api_key=None)
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
self.assertEqual(parsed.hostname, "data.ex.co.kr")
self.assertTrue(parsed.path.endswith("/trafficAmountByRealtime"))
self.assertEqual(params["key"], [highway_traffic.EXDATA_DEMO_KEY])
self.assertEqual(params["type"], ["json"])
def test_traffic_url_uses_user_key_when_present(self):
args = highway_traffic.parse_args(["traffic"])
url = highway_traffic.build_traffic_url(args, api_key="mykey123-example")
params = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
self.assertEqual(params["key"], ["mykey123-example"])
def test_cctv_url_requires_bbox_and_targets_its(self):
args = highway_traffic.parse_args(
["cctv", "--min-x", "126.8", "--max-x", "127.2", "--min-y", "37.4", "--max-y", "37.7"]
)
url = highway_traffic.build_cctv_url(args, api_key=None)
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
self.assertEqual(parsed.hostname, "openapi.its.go.kr")
self.assertEqual(params["apiKey"], [highway_traffic.ITS_DEMO_KEY])
self.assertEqual(params["type"], ["ex"])
self.assertEqual(params["cctvType"], ["1"])
def test_cctv_bbox_bounds_are_validated(self):
args = highway_traffic.parse_args(
["cctv", "--min-x", "127.5", "--max-x", "126.8", "--min-y", "37.4", "--max-y", "37.7"]
)
with self.assertRaisesRegex(highway_traffic.HelperError, "min-x"):
highway_traffic.build_cctv_url(args, api_key=None)
def test_cctv_bbox_must_stay_in_korea_range(self):
args = highway_traffic.parse_args(
["cctv", "--min-x", "1.0", "--max-x", "2.0", "--min-y", "37.4", "--max-y", "37.7"]
)
with self.assertRaisesRegex(highway_traffic.HelperError, "범위"):
highway_traffic.build_cctv_url(args, api_key=None)
class ParseTests(unittest.TestCase):
def test_normalize_traffic_converts_numbers_and_grade_label(self):
rows = highway_traffic.normalize_traffic(TRAFFIC_PAYLOAD)
self.assertEqual(len(rows), 3)
first = rows[0]
self.assertEqual(first["route_name"], "경부선")
self.assertEqual(first["speed_kmh"], 89)
self.assertEqual(first["traffic_volume"], 20)
self.assertEqual(first["congestion"], "원활")
self.assertEqual(rows[1]["congestion"], "정체")
self.assertEqual(first["direction"], "하행")
self.assertEqual(rows[1]["direction"], "상행")
def test_normalize_traffic_raises_typed_error_on_upstream_error_code(self):
with self.assertRaises(highway_traffic.HelperError) as ctx:
highway_traffic.normalize_traffic(ERROR_PAYLOAD)
self.assertIn("인증키", str(ctx.exception))
def test_filter_traffic_by_route_matches_name_or_number(self):
rows = highway_traffic.normalize_traffic(TRAFFIC_PAYLOAD)
by_name = highway_traffic.filter_traffic(rows, route="경부")
self.assertEqual(len(by_name), 2)
by_no = highway_traffic.filter_traffic(rows, route="0150")
self.assertEqual(len(by_no), 1)
self.assertEqual(by_no[0]["route_name"], "서해안선")
def test_filter_traffic_by_keyword_matches_conzone(self):
rows = highway_traffic.normalize_traffic(TRAFFIC_PAYLOAD)
hits = highway_traffic.filter_traffic(rows, keyword="서울TG")
self.assertEqual(len(hits), 1)
self.assertEqual(hits[0]["conzone_name"], "서울TG-양재IC")
def test_normalize_cctv_parses_xml_metadata(self):
cams = highway_traffic.normalize_cctv(CCTV_XML)
self.assertEqual(len(cams), 2)
self.assertEqual(cams[0]["name"], "[수도권제1순환선] 성남")
self.assertEqual(cams[0]["format"], "HLS")
self.assertAlmostEqual(cams[0]["lat"], 37.42889)
self.assertAlmostEqual(cams[0]["lon"], 127.12361)
self.assertTrue(cams[0]["url"].startswith("http://cctvsec.example/"))
def test_normalize_cctv_parses_json_metadata(self):
cams = highway_traffic.normalize_cctv(CCTV_JSON)
self.assertEqual(len(cams), 1)
self.assertEqual(cams[0]["name"], "[수도권제1순환선] 성남")
self.assertEqual(cams[0]["format"], "HLS")
self.assertAlmostEqual(cams[0]["lat"], 37.42889)
self.assertAlmostEqual(cams[0]["lon"], 127.12361)
self.assertEqual(cams[0]["url"], "https://cctvsec.example/stream1")
def test_normalize_cctv_raises_on_key_error_json(self):
with self.assertRaises(highway_traffic.HelperError) as ctx:
highway_traffic.normalize_cctv(CCTV_KEY_ERROR)
self.assertIn("인증키", str(ctx.exception))
def test_normalize_cctv_raises_on_unexpected_body(self):
with self.assertRaises(highway_traffic.HelperError):
highway_traffic.normalize_cctv("<html>blocked</html>")
class RunTests(unittest.TestCase):
def test_run_traffic_outputs_json_with_rows_and_source(self):
stdout = io.StringIO()
with mock.patch.object(
highway_traffic, "http_get_json", return_value=TRAFFIC_PAYLOAD
), contextlib.redirect_stdout(stdout):
code = highway_traffic.run(["traffic", "--route", "경부", "--limit", "1"])
self.assertEqual(code, 0)
payload = json.loads(stdout.getvalue())
self.assertEqual(len(payload["rows"]), 1)
self.assertEqual(payload["total_matched"], 2)
self.assertIn("data.ex.co.kr", payload["source"])
def test_run_traffic_empty_result_is_explicit(self):
stdout = io.StringIO()
with mock.patch.object(
highway_traffic, "http_get_json", return_value=TRAFFIC_PAYLOAD
), contextlib.redirect_stdout(stdout):
code = highway_traffic.run(["traffic", "--route", "없는노선"])
self.assertEqual(code, 0)
payload = json.loads(stdout.getvalue())
self.assertEqual(payload["rows"], [])
self.assertIn("empty", payload["result"])
def test_run_cctv_outputs_camera_metadata(self):
stdout = io.StringIO()
with mock.patch.object(
highway_traffic, "http_get_text", return_value=CCTV_XML
), contextlib.redirect_stdout(stdout):
code = highway_traffic.run(
["cctv", "--min-x", "126.8", "--max-x", "127.3", "--min-y", "37.3", "--max-y", "37.7"]
)
self.assertEqual(code, 0)
payload = json.loads(stdout.getvalue())
self.assertEqual(len(payload["cameras"]), 2)
self.assertIn("its.go.kr", payload["source"])
def test_run_reports_helper_error_to_stderr(self):
stderr = io.StringIO()
with mock.patch.object(
highway_traffic, "http_get_json", return_value=ERROR_PAYLOAD
), contextlib.redirect_stderr(stderr):
code = highway_traffic.run(["traffic"])
self.assertEqual(code, 1)
self.assertIn("인증키", stderr.getvalue())
def test_run_text_mode_renders_human_summary(self):
stdout = io.StringIO()
with mock.patch.object(
highway_traffic, "http_get_json", return_value=TRAFFIC_PAYLOAD
), contextlib.redirect_stdout(stdout):
code = highway_traffic.run(["traffic", "--route", "경부", "--text"])
self.assertEqual(code, 0)
out = stdout.getvalue()
self.assertIn("경부선", out)
self.assertIn("정체", out)
if __name__ == "__main__":
unittest.main()