web_app/src/common/rpcService.ts

// Copyright 2024 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.

import { WebSerial, pw_hdlc, pw_rpc, pw_status } from "pigweedjs";
import { ProtoCollection } from "../../protos/collection/collection";
import { BlinkRequest } from "../../protos/collection/blinky/blinky_pb";
import {
  OnboardTempStreamRequest,
  OnboardTempResponse,
} from "../../protos/collection/board/board_pb";
import {
  MeasureStreamRequest,
  Measurement,
} from "../../protos/collection/air_sensor/air_sensor_pb";
import { State } from "../../protos/collection/state_manager/state_manager_pb";
class RPCService {
  transport;
  decoder;
  encoder;
  rpcAddress;
  channels;
  client;
  blinkService;
  boardTempService;
  measureService;
  stateService;
  constructor(rpcAddress = 82) {
    this.transport = new WebSerial.WebSerialTransport();
    this.decoder = new pw_hdlc.Decoder();
    this.encoder = new pw_hdlc.Encoder();
    this.rpcAddress = rpcAddress;

    this.channels = [
      new pw_rpc.Channel(1, (bytes: Uint8Array) => {
        const hdlcBytes = this.encoder.uiFrame(this.rpcAddress, bytes);
        this.transport.sendChunk(hdlcBytes);
      }),
    ];
    this.client = pw_rpc.Client.fromProtoSet(
      this.channels,
      new ProtoCollection(),
    );
    this.blinkService = this.client.channel().methodStub("blinky.Blinky.Blink");
    this.boardTempService = this.client
      .channel()
      .methodStub("board.Board.OnboardTempStream");
    this.measureService = this.client
      .channel()
      .methodStub("air_sensor.AirSensor.MeasureStream");
    this.stateService = this.client
      .channel()
      .methodStub("state_manager.StateManager.GetState");
  }

  async connect() {
    await this.transport.connect();

    this.transport.chunks.subscribe((item: any) => {
      const decoded = this.decoder.process(item);
      for (const frame of decoded) {
        if (frame.address === this.rpcAddress) {
          this.client.processPacket(frame.data);
        }
      }
    });
  }

  async disconnect() {
    await this.transport.disconnect();
  }

  async blink(times: number, intervalMs: number = 300) {
    const req = new BlinkRequest();
    req.setBlinkCount(times);
    req.setIntervalMs(intervalMs);
    const [status, response] = await this.blinkService.call(req);
  }

  async streamTemp(onTemp?: (temp: number) => void) {
    const req = new OnboardTempStreamRequest();
    req.setSampleIntervalMs(500);
    await this.boardTempService.invoke(
      req,
      (m: OnboardTempResponse) => {
        console.log(m.toObject());
        if (onTemp) onTemp(m.getTemp());
      },
      undefined,
      (err) => {
        console.error(err);
      },
    );
  }

  async streamMeasure(onMeasure?: (measurement: Measurement) => void) {
    // We check if this RPC exists on device,
    // if not we fallback to just onboardTemp.
    return new Promise(async (resolve, reject) => {
      let resolveCalled = false;
      const req = new MeasureStreamRequest();
      req.setSampleIntervalMs(500);
      await this.measureService.invoke(
        req,
        (m: Measurement) => {
          if (!resolveCalled) {
            resolveCalled = true;
            resolve(true);
          }
          if (onMeasure) onMeasure(m);
        },
        undefined,
        (status: pw_status.Status) => {
          if ((status = pw_status.Status.NOT_FOUND)) {
            reject(status);
          }
        },
      );
    });
  }

  async getState(): Promise<State> {
    const [status, response] = await this.stateService.call();
    return response;
  }
}

// We keep a singleton of this service.
const rpc = new RPCService();

export function getRpcService(): RPCService {
  return rpc;
}