Code reference

Auto-generated from the NumPy-style docstrings in src/ via mkdocstrings. For the HTTP endpoints, see the HTTP API page instead.

hid_keyboard_server

The HTTP bridge: turns requests into USB-HID reports, serves the web remote and HLS segments, and relays RTSP to HLS/WebRTC.

hid_keyboard_server

pi-remote HID keyboard + consumer bridge + web remote + Broadlink IR.

Stdlib only (no third-party imports in this process). Endpoints: GET / -> web remote UI (remote.html) GET/POST /type text -> type a string on the USB keyboard GET/POST /key key[+mod] -> single key (named or single char) GET/POST /press key+mods -> key combo, e.g. Ctrl+A, Alt+Tab GET/POST /media name -> consumer control (HOME/PLAYPAUSE/VOLUP/...) GET/POST /ir cmd=NAME -> send a learned IR code via Broadlink (ir_tool.py)

Auth (optional): set PI_REMOTE_API_KEY; then every request must pass ?token=... or the header X-API-Key. The remote UI is always served so it can read the token from its own URL.

Configuration is read from environment variables (see config/config.example.env): PI_REMOTE_PORT, PI_REMOTE_API_KEY, PI_REMOTE_KEY_DELAY, PI_REMOTE_HTML, PI_REMOTE_IR_TOOL, PI_REMOTE_HID_KBD, PI_REMOTE_HID_CONSUMER

Handler

Bases: BaseHTTPRequestHandler

HTTP request handler exposing the pi-remote API and web remote.

Routes requests to the keyboard / consumer / IR / preview helpers, serves the remote UI and HLS segments, applies optional API-key auth, and emits CORS headers. Methods prefixed with _ are internal helpers; the do_* methods are the standard :class:http.server.BaseHTTPRequestHandler entry points.

Source code in src/hid_keyboard_server.py

