Metadata-Version: 2.4
Name: is-wire-sea
Version: 3.0.0
Summary: Camada de suporte AMQP em Python para a arquitetura IS
Author-email: LabSEA <labsea.gua@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/LabSEA/is-wire-sea
Project-URL: Repository, https://github.com/LabSEA/is-wire-sea
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3 :: Only
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: amqp<6,>=5.3.1
Requires-Dist: colorlog<7,>=6.12
Requires-Dist: opentelemetry-api<2,>=1.44
Requires-Dist: opentelemetry-propagator-b3<2,>=1.44
Requires-Dist: prometheus-client<1,>=0.26
Requires-Dist: protobuf<8,>=5.29
Provides-Extra: tracing
Requires-Dist: opentelemetry-exporter-otlp-proto-grpc<2,>=1.44; extra == "tracing"
Requires-Dist: opentelemetry-sdk<2,>=1.44; extra == "tracing"
Provides-Extra: zipkin
Requires-Dist: opentelemetry-exporter-zipkin-json<2,>=1.44; extra == "zipkin"
Requires-Dist: opentelemetry-sdk<2,>=1.44; extra == "zipkin"
Provides-Extra: codegen
Requires-Dist: grpcio-tools==1.69.0; extra == "codegen"
Provides-Extra: dev
Requires-Dist: build<2,>=1.5; extra == "dev"
Requires-Dist: pytest<10,>=8.3; extra == "dev"
Requires-Dist: pytest-cov<8,>=7; extra == "dev"
Requires-Dist: ruff<1,>=0.12; extra == "dev"
Requires-Dist: tox<5,>=4.25; extra == "dev"
Requires-Dist: twine<8,>=6; extra == "dev"
Dynamic: license-file

# is-wire-sea

Middleware AMQP em Python para a arquitetura IS. A distribuição no PyPI se chama `is-wire-sea`; o pacote Python estável continua sendo `is_wire`.

Na versão 3.0, remover bindings de filas nomeadas exige
`unsubscribe(topic, shared_queue=True)`. Consulte [MIGRATION.md](MIGRATION.md)
antes de atualizar aplicações da série 2.x. O formato wire permanece compatível.

## Instalação

```shell
python -m pip install is-wire-sea
```

Há suporte para Python 3.10 a 3.14. Instale um exportador OTLP com:

```shell
python -m pip install 'is-wire-sea[tracing]'
```

## Broker

Implantações de produção devem fixar o RabbitMQ pela versão de patch e pelo digest da imagem.
A base de desenvolvimento é o RabbitMQ 4.3.1:

```shell
docker run --rm -p 5672:5672 \
  rabbitmq:4.3.1@sha256:6a46d2aef889d2a8cc28ac91b4a1ca0116a4a151de10cada219ee7685dc01c5b
```

RabbitMQ 3.7.6 é apenas um alvo de migração. Mova as implantações existentes da versão 3.7.6 para um cluster 4.3 novo usando uma migração blue-green; não reutilize diretamente o diretório de dados antigo.

## Pub/sub

A API existente de entrega no máximo uma vez é preservada:

```python
from is_wire.core import Channel, Message, Subscription

with Channel("amqp://guest:guest@localhost:5672") as channel:
    subscription = Subscription(channel)
    subscription.subscribe("Camera.Config")

    channel.publish(Message(content=b"hello"), topic="Camera.Config")
    received = channel.consume(timeout=1.0)
```

Assinaturas anônimas são exclusivas e temporárias. Assinaturas nomeadas são duráveis, compartilhadas entre réplicas e expiram após cinco minutos sem uso.

## Fluxos de imagens em tempo real

Use `StreamSubscription` quando quadros antigos puderem ser descartados. Réplicas com o mesmo grupo compartilham uma fila, então cada quadro é processado por uma réplica em vez de ser copiado para todas elas.

```python
from is_msgs.image_pb2 import Image
from is_wire.core import Channel, Message, StreamSubscription


def process(message):
    image = message.unpack(Image)
    # image.data já contém bytes JPEG, WebP ou PNG.


with Channel("amqp://guest:guest@localhost:5672") as channel:
    stream = StreamSubscription(channel, group="detector")
    stream.subscribe("Camera.*.Frame")
    stream.run(process)
```

Publique um quadro com semântica de fluxo:

```python
image = Image(data=encoded_jpeg_bytes)
channel.publish_stream(Message(content=image), topic="Camera.0.Frame")
```

