コード書いてますか
当社はHULFTを開発しているベンダーです。ユーザーのDXを支援するためのツール・ソリューションを開発・販売・サポートしています。
今年はHULFT SquareというiPaaSをローンチしました。HULFT開発については、当然ながらそれに関わるエンジニアは日々コードを書いています。
一方、SI業務も当社の生業の半分を占めていますが、伝統的なPjM業務に加えて、ノーコードツール、SaaS、PaaSの活用が広がり、コードを書くということが少なくなってきています。
私自身も今年の4月からCTOになりめっきりとコードを書くことがなくなりました。そこでせっかくのアドベントカレンダーということもあって、久しぶりにコードを書いてみたので改めて、現代のコードを書くことについて振り返りたいと思います。
コードを書いているけど、ちょっと最近楽しくないなとか、悩んでいる人はぜひご覧ください。
コードを書く前に必要なこと
HULFT Squareは日米共同で開発しています。私はそこで4年間PdMを担当しています。その際感じたことの詳細はこちらに記載していますが、コードを書くということは何かの目的を達成するためです。
やってみた系も大事ですが、実現したいことがあるともっと楽しいです。
私はHULFT Squareの要件を書くときはProject Charterというのを書きます。これはこの４年間でUSチームから学んだ手法です。
大きな項目としては以下になります。どのように実装するかや、細かいUI、非機能などはアーキテクトや開発チームを信頼してほぼ全部任せています。
- Purpose（ユーザーの解決したいものとかペインポイントなど）
- Objectives 何を実現するか
- 成果物
- スコープアウトにするもの
今回、コードを書くにあたり、これに沿って作りたいものを書いてみました。
Purpose 目的
私はCTOとして毎週月曜日のお昼に社員向けに先週の経営会議や本部長会などの重要情報や技術情報をラジオのようにZoomで話している。録画した動画ファイルをBoxに保存しているが、ユニークな視聴者数はライブ視聴と合わせても毎回70名程度だ。全社員の10％程度である。
有意義な情報であるため、もっと多くの社員に聞いてほしい。何よりも会社の動きをもっと知ってもらい、行動のサイロを減らしていきたい。
Objectives 実現したいこと
- 動画ファイルを参照することは時間がかかる。サマリ情報をテキスト化してSlackで配布することで、興味のない人の関心を引きたい
- 動画ファイルの情報は経営情報など機微なデータが含まれており、動画の変換はSaaSを利用するのではなく、情報漏洩が絶対に発生しない仕組みとすること。
- ユーザーに届けるまでの処理はできる限り自動化し、新たな業務を増やさないこと
成果物
- Zoom録画を安全にローカルに保存する仕組み
- 動画をオンプレミスでテキストに変換するアプリケーション
- テキスト情報をサマリして、Slackに投稿する仕組み
上記の要件を元にTechnical Requrirement Documentを書くことになりますが、今回は量も多くなるのでそこは割愛します。
さらに今回は上記成果物の２つ目の部分だけのコードについて話を進めていきます。
現代のコードを書くということ
SIであれば誰かが詳細設計書まで書いて、プログラマがそれを上から順に実装していくことになりますが、それだとプログラマの創造性を発揮できません。プログラミングは本当に創造的な作業です。要件定義フェーズで考えたことが最強であったことはかつて一度もないと思います。コードを書きながら、動かしながら改善していくのです。これがプログラマの仕事であり、アプリケーションの価値向上に責任を持つのはプログラマも同じなのです。
それを変更管理でやろうとするとスピードと意欲が削がれます。要件を作る人とコードを書く人は近ければ近い方がいいのです。
では、先の要件に沿ってコードを部分的に書きながら話を進めていきます。
動画をテキストに変換する
オンプレミスでの実行が要件なので、オンプレミスで動作するOpenAI Whisperを使うのが妥当ですよね。そこから書き始めます。一番早めにフィージビリティをとらなきゃいけないところからやりたいですよね。
まあ、最初はこんな感じですね。ちょっといきなり、cuda使ってるので反則な感じもしますが、CPUを使うと遅すぎるのはわかっているので、GPUで計算するようにしています。こういうのは経験の差ですのでたくさん地雷を踏むしかありません。
model = whisper.load_model("large", device="cuda")
_ = model.half()
_ = model.cuda()
for m in model.modules():
if isinstance(m, whisper.model.LayerNorm):
m.float()
result = model.transcribe("sample.mp4", language="ja")
動かしながら、今はLarge-v3の方がいいとか、initial_promptを使った方が出力精度が上がるとかそういうことを試しながら行います。
ちなみにこのプログラムはGTX4090を搭載した端末で動作させているので1時間の動画であれば6分程度でテキストに変換できます。
抽出したテキストを書き出す
次にWhisperで読み込んだテキストをファイルに書き出します。
text_path = path + ".txt"
with open(text_path, "w") as text_file:
text_file.write(result["text"])
print(f"Saved transcript to {text_path}")
Windowsで動かしていると結構書き出し中にエラーが出るんですね。encoding='utf-8', errors='ignore'みたいな属性を入れつつ、これらを回避していきます。Whisperが出力する日本語が変換不可な例が出たのでそれを抑制しています。
ファイルの取得
次はファイルの探索ですね。いつまでもソースコードに直接ファイル名を記述するわけにはいかないので。今回は特定ディレクトリのファイル監視をしたいと思います。自動化したいので、プログラム自体は常駐させて、フォルダにファイルが置かれたら、Whisperを動かすようにしたい。ここではwatchdogを使います。
そうやって動かしながら試していると、次に複数ファイルが同時に置かれた場合の処理も必要になってくるなということで最終的にはQueueを使った順次処理になりました。
こういうのはプログラマが考えた方が圧倒的に生産性高いですよね。要件フェーズで決めるよりも、より最適な選択が取れると思います。決定を遅延するという手法です。だからこそ、プログラマには高いスキルが求められると思います。
一旦のコードの完成
最終的なコードは以下となります。
import os
import threading
import time
from queue import Queue
import whisper
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
class Watcher:
DIRECTORY_TO_WATCH = "E:\\Share\\toText"
def __init__(self):
self.observer = Observer()
self.queue = Queue()
def run(self):
event_handler = Handler(self.queue)
self.observer.schedule(event_handler, self.DIRECTORY_TO_WATCH, recursive=True)
self.observer.start()
Processor(self.queue).start()
try:
while True:
time.sleep(5)
except:
self.observer.stop()
print("Error")
self.observer.join()
class Handler(FileSystemEventHandler):
def __init__(self, queue):
self.queue = queue
def on_any_event(self, event):
if event.is_directory or not event.event_type == 'created':
return None
# ファイルが作成されたときのイベント
print(f"Received created event - {event.src_path}.")
self.queue.put(event.src_path)
class Processor(threading.Thread):
def __init__(self, queue):
threading.Thread.__init__(self)
self.queue = queue
def run(self):
while True:
path = self.queue.get()
if path.endswith('.mp4') or path.endswith('.m4a'):
self.process_file(path)
self.queue.task_done()
def process_file(self, path):
print(f"Processing file: {path}")
model = whisper.load_model("large-v3", device="cuda")
_ = model.half()
_ = model.cuda()
for m in model.modules():
if isinstance(m, whisper.model.LayerNorm):
m.float()
result = model.transcribe(path, verbose=True, language="ja",
initial_prompt="HULFT Square iPaaS です。 ます。 でした。")
# テキストをファイルに保存
text_path = path + ".txt"
with open(text_path, "w", encoding='utf-8', errors='ignore') as text_file:
text_file.write(result["text"])
print(f"Saved transcript to {text_path}")
# 元の動画ファイルを削除
os.remove(path)
print(f"Deleted original file {path}")
if __name__ == '__main__':
w = Watcher()
w.run()
サマリ
最初にも書きましたが、アプリケーションというのは１回書いたら終わりではありません。このプログラムを常駐させることで、さらに厄介な問題も出るかもしれない。それが出たら直せばいいのです。それを高速に、そして安全に実現するためにマイクロサービスとか、CI/CDとかの進化が一方で進んでいるんだと思います。
プログラマはコードを書くだけが仕事ではありません。現代のプログラマはクラウドと現代のAIそして、最新のマシンスペックを利用することで魔法使いになれます。
そのためになぜコードを書くのか、このコードは誰の何を解決するのか？そういうことを考えて年末、年始にコードを書いてみてください。
とても面白い変化が自身に起きることは間違いありません。
年末、年始のおすすめとして、こちらの本「世界一流エンジニアの思考法」や「メタ思考～「頭のいい人」の思考法を身につける」は今回の話における考え方としてとても良いと思うのでぜひ読んでみてください。
あと、私が20年前から何回も読んでいる「考える技術・書く技術 問題解決力を伸ばすピラミッド原則」は時代を超える名著だと思います。考えを整理するのにもってこいです。生成AIと対話するのにも役立ちます。
ぜひ、何かを解決するコードを一緒に書きましょう。
当社はいつでも社員を募集しています。
また、Xのフォローもお待ちしています。