Архивы Python - Страница 4 из 12

Python: Получение состояния реле Sonoff Basic R3

09.01.2023 Павел Грибов Python

Задача: получить состояние реле и записать его состояние в БД

Решение:

Реле отдает своё состояние по ссылке:

http://{ip}:8081/zeroconf/info

1	http://{ip}:8081/zeroconf/info

Однако в зависимости от ревизии прошивки, данные могут немного отличаться. В нижеприведённом скрипте это учтено:

#!/usr/bin/python3
import config
import pymysql
import requests
from urllib3.exceptions import InsecureRequestWarning
import random

ip="192.168.88.245"             # IP реле
source=2        # источник - реле SonOff_1 (Гостинная)
place=1         # расположение реле - гостиннная

# здесь кусок кода который опрашивает текущее состояние реле
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
post_params = '{"deviceid": "","data": {}}'
response = requests.post(f"http://{ip}:8081/zeroconf/info", data=post_params, verify=False)
try:
    res = response.json()
    print(f"пришло:{res}")
    switch=True
    signal=0
    if type(res['data'])==str:
      res["data"]=json.loads(res["data"])
    if res["data"]["switch"]=="off":switch=False
    if "signalStrength" in res["data"]:signal=res["data"]["signalStrength"]

except Exception as e:
    print(f"Ошибка:{e}")
    exit(-1)

print(signal)
print(switch)

# соединяемся с БД
con=pymysql.connect(host=config.gconfig['mysql_host'],
                port=3306,
                user=config.gconfig['mysql_user'],
                password=config.gconfig['mysql_password'],
                database=config.gconfig['mysql_base'],
                cursorclass=pymysql.cursors.DictCursor
        )
# ложим данные в БД
with con.cursor() as cursor:
    sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},3,{switch},now())";
    cursor.execute(sql)
    con.commit()
    sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},4,{signal},now())";
    cursor.execute(sql)
    con.commit()

#!/usr/bin/python3

import config

import pymysql

import requests

from urllib3.exceptions import InsecureRequestWarning

import random

ip="192.168.88.245" # IP реле

source=2 # источник - реле SonOff_1 (Гостинная)

place=1 # расположение реле - гостиннная

# здесь кусок кода который опрашивает текущее состояние реле

requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

post_params = '{"deviceid": "","data": {}}'

response = requests.post(f"http://{ip}:8081/zeroconf/info", data=post_params, verify=False)

try:

res = response.json()

print(f"пришло:{res}")

switch=True

signal=0

if type(res['data'])==str:

res["data"]=json.loads(res["data"])

if res["data"]["switch"]=="off":switch=False

if "signalStrength" in res["data"]:signal=res["data"]["signalStrength"]

except Exception as e:

print(f"Ошибка:{e}")

exit(-1)

print(signal)

print(switch)

# соединяемся с БД

con=pymysql.connect(host=config.gconfig['mysql_host'],

port=3306,

user=config.gconfig['mysql_user'],

password=config.gconfig['mysql_password'],

database=config.gconfig['mysql_base'],

cursorclass=pymysql.cursors.DictCursor

)

# ложим данные в БД

with con.cursor() as cursor:

sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},3,{switch},now())";

cursor.execute(sql)

con.commit()

sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},4,{signal},now())";

cursor.execute(sql)

con.commit()

Оставить комментарий sonoff, получение состояния, реле, умный дом

Ротация логов в приложении на Python

17.10.2022 Павел Грибов Python

Задача: в приложении используется встроенный logging. Необходимо организовать ротацию логов со сжатием «архива»

Решение:

import coloredlogs,logging
from logging.handlers import TimedRotatingFileHandler
import gzip

class GZipRotator:
    def __call__(self, source, dest):
        os.rename(source, dest)
        f_in = open(dest, 'rb')
        f_out = gzip.open("%s.gz" % dest, 'wb')
        f_out.writelines(f_in)
        f_out.close()
        f_in.close()
        os.remove(dest)

rotation_logging_handler = TimedRotatingFileHandler(gconfig["logging_file"], when='D', interval=1, backupCount=5)
rotation_logging_handler.rotator=GZipRotator();
fh = logging.FileHandler('spam.log')
fh.filemode="a"

logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(message)s',datefmt='%d.%m.%Y %H:%M:%S',level=gconfig["logging_level"],handlers=[rotation_logging_handler,fh])
logging.captureWarnings(True)

import coloredlogs,logging

from logging.handlers import TimedRotatingFileHandler

import gzip

class GZipRotator:

def __call__(self, source, dest):

os.rename(source, dest)

f_in = open(dest, 'rb')

f_out = gzip.open("%s.gz" % dest, 'wb')

f_out.writelines(f_in)

f_out.close()

f_in.close()

os.remove(dest)

rotation_logging_handler = TimedRotatingFileHandler(gconfig["logging_file"], when='D', interval=1, backupCount=5)

rotation_logging_handler.rotator=GZipRotator();