class Handler(BaseHTTPRequestHandler):
    """HTTP request handler exposing the pi-remote API and web remote.

    Routes requests to the keyboard / consumer / IR / preview helpers, serves the
    remote UI and HLS segments, applies optional API-key auth, and emits CORS
    headers. Methods prefixed with ``_`` are internal helpers; the ``do_*``
    methods are the standard
    :class:`http.server.BaseHTTPRequestHandler` entry points.
    """
    def _auth(self, q):
        """Return whether the request is authorized.

        Parameters
        ----------
        q : dict
            Parsed query string (values are lists, as returned by ``parse_qs``).

        Returns
        -------
        bool
            ``True`` if no API key is configured, or the request supplies the
            matching key via the ``X-API-Key`` header or the ``token`` query
            parameter.
        """
        if not API_KEY: return True
        return (self.headers.get('X-API-Key') or q.get('token',[''])[0]) == API_KEY
    def _cors(self):
        """Emit permissive CORS headers (origin ``*``) on the current response."""
        self.send_header('Access-Control-Allow-Origin','*')
        self.send_header('Access-Control-Allow-Headers','Content-Type,X-API-Key')
    def _send(self, code, obj):
        """Send a JSON response.

        Parameters
        ----------
        code : int
            HTTP status code.
        obj : object
            JSON-serializable response body.

        Returns
        -------
        None
        """
        body = json.dumps(obj).encode()
        self.send_response(code); self._cors()
        self.send_header('Content-Type','application/json')
        self.send_header('Content-Length',str(len(body)))
        self.end_headers(); self.wfile.write(body)
    def _html(self):
        """Serve the web remote page (``REMOTE_HTML``), or a placeholder if absent.

        Returns
        -------
        None
        """
        try:
            with open(REMOTE_HTML,'rb') as f: body = f.read()
        except FileNotFoundError:
            body = b'<h1>remote.html not found</h1>'
        self.send_response(200); self._cors()
        self.send_header('Content-Type','text/html; charset=utf-8')
        self.send_header('Content-Length',str(len(body)))
        self.end_headers(); self.wfile.write(body)
    def _serve_stream(self, path):
        """Serve an HLS playlist or segment from ``STREAM_DIR``.

        Only filenames matching ``*.m3u8`` / ``*.ts`` are served (no auth, so any
        HLS player can read them); anything else returns 404.

        Parameters
        ----------
        path : str
            Request path beginning with ``/stream/`` (e.g. ``/stream/live.m3u8``).

        Returns
        -------
        None
        """
        name = path[len('/stream/'):]
        if not re.match(r'^[A-Za-z0-9_.\-]+\.(m3u8|ts)$', name):
            return self._send(404, {'error': 'not found'})
        try:
            with open(os.path.join(STREAM_DIR, name), 'rb') as f: body = f.read()
        except FileNotFoundError:
            return self._send(404, {'error': 'not found'})
        ctype = 'application/vnd.apple.mpegurl' if name.endswith('.m3u8') else 'video/mp2t'
        self.send_response(200); self._cors()
        self.send_header('Content-Type', ctype)
        self.send_header('Cache-Control', 'no-store')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers(); self.wfile.write(body)
    def _parse(self):
        """Split the request URL into its path and parsed query.

        Returns
        -------
        tuple of (str, dict)
            The URL path and the ``parse_qs`` mapping of the query string.
        """
        u = urlparse(self.path); return u.path, parse_qs(u.query)
    def _json(self):
        """Read and parse the request body as JSON.

        Returns
        -------
        dict
            The parsed object, ``{}`` if the body is empty, or
            ``{"text": <raw>}`` if the body is not valid JSON.
        """
        n = int(self.headers.get('Content-Length',0) or 0)
        if not n: return {}
        raw = self.rfile.read(n)
        try: return json.loads(raw)
        except Exception: return {'text': raw.decode('utf-8','ignore')}
    def do_OPTIONS(self):
        """Answer a CORS preflight request with the allowed methods and headers."""
        self.send_response(204); self._cors()
        self.send_header('Access-Control-Allow-Methods','GET,POST,OPTIONS')
        self.send_header('Content-Length','0'); self.end_headers()
    def do_GET(self):
        """Handle an HTTP GET request (parameters from the query string)."""
        self._handle(False)
    def do_POST(self):
        """Handle an HTTP POST request (JSON body, with query-string fallback)."""
        self._handle(True)
    def _handle(self, post):
        """Dispatch a request to the matching endpoint.

        Serves the UI and HLS files without auth, enforces the API key for
        everything else, then routes ``/type``, ``/key``, ``/press``, ``/media``,
        ``/ir``, ``/stream/start``, ``/stream/stop`` and ``/webrtc/start``.
        Helper exceptions are mapped to HTTP error codes (400 for bad input, 502
        for IR/relay failures, 503 for a missing HID device).

        Parameters
        ----------
        post : bool
            ``True`` for POST (read a JSON body), ``False`` for GET.

        Returns
        -------
        None
        """
        path, q = self._parse()
        if path in ('/', '/remote', '/index.html'): return self._html()
        if path.startswith('/stream/') and path not in ('/stream/start', '/stream/stop'):
            return self._serve_stream(path)
        if not self._auth(q): return self._send(401, {'error':'unauthorized'})
        d = self._json() if post else {}
        def gv(k):
            """Get request value `k` from the JSON body (POST) or query string."""
            if post and k in d: return d[k]
            return q.get(k, [''])[0]
        m = d.get('mods') or d.get('mod') or q.get('mod')
        if m and not isinstance(m, list): m = [m]
        try:
            if path == '/type':           return self._send(200, {'typed': type_text(gv('text'))})
            if path in ('/key','/press'): press(gv('key'), m); return self._send(200, {'pressed': gv('key')})
            if path == '/media':          media(gv('key')); return self._send(200, {'media': gv('key')})
            if path == '/ir':             ir_send(gv('cmd')); return self._send(200, {'ir': gv('cmd')})
            if path == '/stream/start':   return self._send(200, {'hls': stream_start(gv('url'))})
            if path == '/stream/stop':    stream_stop(); return self._send(200, {'stopped': True})
            if path == '/webrtc/start':   return self._send(200, {'embed': webrtc_start(gv('url'), self.headers.get('Host',''))})
        except KeyError as e:          return self._send(400, {'error':'unknown key: %s' % e})
        except ValueError as e:        return self._send(400, {'error':'%s' % e})
        except RuntimeError as e:      return self._send(502, {'error':'%s' % e})
        except FileNotFoundError as e: return self._send(503, {'error':'device missing: %s' % e})
        return self._send(200, {'status':'ok','endpoints':['/','/type','/key','/press','/media','/ir','/stream/start','/stream/stop','/webrtc/start']})
    def log_message(self, *a):
        """Suppress the default per-request logging to stderr."""
        pass

