Pythonでドラゴン曲線

ドラゴン曲線を描く.

実装

線分の始点から終点へ向かうベクトルを回転させて 倍する処理を繰り返すように実装した. 回転させる角度はとが交互になるようにする.

def dragon(i, s, e):
    res = []
    def R(t):
        t = t *  np.pi / 4 
        return np.array([
            [np.cos(t), -np.sin(t)],
            [np.sin(t),  np.cos(t)]
        ]) / np.sqrt(2)

    def _dragon(gen, t, s, e):
        if gen == 0:
            return

        m = R(t)@(e - s) + s
        _dragon(gen - 1,  1, s, m)
        res.append(m)
        _dragon(gen - 1, -1, m, e)

    res.append(s)
    _dragon(i, 1, s, e)
    res.append(e)

    return np.array(res)

回帰回数による変化

雑記

アニメーションは最初APNGで作成したのだが, はてなブログでは 表示されなかったので泣く泣くgifへ変換した.

参考

• ドラゴン曲線 - Wikiwand

FastAPIを使用した画像処理APIの実装

概要

画像を入力とし, 画像を返すAPIをFastAPIで実装する. (FastAPIでAPI部分を実装するより, jsでStream APIを使用するほうが大変であった)

実装

準備

$pip install pillow fastapi jinja2 aiofiles python-multipart uvicorn

API側の実装

画像処理の内容は入力された画像をタイル状に並べるだけのかんたんな処理

UploadFileのリストとして受け取り, StreamingResponseとして返す.

データの読み書きはByteIOを使用して行う.

ByteIOを使用してpillowの画像を返す場合は, 返す前にseekで先頭に戻す必要がある.