O preset de fluxo usa confirmações manuais, prefetch 1, fila com tamanho um, descarte do item mais antigo quando há overflow, TTL de dois segundos e mensagens não persistentes. Uma exceção no callback rejeita o quadro sem recolocá-lo na fila. Use outro grupo somente quando outra etapa de processamento realmente precisar de sua própria cópia.

`Image.data` deve conter bytes binários codificados da imagem. Não envie RGB/BGR bruto, a menos que o orçamento de rede permita explicitamente; não codifique o corpo em base64 e não aplique compressão gzip/zstd genérica a um payload JPEG, WebP ou PNG já comprimido.

Use um `Channel` separado para o tráfego de imagens e para o tráfego de RPC/controle, evitando que quadros grandes atrasem mensagens pequenas de controle.

### Lifecycle e callbacks

`Subscription.close()` cancela o consumer e remove o registro no `Channel`.
Também pode ser usada como context manager. Ela mantém filas nomeadas e seus
bindings compartilhados; entregas manuais ainda no buffer local são recolocadas
na fila. Em uma subscription padrão, mensagens já retornadas por `consume()`
continuam sob responsabilidade do chamador. Fechar uma subscription de stream
fecha seu canal AMQP dedicado e pode reentregar frames em flight.

`unsubscribe(topic)` continua disponível para filas anônimas. Em filas nomeadas,
o binding é compartilhado por todas as réplicas: a remoção exige
`unsubscribe(topic, shared_queue=True)` e coordenação entre os processos,
inclusive durante reconexões. As assinaturas da mesma conexão são atualizadas
juntas; outras conexões não recebem essa atualização de configuração. Para
encerrar apenas o consumidor local, use `close()`.

`StreamSubscription.stop()` sinaliza o loop; não cancela o consumer. O callback
executa em uma thread daemon e não pode ser morto com segurança pelo Python.
Depois do período de graça, `run()` rejeita o frame e retorna, mas o callback
pode continuar executando. Uma nova chamada a `run()` é recusada enquanto esse
callback anterior estiver vivo. Nunca suponha que rejeitar o frame cancela os
efeitos externos do callback. Um grupo admite um frame pendente **mais** um em
flight por consumer; TTL não interrompe trabalho que já começou.

`Channel` tem um único dono de I/O; múltiplos consumidores não devem executar
loops de leitura simultâneos na mesma conexão. Durante callbacks longos, pode
ser útil ativar `stream.run(process, poll_during_callback=True)` para drenar a
conexão e atender heartbeats. Essa opção exige que o callback deixe ack/reject
para o loop e use um `Channel` separado, criado na thread do callback, para
publicar resultados. O default preserva o agendamento existente. `stop()` pode
ser chamado de outra thread; `close()` deve ser chamado pelo dono do I/O.

## Protobuf

```python
from google.protobuf.struct_pb2 import Struct
from is_wire.core import ContentType, Message

value = Struct()
value["camera"] = "front"

binary = Message(content=value)
assert binary.content_type == ContentType.PROTOBUF
assert binary.unpack(Struct) == value

json_message = Message(content_type=ContentType.JSON)
json_message.pack(value)
```

Há suporte para Protobuf 5 a 7. Os corpos binários existentes e as convenções de propriedades
AMQP continuam compatíveis com `is-wire` 1.2.1.

Envelopes AMQP incompatíveis com o formato IS, como IDs de correlação não
hexadecimais ou cabeçalhos `rpc-status` inválidos, são ignorados sem interromper
o consumo. Entregas com ack manual são rejeitadas sem requeue; uma dead-letter
exchange configurada no broker pode recebê-las. O timeout original continua
valendo e o aviso de descarte não inclui o corpo nem os cabeçalhos. Isso não
altera o tratamento de um payload RPC inválido dentro de um envelope válido.

## RPC

```python
from google.protobuf.struct_pb2 import Struct
from is_wire.core import Channel
from is_wire.rpc import ServiceProvider


def echo(request, context):
    return request


channel = Channel("amqp://guest:guest@localhost:5672")
provider = ServiceProvider(channel)
provider.delegate("Echo", echo, Struct, Struct)
provider.run()
```

As filas de serviços RPC usam confirmações manuais e prefetch 16. A requisição é confirmada depois que `basic_publish()` retorna. Publicações não
usam publisher confirms nem `mandatory`: isso não comprova aceite ou roteamento
da resposta pelo broker. Falhas entre processamento e ack podem reentregar a
requisição, e uma resposta pode se perder mesmo após o ack. Os handlers devem
ser idempotentes; a semântica atual não é exactly-once e não garante entrega
persistente da resposta após reinício do broker.