do_OPTIONS

do_OPTIONS()

Answer a CORS preflight request with the allowed methods and headers.

Source code in src/hid_keyboard_server.py

def do_OPTIONS(self):
    """Answer a CORS preflight request with the allowed methods and headers."""
    self.send_response(204); self._cors()
    self.send_header('Access-Control-Allow-Methods','GET,POST,OPTIONS')
    self.send_header('Content-Length','0'); self.end_headers()

do_GET

do_GET()

Handle an HTTP GET request (parameters from the query string).

Source code in src/hid_keyboard_server.py

def do_GET(self):
    """Handle an HTTP GET request (parameters from the query string)."""
    self._handle(False)

do_POST

do_POST()

Handle an HTTP POST request (JSON body, with query-string fallback).

Source code in src/hid_keyboard_server.py

def do_POST(self):
    """Handle an HTTP POST request (JSON body, with query-string fallback)."""
    self._handle(True)

log_message

log_message(*a)

Suppress the default per-request logging to stderr.

Source code in src/hid_keyboard_server.py

def log_message(self, *a):
    """Suppress the default per-request logging to stderr."""
    pass

char_to_report

char_to_report(c)

Map a single character to its USB HID keyboard report.

Parameters:

Name	Type	Description	Default
c	str	A single character.	required

Returns:

Type	Description
tuple of (int, int) or None	(modifier_bitmask, usage_id) for the character, or None if the character cannot be typed.

Source code in src/hid_keyboard_server.py

def char_to_report(c):
    """Map a single character to its USB HID keyboard report.

    Parameters
    ----------
    c : str
        A single character.

    Returns
    -------
    tuple of (int, int) or None
        ``(modifier_bitmask, usage_id)`` for the character, or ``None`` if the
        character cannot be typed.
    """
    if c in _SHIFTED: return (SHIFT, _BASE[_SHIFTED[c]])
    if c.isupper() and c.lower() in _BASE: return (SHIFT, _BASE[c.lower()])
    if c in _BASE: return (0, _BASE[c])
    return None

type_text

type_text(text)

Type a string on the USB keyboard, one character at a time.

Characters with no HID mapping are skipped. A KEY_DELAY pause is inserted between keystrokes so the host does not drop fast input.

Parameters:

Name	Type	Description	Default
text	str	The text to type.	required

Returns:

Type	Description
int	The number of characters actually sent.

Source code in src/hid_keyboard_server.py

def type_text(text):
    """Type a string on the USB keyboard, one character at a time.

    Characters with no HID mapping are skipped. A ``KEY_DELAY`` pause is inserted
    between keystrokes so the host does not drop fast input.

    Parameters
    ----------
    text : str
        The text to type.

    Returns
    -------
    int
        The number of characters actually sent.
    """
    n = 0
    for c in text:
        r = char_to_report(c)
        if r is None: continue
        _write_kbd(*r); n += 1; time.sleep(KEY_DELAY)
    return n

press

press(key, mods=None)

Press a single key (named key or character) with optional modifiers.

Parameters:

Name	Type	Description	Default
key	str	A named key (e.g. "ENTER", "DOWN", "F5") or a single character (e.g. "a").	required
mods	iterable of str	Modifier names to hold while pressing (e.g. ["CTRL"] for Ctrl+key).	None

Returns:

Type	Description
None

Raises:

Type	Description
KeyError	If key is neither a known named key nor a mappable single character.

Source code in src/hid_keyboard_server.py

def press(key, mods=None):
    """Press a single key (named key or character) with optional modifiers.

    Parameters
    ----------
    key : str
        A named key (e.g. ``"ENTER"``, ``"DOWN"``, ``"F5"``) or a single
        character (e.g. ``"a"``).
    mods : iterable of str, optional
        Modifier names to hold while pressing (e.g. ``["CTRL"]`` for Ctrl+key).

    Returns
    -------
    None

    Raises
    ------
    KeyError
        If `key` is neither a known named key nor a mappable single character.
    """
    mod = _modmask(mods); key = str(key)
    if key.upper() in _NAMED:
        bm, code = _NAMED[key.upper()]; mod |= bm
    elif len(key) == 1:
        r = char_to_report(key)
        if r is None: raise KeyError(key)
        bm, code = r; mod |= bm
    else: raise KeyError(key)
    _write_kbd(mod, code)

