Initial import

Author: fcrespel

.github/workflows/build.yml

@@ -0,0 +1,31 @@
+name: build
+
+on: [push, workflow_dispatch]
+
+env:
+  DOCKER_REGISTRY: ghcr.io
+  DOCKER_REPO: ${{ github.repository_owner }}
+  DOCKER_IMAGE: rc-server
+  DOCKER_TAG: ${{ github.ref_name }}
+  DOCKER_PLATFORM: linux/arm64
+
+jobs:
+  build-docker:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+    - name: Set up QEMU
+      uses: docker/setup-qemu-action@v3
+      with:
+        platforms: arm64
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@v3
+    - name: Login to Docker registry
+      uses: docker/login-action@v3
+      with:
+        registry: ${{ env.DOCKER_REGISTRY }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
+    - name: Build and publish image
+      run: docker buildx build --platform ${DOCKER_PLATFORM} --tag "${DOCKER_REGISTRY}/${DOCKER_REPO}/${DOCKER_IMAGE}:${DOCKER_TAG}" --push .

.gitignore

@@ -0,0 +1 @@
+__pycache__/

Dockerfile

@@ -0,0 +1,20 @@
+# Ubuntu 22.04 base image
+FROM ubuntu:22.04
+
+# Install Python
+RUN apt-get -q update && DEBIAN_FRONTEND="noninteractive" apt-get -q install -y -o Dpkg::Options::="--force-confnew" --no-install-recommends build-essential python3-dev python3-pip && rm -rf /var/lib/apt/lists/*
+
+# Create app directory
+WORKDIR /app
+
+# Install requirements
+COPY requirements.txt ./
+RUN pip install -r requirements.txt
+
+# Copy app
+COPY app/ ./
+
+# Run app
+ENTRYPOINT ["/usr/bin/python3", "server.py"]
+CMD ["-a", "0.0.0.0"]
+EXPOSE 8001

README.md

@@ -0,0 +1,82 @@
+# Remote Control REST API
+
+REST API to use an RF 433 MHz transmitter as a remote control. It currently only supports the Chacon DIO 1.0 protocol.
+
+It has only been tested on an ODROID-C4 board and uses an ODROID-specific WiringPi version. It should be easy to adapt it for other boards (e.g. Raspberry Pi).
+
+## Installation
+
+Execute the following commands on a Debian or Ubuntu system to install the required dependencies:
+```
+apt-get update
+apt-get install -y build-essential python3-dev python3-pip
+pip install -r requirements.txt
+```
+
+## Usage
+
+First, connect an RF 433 MHz transmitter to the GPIO pin of your choice. Take note of the corresponding WiringPi pin number (see [pinout.xyz](https://pinout.xyz/pinout/wiringpi)).
+
+### Server
+
+Execute the following command to run the server locally:
+
+```
+./app/server.py
+```
+
+You may then go to http://127.0.0.1:8001 to browse the documentation and test the API.
+
+The following arguments are available:
+
+```
+./app/server.py [-h] [-a ADDRESS] [-p PORT] [-g GPIO] [-l LOG_LEVEL]
+
+Optional arguments:
+  -h, --help                           Show help message and exit
+  -a ADDRESS, --address ADDRESS        Address to bind to (default: 127.0.0.1)
+  -p PORT, --port PORT                 Port to listen on (default: 8001)
+  -g GPIO, --gpio GPIO                 GPIO WiringPi pin number (default: 0)
+  -l LOG_LEVEL, --log-level LOG_LEVEL  Log level: CRITICAL, ERROR, WARNING, INFO, DEBUG (default: INFO)
+```
+
+A Docker image is also available for the arm64 architecture:
+
+```
+docker run -it --rm --privileged -p 8001:8001 ghcr.io/fcrespel/rc-server:master [-h] [-a ADDRESS] [-p PORT] [-g GPIO] [-l LOG_LEVEL]
+```
+
+You may want to run it in the background using commands such as the following:
+
+```
+# Create and start container
+docker run -d --name rc-server --privileged -p 127.0.0.1:8001:8001 ghcr.io/fcrespel/rc-server:master
+
+# Stop server
+docker stop rc-server
+
+# Start server
+docker start rc-server
+
+# Show live logs
+docker logs -f rc-server
+```
+
+NOTE: the API port is not secured, make sure to only expose it locally or to trusted clients.
+
+### Client
+
+You may call the API with any HTTP client such as curl:
+
+```
+# Replace 12345678 with the actual Chacon DIO 1.0 sender code (arbitrary 26-bit integer)
+
+# Get button 1 status:
+curl -sSf -XGET http://127.0.0.1:8001/chacondio10/12345678/1
+
+# Set button 1 to ON:
+curl -sSf -XPUT http://127.0.0.1:8001/chacondio10/12345678/1 -d 1
+
+# Set button 1 to OFF:
+curl -sSf -XPUT http://127.0.0.1:8001/chacondio10/12345678/1 -d 0
+```

app/chacondio10/__init__.py

@@ -0,0 +1,31 @@
+import argparse
+
+import odroid_wiringpi as wiringpi
+
+from .protocol import transmit
+
+
+def parse_args():
+  parser = argparse.ArgumentParser(description="Chacon DIO 1.0 remote control")
+  parser.add_argument("-g", "--gpio", help="GPIO WiringPi pin number (default: 0)", type=int, choices=range(0, 30), metavar="[0-29]", default=0)
+  parser.add_argument("-s", "--sender", help="Sender code 26-bit number", type=int, required=True)
+  parser.add_argument("-b", "--button", help="Button number between 0 and 15, -1 for all (group function)", type=int, choices=range(-1, 16), metavar="[0-15]", required=True)
+  parser.add_argument("-o", "--onoff", help="0 (OFF) or 1 (ON)", type=int, choices=range(0, 2), metavar="[0-1]", required=True)
+  parser.add_argument("-r", "--repeat", help="Number of times to repeat the message (default: 5)", type=int, default=5)
+  return parser.parse_args()
+
+def main():
+  args = parse_args()
+  group = True if args.button < 0 else False
+  button = 0 if args.button < 0 else args.button
+  onoff = True if args.onoff > 0 else False
+
+  if wiringpi.wiringPiSetup() == -1:
+    raise Exception("Failed to initialize WiringPi")
+  wiringpi.pinMode(args.gpio, wiringpi.OUTPUT)
+
+  for i in range(args.repeat):
+    transmit(args.gpio, args.sender, group, button, onoff)
+
+if __name__ == "__main__":
+  main()

app/chacondio10/cli.py

@@ -0,0 +1,64 @@
+import odroid_wiringpi as wiringpi
+
+TIME_HIGH_LOCK = 275
+TIME_LOW_LOCK1 = 9900
+TIME_LOW_LOCK2 = 2675
+
+TIME_HIGH_DATA = 275        # 310 or 275 or 220
+TIME_LOW_DATA_LONG = 1225   # 1340 or 1225 or 1400
+TIME_LOW_DATA_SHORT = 275   # 310 or 275 or 350
+
+def sendBit(pin: int, b: bool):
+  if b:
+    wiringpi.digitalWrite(pin, wiringpi.HIGH)
+    wiringpi.delayMicroseconds(TIME_HIGH_DATA)
+    wiringpi.digitalWrite(pin, wiringpi.LOW)
+    wiringpi.delayMicroseconds(TIME_LOW_DATA_LONG)
+  else:
+    wiringpi.digitalWrite(pin, wiringpi.HIGH)
+    wiringpi.delayMicroseconds(TIME_HIGH_DATA)
+    wiringpi.digitalWrite(pin, wiringpi.LOW)
+    wiringpi.delayMicroseconds(TIME_LOW_DATA_SHORT)
+
+def sendPair(pin: int, b: bool):
+  sendBit(pin, b)
+  sendBit(pin, not b)
+
+def sendWord(pin: int, word: int, bits: int):
+  for bit in reversed(range(bits)):
+    if word & (1 << bit):
+      sendPair(pin, True)
+    else:
+      sendPair(pin, False)
+
+def transmit(pin: int, sender: int, group: bool, button: int, onoff: bool):
+  # Start lock
+  wiringpi.digitalWrite(pin, wiringpi.HIGH);
+  wiringpi.delayMicroseconds(TIME_HIGH_LOCK);
+  wiringpi.digitalWrite(pin, wiringpi.LOW);
+  wiringpi.delayMicroseconds(TIME_LOW_LOCK1);
+  wiringpi.digitalWrite(pin, wiringpi.HIGH);
+  wiringpi.delayMicroseconds(TIME_HIGH_LOCK);
+  wiringpi.digitalWrite(pin, wiringpi.LOW);
+  wiringpi.delayMicroseconds(TIME_LOW_LOCK2);
+  wiringpi.digitalWrite(pin, wiringpi.HIGH);
+
+  # Sender code (26 bits)
+  sendWord(pin, sender, 26);
+
+  # Group bit
+  sendPair(pin, group);
+
+  # On/off bit
+  sendPair(pin, onoff);
+
+  # Button number (4 bits)
+  sendWord(pin, button, 4);
+
+  # End lock
+  wiringpi.digitalWrite(pin, wiringpi.HIGH);
+  wiringpi.delayMicroseconds(TIME_HIGH_LOCK);
+  wiringpi.digitalWrite(pin, wiringpi.LOW);
+
+  # Delay before next transmission
+  wiringpi.delay(10)

app/chacondio10/protocol.py

@@ -0,0 +1,25 @@
+from fastapi import APIRouter, Body, Path, Request
+
+from .protocol import transmit
+
+router = APIRouter(prefix="/chacondio10", tags=["chacondio10"])
+
+@router.get("/{sender}/{button}")
+async def get_button(request: Request, sender: int = Path(ge=0, le=67108863), button: int = Path(ge=0, le=15)):
+  if sender in request.app.state.chacondio10 and button in request.app.state.chacondio10[sender]:
+    return request.app.state.chacondio10[sender][button]
+  else:
+    return 0
+
+@router.put("/{sender}/{button}")
+async def put_button(request: Request, sender: int = Path(ge=0, le=67108863), button: int = Path(ge=0, le=15), onoff: int = Body(ge=0, le=1), repeat: int = 5):
+  if not sender in request.app.state.chacondio10:
+    request.app.state.chacondio10[sender] = {}
+  if onoff > 0:
+    request.app.state.chacondio10[sender][button] = 1
+    for i in range(repeat):
+      transmit(request.app.state.gpio, sender, False, button, True)
+  else:
+    request.app.state.chacondio10[sender][button] = 0
+    for i in range(repeat):
+      transmit(request.app.state.gpio, sender, False, button, False)

app/chacondio10/routes.py

@@ -0,0 +1,50 @@
+#!/usr/bin/python3
+
+import argparse
+import logging
+from contextlib import asynccontextmanager
+
+import odroid_wiringpi as wiringpi
+import uvicorn
+from chacondio10.routes import router as chacondio10_router
+from fastapi import FastAPI
+from fastapi.responses import RedirectResponse
+
+logger = logging.getLogger("uvicorn.error")
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+  logger.info(f"Setting up WiringPi for GPIO {app.state.gpio}")
+  if wiringpi.wiringPiSetup() == -1:
+    raise Exception("Failed to initialize WiringPi")
+  wiringpi.pinMode(app.state.gpio, wiringpi.OUTPUT)
+  yield
+
+app = FastAPI(title="Remote Control REST API", description="REST API to use an RF 433 MHz transmitter as a remote control", version="1.0", lifespan=lifespan)
+app.include_router(chacondio10_router)
+app.state.gpio = 0
+app.state.chacondio10 = {}
+
+@app.get("/", include_in_schema=False)
+async def home_page():
+  return RedirectResponse("/docs")
+
+@app.get("/health", tags=["health"])
+async def health():
+  return {"status": "UP"}
+
+def parse_args():
+  parser = argparse.ArgumentParser(description=app.title)
+  parser.add_argument("-a", "--address", help="Address to bind to (default: 127.0.0.1)", type=str, default="127.0.0.1")
+  parser.add_argument("-p", "--port", help="Port to listen on (default: 8001)", type=int, default=8001)
+  parser.add_argument("-g", "--gpio", help="GPIO WiringPi pin number (default: 0)", type=int, choices=range(0, 30), metavar="GPIO", default=0)
+  parser.add_argument("-l", "--log-level", help="Log level (default: INFO)", type=str, default="INFO", choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"])
+  return parser.parse_args()
+
+def main():
+  args = parse_args()
+  app.state.gpio = args.gpio
+  uvicorn.run(app, host=args.address, port=args.port, log_level=logging.getLevelName(args.log_level))
+
+if __name__ == "__main__":
+  main()

app/server.py

@@ -0,0 +1,3 @@
+fastapi~=0.111.0
+uvicorn[standard]~=0.29.0
+odroid-wiringpi~=3.16.2