Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions examples/contrib/explain.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,14 @@

Created on 7/19/2023 to support Python 3.8 to 3.11 on macOS, Ubuntu, or Windows.
"""
from __future__ import annotations

import contextlib
import os
import shutil
import tempfile
from dataclasses import dataclass
from html.parser import HTMLParser
from typing import List, Optional, Tuple, Union
from urllib import request
from urllib.error import HTTPError

Expand All @@ -27,10 +27,10 @@ class ParsedModbusResult: # pylint: disable=too-many-instance-attributes
unit_id: int
func_code: int
is_receive: bool
zero_index_reg: Optional[int] = None
quantity: Optional[int] = None
byte_count: Optional[int] = None
registers: Optional[List[int]] = None
zero_index_reg: int | None = None
quantity: int | None = None
byte_count: int | None = None
registers: list[int] | None = None

def summarize(self) -> dict:
"""Get a summary representation for readability."""
Expand All @@ -46,7 +46,7 @@ def explain_with_rapid_scada(
packet: str,
is_modbus_tcp: bool = True,
is_receive: bool = False,
timeout: Union[float, Tuple[float, float], None] = 15.0,
timeout: float | tuple[float, float] | None = 15.0,
) -> ParsedModbusResult:
"""
Explain a Modbus packet using https://rapidscada.net/modbus/.
Expand All @@ -69,7 +69,7 @@ def __init__(self, *, convert_charrefs=True):
self._data = []

@property
def data(self) -> List[str]:
def data(self) -> list[str]:
return self._data

def handle_data(self, data: str) -> None:
Expand Down Expand Up @@ -101,7 +101,7 @@ def handle_data(self, data: str) -> None:
parser.feed(response_data)

# pylint: disable-next=dangerous-default-value
def get_next_field(prior_field: str, data: List[str] = parser.data) -> str:
def get_next_field(prior_field: str, data: list[str] = parser.data) -> str:
return data[data.index(prior_field) + 1]

def parse_next_field(prior_field: str, split_index: int = 0) -> int:
Expand All @@ -114,7 +114,7 @@ def parse_next_field(prior_field: str, split_index: int = 0) -> int:
"func_code": parse_next_field("Function code"),
"is_receive": is_receive,
}
is_receive_fn_code: Tuple[bool, int] = is_receive, base_result_data["func_code"]
is_receive_fn_code: tuple[bool, int] = is_receive, base_result_data["func_code"]
if is_receive_fn_code in [(False, 0x03), (True, 0x10)]:
return ParsedModbusResult(
**base_result_data,
Expand All @@ -136,7 +136,7 @@ def parse_next_field(prior_field: str, split_index: int = 0) -> int:
)