media

media(name)

Send a consumer-control (media) usage to the second HID device.

Parameters:

Name	Type	Description	Default
name	str	Consumer-control name (e.g. "PLAYPAUSE", "VOLUP", "HOME"); case-insensitive.	required

Returns:

Type	Description
None

Raises:

Type	Description
KeyError	If name is not a known consumer-control name.

Source code in src/hid_keyboard_server.py

def media(name):
    """Send a consumer-control (media) usage to the second HID device.

    Parameters
    ----------
    name : str
        Consumer-control name (e.g. ``"PLAYPAUSE"``, ``"VOLUP"``, ``"HOME"``);
        case-insensitive.

    Returns
    -------
    None

    Raises
    ------
    KeyError
        If `name` is not a known consumer-control name.
    """
    name = (name or "").upper()
    if name not in _CONSUMER: raise KeyError(name)
    code = _CONSUMER[name]
    with open(HID_CONSUMER, 'rb+') as fd:
        fd.write(bytes([code & 0xFF, (code >> 8) & 0xFF])); fd.write(bytes(2))

ir_send

ir_send(name)

Send a learned IR code by shelling out to ir_tool.py.

Parameters:

Name	Type	Description	Default
name	str	Name of a previously learned IR code (e.g. "power").	required

Returns:

Type	Description
None

Raises:

Type	Description
KeyError	If name is empty or contains path separators (/ or ..).
RuntimeError	If the ir_tool.py send subprocess exits non-zero (e.g. the Broadlink is unreachable or the code is unknown).

Source code in src/hid_keyboard_server.py

def ir_send(name):
    """Send a learned IR code by shelling out to ``ir_tool.py``.

    Parameters
    ----------
    name : str
        Name of a previously learned IR code (e.g. ``"power"``).

    Returns
    -------
    None

    Raises
    ------
    KeyError
        If `name` is empty or contains path separators (``/`` or ``..``).
    RuntimeError
        If the ``ir_tool.py send`` subprocess exits non-zero (e.g. the Broadlink
        is unreachable or the code is unknown).
    """
    name = (name or "").strip()
    if not name or "/" in name or ".." in name: raise KeyError(name)
    env = dict(os.environ)
    r = subprocess.run([sys.executable, IR_TOOL, "send", name],
                       capture_output=True, text=True, timeout=15, env=env)
    if r.returncode != 0:
        raise RuntimeError((r.stderr or r.stdout).strip() or "ir send failed")

stream_start

stream_start(url)

Start an ffmpeg RTSP->HLS relay (remux only) for the live preview.

Any previous relay is stopped and the output directory is cleared first. Video is copied (no transcoding) and audio is dropped, so it is light enough for a Pi Zero. The RTSP_TRANSPORT setting is applied unless it is "auto".

Parameters:

Name	Type	Description	Default
url	str	The RTSP source URL.	required

Returns:

Type	Description
str	The path of the generated HLS playlist (/stream/live.m3u8).

Raises:

Type	Description
ValueError	If url is not an rtsp:// / rtsps:// URL.
RuntimeError	If the ffmpeg binary is not installed.

Source code in src/hid_keyboard_server.py

