pyolive-ng 1.1.1

Python API for Athena pywok (Next Generation with Athena Broker).

pyolive-ng

Python API for Athena pywok (Next Generation with Athena Broker).

시스템 요구사항

중요: pyolive-ng는 NNG (Nanomsg Next Gen) v1.11 네이티브 라이브러리가 시스템에 설치되어 있어야 합니다.

NNG v1.11 설치

Linux

# Debian/Ubuntu
sudo apt-get install libnng-dev

# 또는 소스에서 빌드 (v1.11)
wget https://github.com/nanomsg/nng/archive/v1.11.0.tar.gz
tar -xzf v1.11.0.tar.gz
cd nng-1.11.0
mkdir build && cd build
cmake ..
make
sudo make install

macOS

# Homebrew
brew install nng

# 또는 소스에서 빌드
brew install cmake
git clone https://github.com/nanomsg/nng.git
cd nng
git checkout v1.11.0
mkdir build && cd build
cmake ..
make
sudo make install

Windows

• NNG v1.11 DLL을 다운로드하여 시스템 PATH에 추가하거나
• 소스에서 빌드

Python 버전 요구사항

중요: pyolive-ng는 Python 3.12 이상이 필요합니다.

# Python 버전 확인
python3 --version  # Python 3.12.x 이상이어야 함

Python 패키지 설치

Install and update using pip:

python3 -m pip install -U pyolive-ng

참고:

• pynng는 Python 바인딩일 뿐이며, 실제로는 시스템에 설치된 NNG 네이티브 라이브러리를 사용합니다.
• Python 3.12 이상에서 최신 성능 개선 및 기능을 활용할 수 있습니다.

설정 파일

pyolive-ng는 $ATHENA_HOME/config/ 디렉토리에서 설정 파일을 읽습니다.

athena-agent.yaml

Athena Broker 연결 설정을 포함합니다:

broker:
  hosts:
    - localhost  # 또는 실제 broker 호스트 주소
  port: 2736     # Athena Broker 기본 포트
  username: guest
  password: guest

worker:
  reload-on-change: false

log:
  level: INFO
  rotate: 3
  size: 10mb

중요:

• broker/hosts는 반드시 설정되어야 합니다.
• Athena Broker 서버가 실행 중이어야 합니다.
• 연결 실패 시 에러 메시지에 설정 파일 경로와 broker URL이 표시됩니다.

A Simple Example

main.py

import asyncio
import argparse

from pyolive.agent import Athena
from registry import register_apps

async def main():
    parser = argparse.ArgumentParser(description='Start Athena pywok')
    parser.add_argument('-ns', required=True, help='namespace (e.g. dps.msm)')
    parser.add_argument('-alias', required=True, help='worker alias (e.g. worker)')
    args = parser.parse_args()

    agent = Athena(namespace=args.ns, alias=args.alias)
    register_apps(agent)

    await agent.run()

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nAthena pywok stopped by user.")

registry.py

from helloworld import HelloWorld

def register_apps(agent):
    app_map = {
        'ovm_hello_py': HelloWorld
    }

    for app_name, cls in app_map.items():
        agent.add_resource(cls, app_name)

helloworld.py

import os
from pyolive.status import AppStatus

class HelloWorld:
    def __init__(self, *, logger, channel, context):
        self.logger = logger
        self.channel = channel
        self.context = context

    async def app_main(self):
        tag = self.context.get_param('tag')
        self.logger.info("parameter tag = %s", tag)

        files = self.context.get_fileset()
        if files:
            self.logger.info("%s, filenames = %s", self.context.action_app, files)

        data = self.context.get_msgbox()
        if data:
            self.logger.info("%s, msgbox = %s", self.context.action_app, data)

        # Example of sending job status message
        await self.channel.publish_notify(self.context, text="HelloWorld completed")

        return AppStatus.OK


# Developer test mode (optional)
if __name__ == '__main__':
    import asyncio
    import sys
    import traceback
    from pyolive.develop import develop_mode

    async def run():
        try:
            # user set
            params = {'tag': 'debug'}
            infile = "."

            log, ch, ctx = await develop_mode(infile, params)
            app = HelloWorld(logger=log, channel=ch, context=ctx)

            result = await app.app_main()
            return 0 if result == AppStatus.OK else 1

        except Exception as e:
            print(f"[ERROR] test run failed: {e}", file=sys.stderr)
            traceback.print_exc()
            return 2

    sys.exit(asyncio.run(run()))