LINUX.ORG.RU

Сообщения dcopm999

 

Хотелось бы узнать Ваше мнение по поводу читаемости кода

Форум — Development

Прошлый вариант

( читать дальше... )

Новый вариант

import netsnmp
import json
from pysmi.reader import HttpReader
from pysmi.writer import CallbackWriter
from pysmi.parser.smi import SmiV2Parser
from pysmi.codegen.jsondoc import JsonCodeGen
from pysmi.compiler import MibCompiler


from snmp.models import MIBList, MIBObject, Hosts, SupportedOid

import logging
logger = logging.getLogger(__name__)

class MIBBaseClass(object):
    def __init__(self):
        self.ListMIBModel = MIBList #Модель со всеми MIBами
        self.ObjectMIBModel = MIBObject #Модель со всеми MIB объектами
        self.HostModel = Hosts #Модель с параметрами подключения к хостам
        self.SupportedOidModel = SupportedOid #Модель с поддерживаемыми oid для конкретного хоста
        self.results = {}
        self.status = ""

class OidSupportedClass(MIBBaseClass):
    def get(self):
        bulk = []
        for host in self.HostModel.objects.all():
            conn = {"DestHost": host.host, "Community": host.community, "Version": host.version}
            for oiditem in self.ObjectMIBModel.objects.all():
                if netsnmp.snmpwalk(oiditem.name, **conn):
                    bulk.append(self.SupportedOidModel(host=host, oid=oiditem))
        self.results = bulk
    def save(self):
        def exists_indb(item):
            return self.SupportedOidModel.objects.filter(
                host__id=item.host.id,
                oid__id=item.oid.id
            ).exists()
        bulk = [item for item in self.results if not exists_indb(item)]
        self.SupportedOidModel.objects.bulk_create(bulk)
class MIBParserClass(MIBBaseClass):
    """
    Парсинг MIB файлов с сайта mibs.snmplabs.com/asn1/
    TODO: Вынести куданибудь self.http_sources
    """
    def __init__(self):
        super(MIBParserClass, self).__init__()
        self.http_sources = [('mibs.snmplabs.com', 80, '/asn1/@mib@')]
    def get(self):
        def _get_result(mib_name, mib_objects, cbCtx):
            self.results[mib_name] = json.loads(mib_objects)
        mib_parser = MibCompiler(SmiV2Parser(), JsonCodeGen(), CallbackWriter(_get_result))
        mib_parser.addSources(*[HttpReader(*item) for item in self.http_sources])
        self.status = mib_parser.compile(
            *{item.name for item in self.ListMIBModel.objects.filter(compiled=False)}
        )
    def save(self):
        def exists_in_miblist(mib_name):
            return self.ListMIBModel.objects.filter(name=mib_name).exists()
        def exists_in_mibobjects(item):
            return self.ObjectMIBModel.objects.filter(name=item.name).exists()
        bulk_listmib = [self.ListMIBModel(name=item) for item in self.results.keys() if not exists_in_miblist(item)]
        self.ListMIBModel.objects.bulk_create(bulk_listmib)
        bulk_mibobjects = []
        for mib_name, mib_objects in self.results.items():
            self.ListMIBModel.objects.filter(name=mib_name).update(compiled=True)
            for item in mib_objects.values():
                values = {"mib": self.ListMIBModel.objects.get(name=mib_name),
                          "typeclass": item.get("class"),
                          "maxaccess": item.get("maxaccess"),
                          "name": item.get("name"),
                          "nodetype": item.get("nodetype"),
                          "oid": item.get("oid"),
                          "status": item.get("status"),
                          "syntax": item.get("syntax"),
                         }
                if item.get("oid") and item.get("syntax"):
                    bulk_mibobjects.append(self.ObjectMIBModel(**values))
        bulk_mibobjects = [item for item in bulk_mibobjects if not exists_in_mibobjects(item)]
        self.ObjectMIBModel.objects.bulk_create(bulk_mibobjects)

 

dcopm999 ()

Спи спокойно дорогой товарищ

Форум — Web-development

Наверное у всех админов витает в воздухе мысль написать свою собственную вэб-мордочку для централизованного поднятия сервисов и заливки конфигов на сервера. Кому интересна данная тема прошу присоединяйтесь...

https://github.com/dcopm999/ssdt.git

Описание модели deploy:

Service: Перечень сервисов которые используются в вашем проекте

  • name - Название сервиса (nginx, apache2, php-fpm, memcached...)
  • image - Логотип сервиса
  • desc - Описание (реверсивный прокси, ... ) необязательное поле

Address: перечень серверов которые используются в вашем проекте

  • addr - ip адрес сервера
  • auth_type - Тип авторизации (user, ssh-key, ...)
  • username - логин необязательное поле
  • password - пароль необязательное поле

DeployGroup: разделение серверов на логические группы, с одинаковым набором сервисов

  • name - Произвольное название группы серверов
  • servicegroup - Перечень сервисов которые необходимо развернуть в группе серверов
  • addressgroup - Перечень серверов на которых будут развернуты сервисы, указанные в группе сервисов

предполагается два типа авторизации на сверверах:

  • 1) user - авторизация через пользователя имеющего права sudo
  • 2) ssh-key - авторизация рутом через ssh ключ

 , ,

dcopm999 ()

Python gstreamer (перепаковка потока в ogg и streaming в локалку)

Форум — Development

Пытаюсь повторить на Python 2.7 такую строчку gst-launch souphttpsrc location='http://........' ! flump3dec! audioconvert ! vorbisenc ! oggmux ! udpsink host='192.168.0.22' port=5000

На Python у меня это выглядит так (в чем косяк?):


import gst, gobject
gobject.threads_init()

class audio_ply:
    
    def __init__(self):
        self.mainloop=gobject.MainLoop()            
        self.pipe=gst.Pipeline("pipe")

        self.src=gst.element_factory_make("souphttpsrc", "src")
        self.src.set_property("location", "http://......")
        self.pipe.add(self.src)
        print "soup started"

        self.decode=gst.element_factory_make("flump3dec", "decode")
        self.pipe.add(self.decode)
        print "flump3dec started"

        self.convert=gst.element_factory_make("audioconvert", "convert")
        self.pipe.add(self.convert)
        print "audioconvert started"

        self.encode=gst.element_factory_make("vorbisenc", "encode")
        self.pipe.add(self.encode)
        print "vorbisenc started"

        self.mux=gst.element_factory_make("oggmux", "mux")
        self.pipe.add(self.mux)
        print "oggmux started"

        self.sink=gst.element_factory_make('udpsink', 'sink')
        self.sink.set_property("host", """127.0.0.1""")
        self.sink.set_property("port", 5000)
        self.pipe.add(self.sink)
        print "sink started"

        self.src.link(self.decode)
        self.decode.link(self.convert)
        self.convert.link(self.encode)
        self.encode.link(self.sink)

    def decode_src_created(self, element, pad):
        pad.link(self.sink.get_static_pad("sink"))

    def run(self):
        self.pipe.set_state(gst.STATE_PLAYING)
        print "State Play"
        self.mainloop.run()

    def done(self):
        self.pipe.set_state(gst.STATE_NULL)
        print "state null"

try:
    play=audio_ply()
    play.run()
finally:
    play.done()

 , ,

dcopm999 ()

RSS подписка на новые темы