def annotate_pymodbus_logs(file: Union[str, os.PathLike]) -> None:
def annotate_pymodbus_logs(file: str | os.PathLike) -> None:
"""Annotate a pymodbus log file in-place with explanations."""
with open(file, encoding="utf-8") as in_file, tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", delete=False
Expand Down
5 changes: 3 additions & 2 deletions pymodbus/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
maintained in the server context and the various methods
should be inserted in the correct locations.
"""
from __future__ import annotations


__all__ = [
"ModbusPlusStatistics",
Expand All @@ -15,7 +17,6 @@

# pylint: disable=missing-type-doc
from collections import OrderedDict
from typing import List

from pymodbus.constants import INTERNAL_ERROR, DeviceInformation
from pymodbus.events import ModbusEvent
Expand Down Expand Up @@ -450,7 +451,7 @@ class ModbusControlBlock:
__counters = ModbusCountersHandler()
__identity = ModbusDeviceIdentification()
__plus = ModbusPlusStatistics()
__events: List[ModbusEvent] = []
__events: list[ModbusEvent] = []

# -------------------------------------------------------------------------#
# Magic
Expand Down
8 changes: 5 additions & 3 deletions pymodbus/framer/base.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
"""Framer start."""
# pylint: disable=missing-type-doc
from typing import Any, Dict, Union
from __future__ import annotations

from typing import Any

from pymodbus.factory import ClientDecoder, ServerDecoder
from pymodbus.logging import Log
Expand All @@ -24,7 +26,7 @@ class ModbusFramer:

def __init__(
self,
decoder: Union[ClientDecoder, ServerDecoder],
decoder: ClientDecoder | ServerDecoder,
client=None,
) -> None:
"""Initialize a new instance of the framer.
Expand All @@ -33,7 +35,7 @@ def __init__(
"""
self.decoder = decoder
self.client = client
self._header: Dict[str, Any] = {
self._header: dict[str, Any] = {
"lrc": "0000",
"len": 0,
"uid": 0x00,
Expand Down
9 changes: 5 additions & 4 deletions pymodbus/payload.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
A collection of utilities for building and decoding
modbus messages payloads.
"""
from __future__ import annotations


__all__ = [
"BinaryPayloadBuilder",
Expand All @@ -11,7 +13,6 @@

# pylint: disable=missing-type-doc
from struct import pack, unpack
from typing import List

from pymodbus.constants import Endian
from pymodbus.exceptions import ParameterException
Expand Down Expand Up @@ -111,7 +112,7 @@ def to_registers(self):
Log.debug("{}", payload)
return payload

def to_coils(self) -> List[bool]:
def to_coils(self) -> list[bool]:
"""Convert the payload buffer into a coil layout that can be used as a context block.

:returns: The coil layout to use as a block
Expand All @@ -120,7 +121,7 @@ def to_coils(self) -> List[bool]:
coils = [bool(int(bit)) for reg in payload for bit in format(reg, "016b")]
return coils

def build(self) -> List[bytes]:
def build(self) -> list[bytes]:
"""Return the payload buffer as a list.

This list is two bytes per element and can
Expand All @@ -133,7 +134,7 @@ def build(self) -> List[bytes]:
buffer += b"\x00" * (length % 2)
return [buffer[i : i + 2] for i in range(0, length, 2)]

def add_bits(self, values: List[bool]) -> None:
def add_bits(self, values: list[bool]) -> None:
"""Add a collection of bits to be encoded.

If these are less than a multiple of eight,
Expand Down
2 changes: 1 addition & 1 deletion pymodbus/repl/client/helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ class Result:
"""Represent result command."""

function_code: int | None = None
data: dict[int, Any] | Any = None
data: dict[int, Any] | None = None

def __init__(self, result):
"""Initialize.
Expand Down
13 changes: 7 additions & 6 deletions pymodbus/repl/server/main.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
"""Repl server main."""
from __future__ import annotations

import asyncio
import contextlib
import json
import logging
import sys
from enum import Enum
from pathlib import Path
from typing import List

import typer

Expand Down Expand Up @@ -49,7 +50,7 @@ class ModbusFramerTypes(str, Enum):
binary = "binary" # pylint: disable=invalid-name


def _completer(incomplete: str, valid_values: List[str]) -> List[str]:
def _completer(incomplete: str, valid_values: list[str]) -> list[str]:
"""Complete value."""
completion = []
for name in valid_values:
Expand All @@ -58,19 +59,19 @@ def _completer(incomplete: str, valid_values: List[str]) -> List[str]:
return completion


def framers(incomplete: str) -> List[str]:
def framers(incomplete: str) -> list[str]:
"""Return an autocompleted list of supported clouds."""
_framers = ["socket", "rtu", "tls", "ascii", "binary"]
return _completer(incomplete, _framers)


def servers(incomplete: str) -> List[str]:
def servers(incomplete: str) -> list[str]:
"""Return an autocompleted list of supported clouds."""
_servers = ["tcp", "serial", "tls", "udp"]
return _completer(incomplete, _servers)


def process_extra_args(extra_args: List[str], modbus_config: dict) -> dict:
def process_extra_args(extra_args: list[str], modbus_config: dict) -> dict:
"""Process extra args passed to server."""
options_stripped = [x.strip().replace("--", "") for x in extra_args[::2]]
extra_args_dict = dict(list(zip(options_stripped, extra_args[1::2])))
Expand Down Expand Up @@ -138,7 +139,7 @@ def run(
help="Modbus framer to use",
),
modbus_port: str = typer.Option("5020", "--modbus-port", "-p", help="Modbus port"),
modbus_slave_id: List[int] = typer.Option(
modbus_slave_id: list[int] = typer.Option(
[1], "--slave-id", "-u", help="Supported Modbus slave id's"
),
modbus_config_path: Path = typer.Option(
Expand Down
7 changes: 4 additions & 3 deletions pymodbus/utilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
A collection of utilities for packing data, unpacking
data computing checksums, and decode checksums.
"""
from __future__ import annotations


__all__ = [
"pack_bitstring",
Expand All @@ -17,7 +19,6 @@

# pylint: disable=missing-type-doc
import struct
from typing import List


class ModbusTransactionState: # pylint: disable=too-few-public-methods
Expand Down Expand Up @@ -101,7 +102,7 @@ def dict_property(store, index):
# --------------------------------------------------------------------------- #
# Bit packing functions
# --------------------------------------------------------------------------- #
def pack_bitstring(bits: List[bool]) -> bytes:
def pack_bitstring(bits: list[bool]) -> bytes:
"""Create a bytestring out of a list of bits.

:param bits: A list of bits
Expand All @@ -128,7 +129,7 @@ def pack_bitstring(bits: List[bool]) -> bytes:
return ret


def unpack_bitstring(data: bytes) -> List[bool]:
def unpack_bitstring(data: bytes) -> list[bool]:
"""Create bit list out of a bytestring.

:param data: The modbus data packet to decode
Expand Down