Vombit/esp32-ir
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

2

Browse Source
This commit is contained in:
Vombit
2026-08-18 17:56:40 +05:00
parent 046284e9d8
commit f10f9de524
4 changed files with 260 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+143 -12
View File
@@ -1,38 +1,169 @@
# esp32-ir
ИК приёмник/передатчик на ESP32-C6, управляемый локально через REPL или удалённо по HTTP API.
- **Приёмник** — GPIO20
- **Передатчик** — GPIO19
### воспроизвести сигнал по имени
```
## Файлы
| Файл | Назначение |
| ---------------- | ------------------------------------------------------------------- |
| `capture.py` | захват ИК-сигнала с приёмника |
| `transmit.py` | воспроизведение сохранённого сигнала через ИК-светодиод (PWM 38кГц) |
| `wifi_server.py` | HTTP-сервер для удалённого управления |
| `.env` | секреты (WiFi, API-ключ) — см. ниже |
| `main.py` | автозапуск сервера при включении платы |
| `dotenv.py` | самописный экземпляр для получения переменных окружения |
---
## Локальное использование (через REPL)
### Воспроизвести сигнал по имени
```python
from transmit import send_code
send_code("power")
```
### сделать новую запись
```
from capture import capture_raw
### Сделать новую запись
```python
from capture import capture_raw, save_code
data = capture_raw()
save_code("name", data)
```
### посмотреть записи
```
### Посмотреть записи
```python
from capture import list_codes
list_codes()
```
### Удалить старые файлы/коды
```python
import os
os.remove("test.py") # удалить конкретный файл
### start server by hands
import json
with open("codes.json", "w") as f:
json.dump({}, f) # очистить все коды
```
---
## HTTP API
### Запуск сервера вручную
```python
import wifi_server
wifi_server.main()
```
При старте сервер подключается к WiFi (данные из `.env`) и выводит в консоль свой IP:
```
WiFi подключен. IP адрес: 192.168.1.123
Сервер запущен: http://192.168.1.123/
```
### Автозапуск
Сервер поднимается автоматически при каждом включении/перезагрузке платы через `main.py`:
```python
import wifi_server
wifi_server.main()
```
---
### Конфигурация (`.env`)
#### пример запущенного сервера
Сервер запущен: `http://192.168.1.22/`
```
WIFI_SSID=ИмяСети
WIFI_PASSWORD=Пароль
API_KEY=длинный_секретный_ключ
```
Записанные команды: `http://192.168.1.22/list`
Если `API_KEY` не задан — сервер работает **без авторизации** (только для локальных тестов).
Исполнение команды: `http://192.168.1.22/send?code=power`
---
### Авторизация
Все эндпоинты, кроме `/`, требуют ключ из `API_KEY`. Передавайте его любым из двух способов:
**Query-параметр:**
```
?key=длинный_секретный_ключ
```
**Заголовок:**
```
X-API-Key: длинный_секретный_ключ
```
Без ключа или с неверным — ответ `401 Unauthorized`.
---
### Эндпоинты
| Метод | Путь | Описание |
| ----- | ---------------------------------- | --------------------------------------------------------------------------- |
| `GET` | `/` | статус сервера, список эндпоинтов (без авторизации) |
| `GET` | `/list` | список сохранённых кодов с количеством импульсов |
| `GET` | `/send?code=ИМЯ` | воспроизвести сохранённый код |
| `GET` | `/record?code=ИМЯ&timeout_ms=5000` | записать новый код (наведите пульт и нажмите кнопку в течение `timeout_ms`) |
| `GET` | `/delete?code=ИМЯ` | удалить сохранённый код |
---
### Примеры использования
**Проверить, что сервер жив:**
```bash
curl http://192.168.1.123/
```
**Посмотреть список записанных команд:**
```bash
curl "http://192.168.1.123/list?key=длинный_секретный_ключ"
```
```json
{"codes": {"power": 67, "tv_power": 23, "light_toggle": 67}}
```
**Воспроизвести команду:**
```bash
curl "http://192.168.1.123/send?code=power&key=длинный_секретный_ключ"
```
```json
{"status": "ok", "sent": "power"}
```
**Записать новую команду** (после запроса наведите пульт на приёмник и нажмите кнопку):
```bash
curl "http://192.168.1.123/record?code=fan_speed1&key=длинный_секретный_ключ"
```
```json
{"status": "ok", "saved": "fan_speed1", "pulses": 67}
```
Если сигнал не поймали за отведённое время — `408 Request Timeout`.
**Удалить команду:**
```bash
curl "http://192.168.1.123/delete?code=fan_speed1&key=длинный_секретный_ключ"
```
```json
{"status": "ok", "deleted": "fan_speed1"}
```
**С заголовком вместо query-параметра:**
```bash
curl -H "X-API-Key: длинный_секретный_ключ" "http://192.168.1.123/send?code=power"
```
---
## Заметки
- Сервер однопоточный: во время `/record` (ожидание сигнала с пульта) остальные запросы не обрабатываются.
- Трафик идёт по обычному HTTP без шифрования — ключ защищает от случайных запросов в локальной сети, но не является криптостойкой защитой.
- IP адрес платы может меняться при переподключении к роутеру — для стабильности рекомендуется закрепить IP за MAC-адресом в настройках роутера (DHCP reservation).
main.py
+6
View File
@@ -1,2 +1,8 @@
import wifi_server
from machine import Pin
from neopixel import NeoPixel
np = NeoPixel(Pin(8), 1)
np[0] = (0, 0, 0)
np.write()
wifi_server.main()
transmit.py
+1 -1
View File
@@ -4,7 +4,7 @@ import json
TRANSMIT_PIN = 19
CARRIER_FREQ = 38000 # Гц, стандартная несущая для большинства ИК-пультов
DUTY = 33 # % заполнения (обычно 30-50% для ИК-светодиода)
DUTY = 50 # % заполнения (обычно 30-50% для ИК-светодиода)
def _send_timings(pwm, pin, timings):
wifi_server.py
+109 -16
View File
@@ -7,6 +7,7 @@ from dotenv import load_env
# ==== НАСТРОЙКИ ====
TRANSMIT_PIN = 19
CODES_FILE = "codes.json"
# ====================
@@ -27,7 +28,7 @@ def connect_wifi(ssid, password, timeout_s=15):
return ip
def load_codes(filename="codes.json"):
def load_codes(filename=CODES_FILE):
try:
with open(filename, "r") as f:
return json.load(f)
@@ -35,6 +36,32 @@ def load_codes(filename="codes.json"):
return {}
def save_codes(codes, filename=CODES_FILE):
with open(filename, "w") as f:
json.dump(codes, f)
def delete_code(name, filename=CODES_FILE):
codes = load_codes(filename)
if name not in codes:
return False
del codes[name]
save_codes(codes, filename)
return True
def capture_and_save(name, timeout_ms=5000, idle_us=8000, recv_pin=None):
from capture import capture_raw, RECV_PIN
pin = recv_pin if recv_pin is not None else RECV_PIN
timings = capture_raw(pin_num=pin, timeout_ms=timeout_ms, idle_us=idle_us)
if not timings:
return False, 0
codes = load_codes()
codes[name] = timings
save_codes(codes)
return True, len(timings)
def send_ir_code(name, pin_num=TRANSMIT_PIN):
from transmit import send_code
send_code(name, pin_num=pin_num)
@@ -71,9 +98,28 @@ def parse_query(path):
return route, params
def handle_request(path):
def handle_request(path, headers, api_key):
route, params = parse_query(path)
# Публичный маршрут - не требует ключа, чтобы можно было проверить что сервер жив
if route == "/":
return 200, {
"status": "ESP32 IR сервер запущен",
"endpoints": {
"/list": "список сохранённых кодов",
"/send?code=ИМЯ": "воспроизвести код",
"/record?code=ИМЯ&timeout_ms=5000": "записать новый код (наведите пульт и нажмите кнопку в течение timeout_ms)",
"/delete?code=ИМЯ": "удалить код"
},
"auth": "передайте ключ через заголовок X-API-Key или параметр ?key="
}
# Проверка ключа для всех остальных маршрутов
if api_key:
provided_key = headers.get("x-api-key") or params.get("key")
if provided_key != api_key:
return 401, {"error": "неверный или отсутствующий API-ключ"}
if route == "/send":
name = params.get("code")
if not name:
@@ -87,25 +133,65 @@ def handle_request(path):
except Exception as e:
return 500, {"error": str(e)}
elif route == "/record":
name = params.get("code")
if not name:
return 400, {"error": "укажите ?code=имя_кода"}
timeout_ms = int(params.get("timeout_ms", "5000"))
try:
ok, count = capture_and_save(name, timeout_ms=timeout_ms)
if ok:
return 200, {"status": "ok", "saved": name, "pulses": count}
else:
return 408, {"error": "сигнал не обнаружен (таймаут)"}
except Exception as e:
return 500, {"error": str(e)}
elif route == "/delete":
name = params.get("code")
if not name:
return 400, {"error": "укажите ?code=имя_кода"}
if delete_code(name):
return 200, {"status": "ok", "deleted": name}
else:
return 404, {"error": f"код '{name}' не найден"}
elif route == "/list":
codes = load_codes()
return 200, {"codes": list(codes.keys())}
elif route == "/":
return 200, {"status": "ESP32 IR сервер запущен", "endpoints": ["/list", "/send?code=ИМЯ"]}
details = {name: len(timings) for name, timings in codes.items()}
return 200, {"codes": details}
else:
return 404, {"error": "неизвестный маршрут"}
def run_server(ip):
def parse_headers(request):
"""Парсит заголовки HTTP-запроса в словарь с ключами в нижнем регистре."""
headers = {}
lines = request.split("\r\n")
for line in lines[1:]:
if not line:
break
if ":" in line:
key, value = line.split(":", 1)
headers[key.strip().lower()] = value.strip()
return headers
def run_server(ip, api_key=None):
addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(addr)
s.listen(1)
print(f"Сервер запущен: http://{ip}/")
if api_key:
print(
f"Примеры: http://{ip}/list?key=... | http://{ip}/send?code=power&key=...")
print("Защита включена: передавайте key через ?key= или заголовок X-API-Key")
else:
print(f"Примеры: http://{ip}/list | http://{ip}/send?code=power")
print("ВНИМАНИЕ: сервер работает без защиты (API_KEY не задан в .env)")
while True:
try:
@@ -116,23 +202,25 @@ def run_server(ip):
first_line = request.split("\r\n", 1)[0]
parts = first_line.split(" ")
path = parts[1] if len(parts) > 1 else "/"
headers = parse_headers(request)
status, body = handle_request(path)
status, body = handle_request(path, headers, api_key)
body_json = json.dumps(body)
body_bytes = body_json.encode("utf-8")
status_text = {200: "OK", 400: "Bad Request", 404: "Not Found",
500: "Internal Server Error"}.get(status, "OK")
status_text = {200: "OK", 400: "Bad Request", 401: "Unauthorized", 404: "Not Found",
408: "Request Timeout", 500: "Internal Server Error"}.get(status, "OK")
response = (
response_headers = (
f"HTTP/1.1 {status} {status_text}\r\n"
"Content-Type: application/json\r\n"
"Content-Type: application/json; charset=utf-8\r\n"
"Access-Control-Allow-Origin: *\r\n"
f"Content-Length: {len(body_json)}\r\n"
f"Content-Length: {len(body_bytes)}\r\n"
"Connection: close\r\n"
"\r\n"
f"{body_json}"
)
conn.send(response.encode())
conn.sendall(response_headers.encode("utf-8"))
conn.sendall(body_bytes)
conn.close()
except Exception as e:
print(f"Ошибка обработки запроса: {e}")
@@ -146,14 +234,19 @@ def main():
env = load_env()
ssid = env.get("WIFI_SSID")
password = env.get("WIFI_PASSWORD")
api_key = env.get("API_KEY")
if not ssid or not password:
print("Ошибка: заполните WIFI_SSID и WIFI_PASSWORD в .env")
return
if not api_key:
print(
"Предупреждение: API_KEY не задан в .env - эндпоинты будут доступны без защиты")
ip = connect_wifi(ssid, password)
if ip:
run_server(ip)
run_server(ip, api_key)
else:
print("Сервер не запущен: нет WiFi")