## OpenTelemetry

`Tracer`, `TracingInterceptor`, `Message.inject_tracing` e `Message.extract_tracing` usam OpenTelemetry e propagação B3 com múltiplos cabeçalhos. IDs de trace legados de 64 bits e IDs de 128 bits são aceitos.
Spans aninhados compartilham o trace ativo e restauram o contexto anterior
ao terminar, inclusive sem instalar o SDK opcional.

```python
from is_wire.core import Message, Tracer

tracer = Tracer()
with tracer.span("publish") as span:
    message = Message(content=b"payload")
    message.inject_tracing(span)
```

### Zipkin

Instale o exportador nativo OpenTelemetry Zipkin JSON v2:

```shell
python -m pip install 'is-wire-sea[zipkin]'
```

Crie um provider compartilhado por processo. Ele agrupa as exportações em segundo plano, aplica amostragem baseada no contexto pai e identifica cada instância de câmera separadamente no Zipkin:

```python
from is_wire.core import Message, ZipkinTracing
from is_wire.rpc import ServiceProvider, TracingInterceptor

tracing = ZipkinTracing(
    service_name="camera-gateway",
    endpoint="http://zipkin:9411/api/v2/spans",
    service_instance_id="camera-5",
    sample_ratio=0.1,
    resource_attributes={"camera.driver": "hikvision"},
)
frame_tracer = tracing.tracer()

provider = ServiceProvider(rpc_channel)
provider.add_interceptor(TracingInterceptor(tracing=tracing))

with frame_tracer.span("camera.frame") as span:
    span.set_attribute("camera.id", "5")
    message = Message(content=image)
    message.inject_tracing(span)
    stream_channel.publish_stream(message, topic="CameraGateway.5.Frame")

# Libere os spans enfileirados durante o encerramento normal do processo.
tracing.shutdown()
```

Não associe corpos de imagem, credenciais, URLs de câmeras ou identificadores sem limite aos spans. Use uma proporção de amostragem menor que um para vídeo contínuo e defina `1.0` apenas em sessões curtas de diagnóstico. Consumidores continuam o trace do produtor com `message.extract_tracing()`.

## Commtrace opcional