def stream_start(url):
    """Start an ffmpeg RTSP->HLS relay (remux only) for the live preview.

    Any previous relay is stopped and the output directory is cleared first.
    Video is copied (no transcoding) and audio is dropped, so it is light enough
    for a Pi Zero. The ``RTSP_TRANSPORT`` setting is applied unless it is
    ``"auto"``.

    Parameters
    ----------
    url : str
        The RTSP source URL.

    Returns
    -------
    str
        The path of the generated HLS playlist (``/stream/live.m3u8``).

    Raises
    ------
    ValueError
        If `url` is not an ``rtsp://`` / ``rtsps://`` URL.
    RuntimeError
        If the ffmpeg binary is not installed.
    """
    global _ffmpeg
    url = (url or "").strip()
    if not re.match(r'^rtsps?://', url, re.I):
        raise ValueError("only rtsp:// URLs can be relayed")
    stream_stop()
    os.makedirs(STREAM_DIR, exist_ok=True)
    for f in glob.glob(os.path.join(STREAM_DIR, "*")):
        try: os.remove(f)
        except OSError: pass
    cmd = [FFMPEG, "-nostdin", "-loglevel", "error", "-fflags", "nobuffer"]
    if RTSP_TRANSPORT and RTSP_TRANSPORT.lower() != "auto":
        cmd += ["-rtsp_transport", RTSP_TRANSPORT]
    cmd += ["-i", url, "-an", "-c:v", "copy",
            "-f", "hls", "-hls_time", "1", "-hls_list_size", "3",
            "-hls_flags", "delete_segments+append_list+omit_endlist",
            "-hls_segment_filename", os.path.join(STREAM_DIR, "seg_%05d.ts"),
            os.path.join(STREAM_DIR, "live.m3u8")]
    try:
        _ffmpeg = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except FileNotFoundError:
        raise RuntimeError("ffmpeg is not installed")
    return "/stream/live.m3u8"

stream_stop

stream_stop()

Stop the running ffmpeg relay, if any.

Terminates the process (escalating to kill on timeout) and resets the module-level handle. Safe to call when no relay is running.

Returns:

Type	Description
None

Source code in src/hid_keyboard_server.py

def stream_stop():
    """Stop the running ffmpeg relay, if any.

    Terminates the process (escalating to kill on timeout) and resets the
    module-level handle. Safe to call when no relay is running.

    Returns
    -------
    None
    """
    global _ffmpeg
    if _ffmpeg and _ffmpeg.poll() is None:
        try:
            _ffmpeg.terminate(); _ffmpeg.wait(timeout=3)
        except Exception:
            try: _ffmpeg.kill()
            except Exception: pass
    _ffmpeg = None

webrtc_start

webrtc_start(url, host)

Register an RTSP source with go2rtc and build a WebRTC player URL.

If url is given it is registered as the preview stream via the go2rtc API (trying PUT then POST); otherwise the preconfigured GO2RTC_STREAM is used. The returned URL uses GO2RTC_PUBLIC if set (for HTTPS / reverse-proxy setups) or the request host otherwise.

Parameters:

Name	Type	Description	Default
url	str	RTSP source URL, or empty to use the preconfigured stream.	required
host	str	The request's Host header, used to build the player URL when GO2RTC_PUBLIC is not set.	required

Returns:

Type	Description
str	An embeddable go2rtc webrtc.html URL.

Raises:

Type	Description
ValueError	If url is non-empty but not an rtsp:// / rtsps:// URL.
RuntimeError	If go2rtc cannot be reached to register the stream.

Source code in src/hid_keyboard_server.py

def webrtc_start(url, host):
    """Register an RTSP source with go2rtc and build a WebRTC player URL.

    If `url` is given it is registered as the ``preview`` stream via the go2rtc
    API (trying ``PUT`` then ``POST``); otherwise the preconfigured
    ``GO2RTC_STREAM`` is used. The returned URL uses ``GO2RTC_PUBLIC`` if set
    (for HTTPS / reverse-proxy setups) or the request host otherwise.

    Parameters
    ----------
    url : str
        RTSP source URL, or empty to use the preconfigured stream.
    host : str
        The request's ``Host`` header, used to build the player URL when
        ``GO2RTC_PUBLIC`` is not set.

    Returns
    -------
    str
        An embeddable go2rtc ``webrtc.html`` URL.

    Raises
    ------
    ValueError
        If `url` is non-empty but not an ``rtsp://`` / ``rtsps://`` URL.
    RuntimeError
        If go2rtc cannot be reached to register the stream.
    """
    name = GO2RTC_STREAM
    url = (url or "").strip()
    if url:
        if not re.match(r'^rtsps?://', url, re.I):
            raise ValueError("only rtsp:// URLs can be relayed")
        name = "preview"
        qs = urllib.parse.urlencode({"name": name, "src": url})
        api = GO2RTC_API.rstrip("/") + "/api/streams?" + qs
        last = None
        for method in ("PUT", "POST"):
            try:
                urllib.request.urlopen(urllib.request.Request(api, method=method), timeout=5).read()
                last = None; break
            except Exception as e:
                last = e
        if last is not None:
            raise RuntimeError("go2rtc not reachable: %s" % last)
    if GO2RTC_PUBLIC:
        base = GO2RTC_PUBLIC.rstrip("/")
    else:
        h = (host or "").split(":")[0] or "127.0.0.1"
        base = "http://%s:%s" % (h, GO2RTC_PORT)
    return "%s/webrtc.html?src=%s" % (base, name)