fh = logging.FileHandler('spam.log')

fh.filemode="a"

logger = logging.getLogger(__name__)

logging.basicConfig(format='%(asctime)s %(message)s',datefmt='%d.%m.%Y %H:%M:%S',level=gconfig["logging_level"],handlers=[rotation_logging_handler,fh])

logging.captureWarnings(True)

Результатом работы будет создание 1 раз в день сжатого файла

Оставить комментарий compress, logging, rotation, ротация логов

Python: из объекта в JSON

12.10.2022 Павел Грибов Python

Если применять json.dumps с объектом, то можно словить ошибку сериализации объекта. Для обхода этого можно применить следующий способ:

def serialize(obj):    
    return obj.__dict__
loc=json.dumps(loc, default=serialize)

def serialize(obj):

return obj.__dict__

loc=json.dumps(loc, default=serialize)

В этом случае, если штатный сериализатор «не справляется», то ему будет передана «подсказка» о типе

Оставить комментарий json, Python3, serialize

uvicorn: запускается только один инстанс при любом количестве workers

16.09.2022 Павел Грибов Python

Может поможет страждущим как и я. Двое суток гугла и янденкса ;(

Во время отладки скрипта на FastApi возникла ситуация, что сервис должен обращаться сам к себе, ну типа небольшая рекруссия. Соответсвенно чтоб трюк сработал, нужно чтобы было запущено несколько инстансов uvicorn одновременно. Ну ноу проблем, запускаю:

uvicorn main:app --reload --port 8000 --workers 10

1	uvicorn main:app --reload --port 8000 --workers 10

И нифига. Запускается только один инстанс. И так и сяк, и гул и яндекс.

Наутро доходит попробовать убрать —reload (ну т.е. перезапускать инстанс автоматически при изменении кода — очень полезная штука во время отладки). И…..заработало! Баг?

Оставить комментарий uvicorn

FastApi: передача заголовка Authorization в header

12.09.2022 Павел Грибов Python

Не тривиальная задача, которая потребовала двух дней поиска решения.

Казалось бы, всё решается просто:

@router.get('/versions',tags=["Credentials"],responses={
        200: {
            "model": List[models.versions_info],
            "description": "Return has code",
            "headers": {"Authorization": {"description":"Token party","type":"string"}}
        }})
async def list_versions(request: Request,token: Union[str, None] = Header(alias="Authorization",default=None)): 
    print(token)
    out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]
    return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})

@router.get('/versions',tags=["Credentials"],responses={

200: {

"model": List[models.versions_info],

"description": "Return has code",

"headers": {"Authorization": {"description":"Token party","type":"string"}}

}})

async def list_versions(request: Request,token: Union[str, None] = Header(alias="Authorization",default=None)):

print(token)

out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]

return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})

Однако вместо ожидаемой передачи заголовка, curl в упор возвращает ровно ничего:

Гугл помог выяснить, что ключевое слово Authorization зарезервированно для использования в Header. На https://stackoverflow.com было подсказано и направление куда копать. А именно, использовать обертку Security:

router = APIRouter()

token_key = fastapi.security.APIKeyHeader(name="Authorization")

class Token(BaseModel):
    token: str

def get_current_token(oauth_header: str = fastapi.Security(token_key)):
    print(f"Token: {oauth_header}")
    return oauth_header


# выдаем список поддерживаемых версий
@router.get('/versions',tags=["Credentials"],responses={
        200: {
            "model": List[models.versions_info],
            "description": "Return has code",
            "headers": {"Authorization": {"description":"Токен участника для авторизации","type":"string"}}
            },
        401: {
            "description": "Unauthorized status code.",
            }
        })
async def list_versions(request: Request,token: funcs.Token = Depends(funcs.get_current_token)): # запрос на получение поддерживаемых версий ПО
    print(f"Токен:{token}")
    if token==None: return Response(status_code=401, content="Unauthorized status code.", media_type="application/text")
    out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]
    return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})

router = APIRouter()

token_key = fastapi.security.APIKeyHeader(name="Authorization")

class Token(BaseModel):

token: str

def get_current_token(oauth_header: str = fastapi.Security(token_key)):

print(f"Token: {oauth_header}")

return oauth_header

# выдаем список поддерживаемых версий

@router.get('/versions',tags=["Credentials"],responses={

200: {

"model": List[models.versions_info],

"description": "Return has code",

"headers": {"Authorization": {"description":"Токен участника для авторизации","type":"string"}}

401: {

"description": "Unauthorized status code.",

}

})

async def list_versions(request: Request,token: funcs.Token = Depends(funcs.get_current_token)): # запрос на получение поддерживаемых версий ПО

print(f"Токен:{token}")

if token==None: return Response(status_code=401, content="Unauthorized status code.", media_type="application/text")

out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]

return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})

Оставить комментарий fastapi, header

« 1 2 3 4 5 6 … 12 »

Пн	Вт	Ср	Чт	Пт	Сб	Вс
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30