`CommTraceClient` acrescenta relatórios de comunicação ao tracing existente.
Ele envia um JSON por UDP no formato usado pelo
[is-person-detector](https://github.com/JoabFelippx/is-person-detector/blob/6d63410f8c7f7ff744061fc29bd79f07b5d404ef/is_person_detector/utils.py).
Zipkin/OTLP continuam recebendo os spans de processamento; commtrace recebe os
horários e os IDs B3 para correlacionar a comunicação entre serviços.
O cliente usa a biblioteca padrão, sem extras ou dependências adicionais.

Para registrar automaticamente mensagens recebidas, configure explicitamente
`commtrace=` no `Channel`. Isso vale para `Subscription`, RPC e
`StreamSubscription`:

```python
from is_wire.core import Channel, CommTraceClient, StreamSubscription

with CommTraceClient("127.0.0.1", 9999, span_name="frame") as commtrace:
    with Channel("amqp://guest:guest@localhost:5672", commtrace=commtrace) as channel:
        stream = StreamSubscription(channel, group="detector")
        stream.subscribe("CameraGateway.*.Frame")
        message = stream.consume(timeout=1.0)
        # O relatório é enviado antes de consume() retornar.
        # Processe a imagem e continue o tracing normalmente.
        message.ack()
```

O endereço e a porta são do seu coletor UDP; `9999` é apenas um exemplo.
O produtor precisa incluir B3, por exemplo com `message.inject_tracing(span)`.
Sem `commtrace=`, não há envio nem abertura de socket de commtrace.
O cliente pode ser compartilhado entre canais e deve ser fechado pelo chamador;
fechar um `Channel` não fecha o cliente.

Para envio manual, depois de um consumo em canal sem `commtrace=`, use:

```python
from is_wire.core import CommTraceClient, now

received_at = now()  # Capture imediatamente após receber a mensagem.
with CommTraceClient("127.0.0.1", 9999) as commtrace:
    sent = commtrace.send(message, received_at=received_at, span_name="frame")
```

O datagrama contém exatamente `timestamp_send`, `timestamp_rcvd`,
`x-b3-flags`, `x-b3-parentspanid`, `x-b3-sampled`, `x-b3-spanid`,
`x-b3-traceid` e `spanname`. Os timestamps são strings com microssegundos Unix;
`timestamp_send` deriva de `message.created_at`, e `timestamp_rcvd` é capturado
na entrega à aplicação. A diferença inclui filas e espera antes do consumo;
ela não isola o tempo de trânsito na rede. Sincronize os relógios dos serviços
para comparar horários de máquinas diferentes.

Contextos B3 de 64/128 bits e a decisão `sampled=false` são preservados.
Mensagens sem contexto B3 válido não geram relatórios. O relatório não inclui
o corpo nem os demais metadados da mensagem, e não altera sua serialização AMQP.

O socket é reutilizado e o envio é não bloqueante por padrão (`timeout=0.0`).
Falhas UDP/DNS retornam `False` e não interrompem o consumo AMQP; não há retries.
`True` significa que o socket local aceitou o datagrama, sem confirmação do
coletor. O primeiro envio resolve o hostname, e DNS não tem deadline rígido;
use um endereço IP quando esse custo importar. Um `timeout` positivo limita a
espera do socket, mas pode acrescentar esse tempo ao consumo automático.
UDP pode perder ou reordenar relatórios. Ativar commtrace acrescenta trabalho
ao processo; não acelera o AMQP nem substitui o tracing de processamento.

## Métricas

As métricas do Prometheus são registradas no registry padrão. `MetricsInterceptor.start_server()` pode expô-las por HTTP. O perfil 2.0 exporta:

- `is_wire_messages_published_total` e `is_wire_published_bytes_total`;
- `is_wire_messages_received_total` e `is_wire_received_bytes_total`;
- `is_wire_stream_processing_seconds`, `is_wire_stream_frame_age_seconds` e
  `is_wire_stream_callback_errors_total`;
- `is_wire_reconnections_total`;
- `is_wire_rpc_duration_seconds` e `is_wire_rpc_requests_total`, particionados por status.

O conteúdo das imagens e os IDs de correlação nunca são usados como labels de
métricas. Topics e grupos, porém, são labels: respostas em filas anônimas geram
séries com IDs de fila que permanecem no registry. Mantenha grupos estáveis e
planeje agrupamento de topics antes de usar essas métricas em tráfego com alta
rotatividade de clientes. Os contadores de publicação medem chamadas que
retornaram, não confirmações recebidas do broker.

## TLS e ciclo de vida da conexão

`amqps://` habilita TLS com verificação de certificado e hostname. Opções SSL adicionais do py-amqp podem ser passadas em `ssl_options`. Consumidores se reconectam com backoff exponencial e jitter; publicações nunca
são repetidas automaticamente, pois a entrega pode já ter ocorrido. O deadline
de `consume(timeout=...)` inclui retries e espera de reconexão, e o backoff
continua entre chamadas com timeout. A conexão inicial ainda propaga falhas.
Os timeouts do py-amqp são por operação: DNS e múltiplos frames do handshake
podem ultrapassar o orçamento, portanto não são deadlines rígidos de tempo real.
`close(timeout=5.0)` limita a espera de I/O do encerramento e sempre coleta o
transporte local. Erros permanentes de autenticação/certificado são propagados.

```python
channel = Channel(
    "amqps://user:password@rabbitmq.example:5671/vhost",
    heartbeat=30,
    ssl_options={"ca_certs": "/etc/ssl/certs/cluster-ca.pem"},
)
```

## Desenvolvimento

```shell
python -m pip install -e '.[dev,tracing]'
pytest  # requer RabbitMQ; use pytest -m "not integration" sem broker
ruff check src tests
python -m build
twine check --strict dist/*
```

Regenere o schema wire interno de forma reproduzível com o compilador Protobuf 5.29 fixado:

```shell
python -m pip install -e '.[codegen]'
python scripts/generate_wire.py
```

Execute o perfil de publicador a 30 FPS / consumidor a 10 FPS contra um broker de teste com:

```shell
python scripts/validate_stream_profile.py --payload-size 1048576 --frames 300
```

Execute o mesmo comando em implantações Kubernetes separadas que compartilhem um `group` para comparar as métricas de entrada/saída do RabbitMQ conforme réplicas são adicionadas.

Consulte [MIGRATION.md](MIGRATION.md), [COMPATIBILITY.md](COMPATIBILITY.md) e
[CHANGELOG.md](CHANGELOG.md) antes de atualizar uma implantação existente.

## Agradecimentos
A equipe do LabVISIO para a base do repositório `is-wire-py`. Para a modernização, a revisão de compatibilidade, os testes e o empacotamento no PyPI da série 2.0 foram realizados com assistência do OpenAI Codex.