wait_for

wait_for(path, timeout=30)

Block until a filesystem path exists or a timeout elapses.

Parameters:

Name	Type	Description	Default
path	str	Path to wait for (e.g. the HID device node).	required
timeout	float	Maximum seconds to wait (default 30). Returns regardless once elapsed.	30

Returns:

Type	Description
None

Source code in src/hid_keyboard_server.py

def wait_for(path, timeout=30):
    """Block until a filesystem path exists or a timeout elapses.

    Parameters
    ----------
    path : str
        Path to wait for (e.g. the HID device node).
    timeout : float, optional
        Maximum seconds to wait (default ``30``). Returns regardless once
        elapsed.

    Returns
    -------
    None
    """
    dl = time.time() + timeout
    while not os.path.exists(path) and time.time() < dl: time.sleep(0.5)

ir_tool

The Broadlink RM4 IR helper (discover / learn / send).

ir_tool

pi-remote Broadlink RM4 IR helper (multi-device aware).

Requires the broadlink package (see docs/ir.md for armv6-safe install).

Usage: ir_tool.py discover # list every Broadlink device on the LAN ir_tool.py use # pick which device is the active blaster ir_tool.py learn [host] # learn an IR code from your original remote ir_tool.py send [host] # send a learned IR code ir_tool.py list # list saved codes

Paths are configurable via environment (defaults shown): PI_REMOTE_IR_STORE = /var/lib/pi-remote/ir_codes.json PI_REMOTE_IR_DEVFILE = /var/lib/pi-remote/ir_device.json

get_device

get_device(host=None, save=True)

Resolve and authenticate the Broadlink device to use.

Resolution order: the explicit host (matched against discovery), then the cached device in DEVFILE, then LAN discovery. With multiple devices and no host/cache, the user is asked to pick one and the process exits.

Parameters:

Name	Type	Description	Default
host	str	IP/host of a specific device to use. If omitted, use the cached device or auto-discover.	None
save	bool	Whether to persist the resolved device to DEVFILE (default True).	True

Returns:

Type	Description
Device	An authenticated device.

Raises:

Type	Description
SystemExit	If no device is found, the requested host is not present, or multiple devices exist and none was selected.

Source code in src/ir_tool.py

def get_device(host=None, save=True):
    """Resolve and authenticate the Broadlink device to use.

    Resolution order: the explicit `host` (matched against discovery), then the
    cached device in ``DEVFILE``, then LAN discovery. With multiple devices and no
    `host`/cache, the user is asked to pick one and the process exits.

    Parameters
    ----------
    host : str, optional
        IP/host of a specific device to use. If omitted, use the cached device or
        auto-discover.
    save : bool, optional
        Whether to persist the resolved device to ``DEVFILE`` (default ``True``).

    Returns
    -------
    broadlink.Device
        An authenticated device.

    Raises
    ------
    SystemExit
        If no device is found, the requested `host` is not present, or multiple
        devices exist and none was selected.
    """
    if host:
        match = None
        for d in broadlink.discover(timeout=5):
            if d.host[0] == host: match = d; break
        if not match: sys.exit("No Broadlink device found at %s" % host)
        match.auth()
        if save: _store_dev(match)
        return match
    info = _load(DEVFILE, None)
    if info:
        try:
            dev = broadlink.gendevice(info["devtype"], (info["host"], 80), bytes.fromhex(info["mac"]))
            dev.auth(); return dev
        except Exception: pass
    devs = broadlink.discover(timeout=5)
    if not devs: sys.exit("No Broadlink device found.")
    if len(devs) > 1:
        lines = "\n".join("  - %s (%s)" % (d.host[0], d.mac.hex()) for d in devs)
        sys.exit("Multiple Broadlink devices found - pick one with 'ir_tool.py use <host>':\n" + lines)
    devs[0].auth(); _store_dev(devs[0]); return devs[0]