@app.post('/api/image-processing')
async def create_image_processing(files: List[UploadFile] = File(...)):
    # open image
    bytes_io = BytesIO(files[0].file.read())
    image = Image.open(bytes_io).convert('RGB')

    # image processing
    data = np.array(image)
    h, w, _ = data.shape
    h = int(h // 2) * 2
    w = int(w // 2) * 2
    data = data[:h, :w, :] \
        .reshape(h // 2, 2, w // 2, 2, -1) \
        .transpose(1, 0, 3, 2, 4) \
        .reshape(h, w, -1)
    content = BytesIO()
    Image.fromarray(data).save(content, format='png')
    content.seek(0)

    # response
    return StreamingResponse(content, media_type='image/png')

フロント側の実装

画像を送信して, そのレスポンスのReadableStreamを処理して表示する処理.

MDN Web Docsのサンプルをそのまま使用している.

postBtn.addEventListener("click", () => {
  //
  if (!imageFile) {
    console.error("no image file");
    return;
  }

  const body = new FormData();
  body.append("files", imageFile);

  fetch("/api/image-processing", {
    method: "POST",
    body: body,
  })
    .then((resp) => {
      const reader = resp.body.getReader();

      return new ReadableStream({
        start(controller) {
          return pump();

          function pump() {
            return reader.read().then(({ done, value }) => {
              if (done) {
                controller.close();
                return;
              }
              controller.enqueue(value);
              return pump();
            });
          }
        },
      });
    })
    .then((stream) => new Response(stream))
    .then((resp) => resp.blob())
    .then((blob) => {
      document.getElementById("output-img").src = URL.createObjectURL(blob);
    });
});

出力結果

画像処理API 例

リポジトリ

github.com

参考資料

• FastAPIとTensorflowで簡単な画像認識APIを作ってみた - Qiita
• Custom Response - HTML, Stream, File, others - FastAPI
• Static Files - FastAPI
• 読み取り可能なストリームの使用 - Web API | MDN
• python - How do I return an image in fastAPI? - Stack Overflow

FastAPIでバックグラウンド処理

概要

fastapi で重い処理を走らせるときの処理パターン

実装

成功するときの系だけで失敗したときのときの処理を書いていないので注意

流れとしては タスク作成 → タスクの進捗確認 → タスクの結果取得

タスク

時間のかかるタスクとして10秒間sleepするタスクを実装

タスクの結果は単純に result data という文字列を返す

class Task:
    def __init__(self):
        self.id = str(uuid.uuid4())
        self.progress = 0
        self.res = None

    def __call__(self):
        for _ in range(10):
            time.sleep(1)
            self.progress += 10
        self.res = 'result data'


TASKS = {}

タスク作成

重いタスクを作成するAPI

作成したタスクのIDを返す

@app.post('/task', status_code=202)
async def create_task(background_tasks: BackgroundTasks):
    task = Task()
    TASKS[task.id] = task
    background_tasks.add_task(task)
    return {'task_id': task.id}

タスク進捗取得

タスクの進捗度合いを取得するAPI

状態と進捗度合いとタスク結果のURIを返す

@app.get('/task/{task_id}')
async def read_task(task_id: str):
    if task_id not in TASKS:
        return {'message': 'nope'}

    task = TASKS[task_id]

    if task.res is None:
        return {
            'status': 'IN_PROGRESS',
            'progress': task.progress,
            'uri': None
        }
    else:
        return {
            'status': 'SUCCEEDED',
            'progress': 100,
            'uri': f'/task/result/{task.id}'
        }

タスク結果取得

タスクの結果を取得するAPI

@app.get('/task/result/{task_id}')
async def read_result(task_id: str):
    if task_id in TASKS:
        task = TASKS[task_id]
        return {'result': task.res}
    else:
        return {'result': 'nothing'}

リポジトリ

github.com

参考資料

• Background Tasks - FastAPI
• Long polling
• REST原則に従った非同期APIの設計 - Qiita
• 非同期サーバーのFastAPIでバックグラウンドで重い処理を書く - Qiita

AVR-Rust + ATtiny13A でLチカ

AVR-Rustが公式にマージされた.
せっかくなので使ってみたいが, Arduinoはチュートリアルで使用していてつまらない.
なので今回は秋月電子で購入可能な格安AVRマイコン ATtiny13 をAVR-Rust を使用してLチカした.

主な実行環境

• OS: Ubuntu 20.04.1 LTS
• cargo: 1.47.0-nightly
• avrdude: 6.3-20171130
• AVRマイコン: ATtiny13A

• シリアル変換モジュール: FT232RL

• 抵抗とLED: 床に落ちてたヤツ

今回の実装は以下のリポジトリにある github.com

書き込む環境の準備

まずATtiny13aにプログラムを書き込む環境を準備する必要がある.

私はここで大きく躓いた.

ポート指定にシリアルポート/dev/ttyUSB0を指定すれば良いと思ったがうまく行かない. どうやらシリアルナンバを指定する必要があったようだ.

シリアルナンバはdmesgで調べることができる.

$ dmesg
...
[44085.473422] usb 3-2: new full-speed USB device number 10 using xhci_hcd
[44085.626927] usb 3-2: New USB device found, idVendor=0403, idProduct=6001, bcdDevice= 6.00
[44085.626931] usb 3-2: New USB device strings: Mfr=1, Product=2, SerialNumber=3
[44085.626934] usb 3-2: Product: FT232R USB UART
[44085.626936] usb 3-2: Manufacturer: FTDI
[44085.626937] usb 3-2: SerialNumber: A505SYG5
[44085.652102] usbcore: registered new interface driver usbserial_generic
[44085.652109] usbserial: USB Serial support registered for generic
[44085.655738] usbcore: registered new interface driver ftdi_sio
[44085.655746] usbserial: USB Serial support registered for FTDI USB Serial Device
[44085.655838] ftdi_sio 3-2:1.0: FTDI USB Serial Device converter detected
[44085.655870] usb 3-2: Detected FT232RL
[44085.656193] usb 3-2: FTDI USB Serial Device converter now attached to ttyUSB0

シリアルナンバはA505SYG5であるとわかった.

シリアルポート設定が判明したところで正しく接続できるか確認する.

$ sudo avrdude -c diecimila -p t13 -P usb:A505SYG5 -B 4200

avrdude: AVR device initialized and ready to accept instructions

Reading | ################################################## | 100% 0.12s

avrdude: Device signature = 0x1e9007 (probably t13)

avrdude: safemode: Fuses OK (E:FF, H:FF, L:6A)

avrdude done.  Thank you.

たぶんできている.

実装

#![feature(asm, lang_items, unwind_attributes)]
#![no_std]
#![no_main]

extern crate avr_delay;
extern crate avr_std_stub;
extern crate avrd;

use avr_delay::delay;

use avrd::attiny13a::{DDRB, PORTB};

#[no_mangle]
pub extern "C" fn main() {
    unsafe {
        core::ptr::write_volatile(DDRB, 0x08);
    }
    let mut out = 0x08;
    loop {
        unsafe {
            core::ptr::write_volatile(PORTB, out);
        }
        delay(1_200_00);
        out ^= 0xFF;
    }
}

書き込み

動作確認

所感

参考

AVRDUDE関連

Atmega328p FT232RL invalid device identifier FT232RLでAVRライターを自作してATtiny85をDigispark互換にするまで - Qiita kemarin-tech : avrdudeでAVRにバイナリコードを書き込む（Arduinoボードで書き込み）

AtCoder ABC177

今回はどこでバグっているのかわからなくて,延々と時間を浪費してしまった.

A

use proconio::input;

#[allow(non_snake_case)]
fn main() {
    input! {
        D: usize,
        T: usize,
        S: usize,
    }

    if (D + S - 1) / S > T {
        println!("No");
    } else {
        println!("Yes");
    }
}

B

use proconio::input;
use proconio::marker::Chars;
use std::cmp;

#[allow(non_snake_case)]
fn main() {
    input! {
        S: Chars,
        T: Chars,
    }

    let mut ans = 1001;
    let sl = S.len();
    let tl = T.len();
    for i in 0..=(sl - tl) {
        let mut cnt = tl;
        for j in 0..tl {
            if S[i + j] == T[j] {
                cnt -= 1;
            }
        }
        ans = cmp::min(ans, cnt);
    }

    println!("{}", ans);
}

C

use proconio::input;

#[allow(non_snake_case)]
fn main() {
    input! {
        N: usize,
        A: [u64; N]
    }

    let M: u64 = 1_000_000_007;

    let mut ans = 0;
    let mut cum = 0;
    for a in A.iter().rev() {
        ans = (ans % M + (a * cum) % M) % M;
        cum = (cum % M + a % M) % M;
    }

    println!("{}", ans % M);
}

D

こいつの答えが合わなくて時間を浪費してしまった. 結局の所原因は, union-findをしっかり理解していないところにある.

use proconio::input;
use proconio::marker::Usize1;

fn find(x: usize, par: &mut Vec<usize>, rnk: &Vec<usize>) -> usize {
    //
    if par[x] == x {
        return x;
    } else {
        par[x] = find(par[par[x]], par, rnk);
        return par[x];
    }
}

fn unite(x: usize, y: usize, par: &mut Vec<usize>, rnk: &mut Vec<usize>) {
    let x = find(x, par, rnk);
    let y = find(y, par, rnk);
    if x == y {
        return;
    }

    if rnk[x] < rnk[y] {
        par[x] = y;
    } else {
        par[y] = x;
        if rnk[x] == rnk[y] {
            rnk[x] += 1;
        }
    }
}

#[allow(non_snake_case)]
fn main() {
    input! {
        N: usize,
        M: usize,
        AB: [(Usize1, Usize1); M]
    }

    let mut par: Vec<_> = (0..N).collect();
    let mut rnk = vec![0; N];

    for &(a, b) in AB.iter() {
        unite(a, b, &mut par, &mut rnk);
    }

    for i in 0..N {
        find(i, &mut par, &mut rnk);
    }
    let mut cnt = vec![0; N];
    for i in 0..N {
        cnt[par[i]] += 1;
    }

    println!("{}", cnt.iter().max().unwrap());
}

タクトスイッチ12個を3ピンで制御する

tl; dr.

• シフトレジスタを使用してボタン入力のピン数を削減
• 必要ピン数(12本)では8bitシフトレジスタでは足りないので, 2つ繋げて使用
• 最終的に出力1ピン, 入力2ピンで構成(正確にはこれにVccとGNDが加えられ5ピン)

動機

せっかくArduinoを購入したのだから, 電子工作でシンセサイザー的な何かを作ろうと思った.
しかし, どう考えたってピンが足りないのでシフトレジスタを使用してピン数を節約した.

使用部品

• 8bit シフトレジスタ 74HC165AP
• プルダウン抵抗 10 [kΩ] x 12個
• 積層セラミックコンデンサ 0.1 [uF] x 2
• タクトスイッチ x 12 (鍵盤なので白7個, 黒5個)

あと実際の完成品には電源確認用のLEDを1つ付けてある.

構成

執筆中

回路

回路図

完成図

完成品

コード

執筆中

動作確認

執筆中

雑記

載せてはいないが今回は配線が非常にキレイにできたので満足している. ただ毎回のようにハンダ付場所をミスるのでどうにかしてほしい.

自作 4桁7セグメント LED 表示器を I2C で制御

[Unfinished]

• マイコン: ATTINY85
• 7セグメント ドライバ: TC4511BP
• 8bit シフトレジスタ: U74HC595AG-D16-T
• 4桁7セグメントLED(カソードコモン): OSL40562-LRA その他抵抗とかコンデンサとかmosfetとか

ブレッドボード上での試作

完成品

(ところで, 試作時にあったデカップリングコンデンサはどこへいった?)

avrdudeがlinuxでattiny85や13へ書き込めなかったの謎

WebAssembly +Rust で Kalman Filter

github.com

https://shimo8810.github.io/kalman-wasm/

AtCoder ABC 176

F問題以外は自力で解けた(時間には間に合っていない).

rust-analyzerくんがクッソ優秀過ぎて, これの型推論と表示がなければ何もかけない体になってしまった.

A: Takoyaki

use proconio::input;
 
#[allow(non_snake_case)]
fn main() {
    input! {
        N: u32,
        X: u32,
        T: u32,
    }
 
    let ans = (N + X - 1) / X * T;
 
    println!("{}", ans);
}

B: Multiple of 9

9 の倍数判定を行う問題. 問題の値が大きすぎるので u64 (1e19 程度)はおろか u128 (1e28 程度)でも収まらないので工夫が必要.

9 の倍数はすべての桁を足し合わせた値もまた 9 の倍数になるので,すべて足し合わせれば良い. 最大で 9 * 200000 = 1.8 * 1e6 なので u32 で収まる.

すべての桁を確認するので時間計算量は .

use proconio::input;
use proconio::marker::Chars;
 
#[allow(non_snake_case)]
fn main() {
    input! {
        N: Chars,
    }
 
    let mut sum = 0;
    for c in N {
        sum += c as u32 - '0' as u32;
    }
 
    if sum % 9 == 0 {
        println!("Yes");
    } else {
        println!("No");
    }
}

C: Step

広義単調増加な数列になるように追加する値の最小値を求める問題. 1つ前の値と比較してより小さければ, 同じ高さになるように値を追加していけば良い.

最終結果は最大で (先頭だけ高くて他が低いパターン) なのでu64が必要.

すべての値を確認するので 時間計算量は .

時間計算量

use proconio::input;
 
#[allow(non_snake_case)]
fn main() {
    input! {
        N: usize,
        A: [u64; N]
    }
 
    let mut ans = 0;
    let mut pre = 0;
    for a in A {
        if pre > a {
            ans += pre - a;
        } else {
            pre = a;
        }
    }
 
    println!("{}", ans);
}

D: Wizard in Maze

ゴールへの最小ワープ回数を求める問題. 隣のマスへの移動にはコストがかからず, ワープするときのみコストがかかる. なのでワープで繋がれた地続き領域間の幅優先探索で解くことができる.

01-bfsを用いることによって, 地続きを先に探索してワープ先をあとに探索するようにできる.

今回はdequeに追加するときではなく取り出したときに探索済みチェックをつけている. ワープ可能なマスを問答無用でdequeに打ち込んでいるので, この辺ちゃんと刈ればもっと高速化が可能で, チェックも追加するときにつけられる.

proconio::marker::Isze1を使用しているのはx + dxなどが負になりうるから. (usizeでもアンダーフローしてstd::usize::MAXになるので結局範囲外になるから耐えるかも?)

すべてのマスに対して探索を行うので 時間計算量 .

use proconio::input;
use proconio::marker::{Chars, Isize1};
use std::collections::{HashSet, VecDeque};
 
#[allow(non_snake_case)]
fn main() {
    input! {
        H: isize,
        W: isize,
        ch: Isize1,
        cw: Isize1,
        dh: Isize1,
        dw: Isize1,
        S: [Chars; H]
    }
    let mut set = HashSet::new();
    let mut deque = VecDeque::new();
 
    deque.push_front((ch, cw, 0));
    while let Some((y, x, d)) = deque.pop_front() {
        // 終了条件
        if y == dh && x == dw {
            println!("{}", d);
            return;
        }
 
        if set.contains(&(y, x)) {
            continue;
        }
        set.insert((y, x));
 
        for dx in -2..=2isize {
            for dy in -2..=2isize {
                // 中心は除外
                if dx == 0 && dy == 0 {
                    continue;
                }
                // 座標決め
                let h = y + dy;
                let w = x + dx;
 
                // 範囲外は除外
                if h < 0 || H <= h || w < 0 || W <= w {
                    continue;
                }
 
                // 移動可能でなければ除外
                if S[h as usize][w as usize] != '.' {
                    continue;
                }
                if dx.abs() + dy.abs() == 1 {
                    if !set.contains(&(h, w)) {
                        deque.push_front((h, w, d));
                    }
                } else {
                    if !set.contains(&(h, w)) {
                        deque.push_back((h, w, d + 1));
                    }
                }
            }
        }
    }
 
    println!("-1");
}

E: Bomber

ボンバーマンにおいて爆弾1つで破壊できる最大個数を求める問題. 結局の所, 爆弾が行と列でそれぞれ最大個数おいてある組が解答になる.

ただし, 行と列の交点に対象物がある場合は-1される. よって爆弾が行と列でそれぞれ最大個数おいてある組のうち, 交点に爆弾が無いものがあればそれが答え. なければそこから-1したものが答えになる.

これは最大個数の組の交点数(= 行 x 列)と最大個数の行と列に置かれている爆弾の個数を比較すると求められる.

すべての爆弾, すべての行, すべての列をそれぞれなめるので時間計算量はかな?

use proconio::input;
use proconio::marker::Usize1;
 
#[allow(non_snake_case)]
fn main() {
    input! {
        H: usize,
        W: usize,
        M: usize,
        T: [(Usize1, Usize1); M]
    }
 
    let mut count_h = vec![0; H];
    let mut count_w = vec![0; W];
 
    for &(y, x) in T.iter() {
        count_h[y] += 1;
        count_w[x] += 1;
    }
    let max_h = count_h.iter().max().unwrap();
    let max_w = count_w.iter().max().unwrap();
 
    let a = count_h.iter().filter(|y| *y == max_h).count();
    let b = count_w.iter().filter(|x| *x == max_w).count();
    let c = T
        .iter()
        .filter(|(y, x)| count_h[*y] == *max_h && count_w[*x] == *max_w)
        .count();
 
    println!("{}", max_h + max_w - if a * b == c { 1 } else { 0 });
}

F Brave CHAIN

解き方がわからなかったので答えを見ました. 頑張って書いています.

Raspberry Piのブート/シャットダウンスイッチの実装

先日, Raspiの電源をUSBを引き抜いて切断しているところを友人に苦言を呈されたので, シャットダウンスイッチを実装することにした.
ついでなのでブートスイッチも実装した.

コードはここに github.com

実装内容

• ブート/シャットダウンスイッチは同じスイッチを使用する
• スイッチを1回押すことでRaspiがブートする
• スイッチを長押しすることでRaspiがシャットダウンする

ブートボタン

halt状態でGPIO3をGNDに落とせば起動するらしい.(詳しくは参考サイトを参照)
押したときにLOWになるようにスイッチをプルアップ回路で実装する.

プルアップ回路

シャットダウンボタン

シャットダウンボタンはスイッチが長押しされたときにシャットダウンコマンドを実行するプログラムを実装する必要がある.
あとは実装したプログラムをsystemdで自動起動するように登録すれば良い.

シャットダウン用コード

シャットダウンプログラムの内容は 1. スイッチが押されるまで待機 2. スイッチが押されたら長押し判定 3. 長押しされたらシャットダウンコマンドを実行 だけである.

今回はRustで実装し, aspiのGPIO操作はrppalを使用した.
unwrapばかりでエラー処理をまともに書いていないので注意.

以下コード.

use std::process::Command;
use std::thread;
use std::time::Duration;

use rppal::gpio::Gpio;
use rppal::gpio::Trigger;

const BTN_PIN: u8 = 3;
const COUNT_MAX: u16 = 400;
const WAIT_MS: u64 = 10;

fn main() {
    // config gpio
    let gpio = Gpio::new().unwrap();
    let mut btn_pin = gpio.get(BTN_PIN).unwrap().into_input();

    // set interrupt
    btn_pin.set_interrupt(Trigger::FallingEdge).unwrap();

    loop {
        btn_pin.poll_interrupt(true, None).unwrap();

        let mut counter = 0;

        loop {
            if btn_pin.is_low() {
                counter += 1;
                if counter > COUNT_MAX {
                    // exec shutdown command
                    Command::new("shutdown")
                        .args(&["-h", "now"])
                        .output()
                        .expect("failed to execute process");
                    break;
                }
            } else {
                break;
            }

            thread::sleep(Duration::from_millis(WAIT_MS));
        }
    }
}

ビルド

コードをビルドしてインストールする.

$ cargo install --path .

このときerror: could not compile `libc`.のようなエラーが出るときが有るので, その時は Cargo.tomlファイルに

[profile.release]
codegen-units = 1

と記述するとコンパイルできるようになる.

詳しくは以下を参照. github.com

cargoでインストールするとcargo配下にインストールされる.

$ which shutd
/home/pi/.cargo/bin/shutd

クロスコンパイルする場合

またraspiでビルドするのが嫌な場合は他のマシンでクロスコンパイルするという方法がある.
クロスコンパイルにはcrossを使用すると良い.

crossを使用してビルドした成果物が動かない場合はprofileの修正をするかgnueabihfからmusleabihfに変えてみると良いかもしれない.

生成物はsftpなりrsyncなりでraspiの好きなディレクトリに投げつけて終了.

systemdへの登録

/usr/lib/systemd/system/shutd.serviceに以下を追記

[Unit]
Description=Shutdown Daemon

[Service]
ExecStart=/home/pi/.cargo/bin/shutd
Restart=always
Type=simple

[Install]
WantedBy=multi-user.target

ExecStartにはシャットダウンプログラムのあるパスを指定すること.

そしてサービスを開始&自動起動設定.

$ sudo systemctl enable shutd.service
$ sudo systemctl start shutd.service

設定できているか確認.

$ systemctl status shutd.service 
● shutd.service - Shutdown Daemon
   Loaded: loaded (/usr/lib/systemd/system/shutd.service; enabled; vendor preset: enabled)
   Active: active (running) since Sun 2020-05-10 18:02:55 JST; 2min 26s ago
 Main PID: 1630 (shutd)
   Memory: 168.0K
   CGroup: /system.slice/shutd.service
           └─1630 /home/pi/.cargo/bin/shutd

May 10 18:02:55 black systemd[1]: Started Shutdown Daemon.

完成品

使用したボタンはLEDが入っているヤツを使用しています.

完成品

参考サイト

• Raspberry Pi 3 にシャットダウン/リブート/再開ボタンを追加する - はむ！の空想具現化したいブログ
• クラゲのIoTテクノロジー
• ラズパイでシャットダウンボタンを付ける(ついでに起動ボタン) - Qiita
• https://docs.golemparts.com/rppal/0.11.3/rppal/gpio/struct.InputPin.html
• RPI safe mode - eLinux.org
• can not compile in release mode in raspberry pi zero · Issue #69420 · rust-lang/rust · GitHub