cmd_discover

cmd_discover()

List every Broadlink device on the LAN (host, MAC, device type).

Returns:

Type	Description
None

Raises:

Type	Description
SystemExit	If no Broadlink device is found.

Source code in src/ir_tool.py

def cmd_discover():
    """List every Broadlink device on the LAN (host, MAC, device type).

    Returns
    -------
    None

    Raises
    ------
    SystemExit
        If no Broadlink device is found.
    """
    devs = broadlink.discover(timeout=5)
    if not devs: sys.exit("No Broadlink device found.")
    for d in devs:
        d.auth()
        print("Found %s  host=%s  mac=%s  devtype=0x%04x" % (d.type, d.host[0], d.mac.hex(), d.devtype))

cmd_use

cmd_use(host)

Select and cache the active IR blaster by host.

Parameters:

Name	Type	Description	Default
host	str	IP/host of the device to make the default (saved to DEVFILE).	required

Returns:

Type	Description
None

Raises:

Type	Description
SystemExit	If no device is found at host.

Source code in src/ir_tool.py

def cmd_use(host):
    """Select and cache the active IR blaster by host.

    Parameters
    ----------
    host : str
        IP/host of the device to make the default (saved to ``DEVFILE``).

    Returns
    -------
    None

    Raises
    ------
    SystemExit
        If no device is found at `host`.
    """
    dev = get_device(host, save=True)
    print("Active IR blaster set to %s (%s)." % (dev.host[0], dev.mac.hex()))

cmd_learn

cmd_learn(name, host=None)

Learn an IR code from the original remote and save it under name.

Puts the device into learning mode and polls for up to ~10 seconds while you press the button on the source remote.

Parameters:

Name	Type	Description	Default
name	str	Key to store the captured code under (e.g. "power").	required
host	str	Specific device to learn on; otherwise the cached/auto-resolved device.	None

Returns:

Type	Description
None

Raises:

Type	Description
SystemExit	If no device is available or no IR signal is captured before timeout.

Source code in src/ir_tool.py

def cmd_learn(name, host=None):
    """Learn an IR code from the original remote and save it under `name`.

    Puts the device into learning mode and polls for up to ~10 seconds while you
    press the button on the source remote.

    Parameters
    ----------
    name : str
        Key to store the captured code under (e.g. ``"power"``).
    host : str, optional
        Specific device to learn on; otherwise the cached/auto-resolved device.

    Returns
    -------
    None

    Raises
    ------
    SystemExit
        If no device is available or no IR signal is captured before timeout.
    """
    dev = get_device(host, save=False); dev.enter_learning()
    print(">> Aim your ORIGINAL remote at Broadlink %s and press the '%s' button now (10s)..."
          % (dev.host[0], name))
    packet = None
    for _ in range(10):
        time.sleep(1)
        try: packet = dev.check_data()
        except Exception: packet = None
        if packet: break
    if not packet: sys.exit("Timed out - no IR captured. Right device? Aim closer.")
    codes = _load(STORE, {}); codes[name] = base64.b64encode(packet).decode(); _save(STORE, codes)
    print("Saved '%s' (%d bytes)." % (name, len(packet)))

cmd_send

cmd_send(name, host=None)

Send a previously learned IR code.

Parameters:

Name	Type	Description	Default
name	str	Name of the saved code to transmit.	required
host	str	Specific device to send from; otherwise the cached/auto-resolved device.	None

Returns:

Type	Description
None

Raises:

Type	Description
SystemExit	If no code named name exists, or no device is available.

Source code in src/ir_tool.py

def cmd_send(name, host=None):
    """Send a previously learned IR code.

    Parameters
    ----------
    name : str
        Name of the saved code to transmit.
    host : str, optional
        Specific device to send from; otherwise the cached/auto-resolved device.

    Returns
    -------
    None

    Raises
    ------
    SystemExit
        If no code named `name` exists, or no device is available.
    """
    codes = _load(STORE, {})
    if name not in codes: sys.exit("No saved code named '%s'." % name)
    get_device(host, save=False).send_data(base64.b64decode(codes[name]))
    print("Sent '%s'%s." % (name, (" via %s" % host) if host else ""))