mwings.parsers.app_io Module
Parsers for packets sent from the remote control app and data structures to store their contents.
PacketParser
Static class for a parser that interprets packets representing data sent from the remote control app.
Inherits from:
common.PacketParserBase
is_valid()
is_valid(bare_packet)
Checks whether the given raw packet is valid.
Arguments
| Name | Type | Description |
|---|---|---|
bare_packet | common.BarePacket | Raw packet data |
Returns
| Type | Value | Description |
|---|---|---|
bool | True | Valid, parsable |
False | Invalid, not parsable |
parse()
parse(bare_packet)
Parses the given raw packet.
Arguments
| Name | Type | Description |
|---|---|---|
bare_packet | common.BarePacket | Raw packet data |
Returns
| Type | Value | Description |
|---|---|---|
optional ParsedPacket | Data of type ParsedPacket | Parsed data |
None | Parsing failed |
ParsedPacket
Data structure to store the result of interpreting packets representing data sent from the remote control app.
Inherits from:
common.ParsedPacketBase
ParsedPacket()
ParsedPacket(*, time_parsed=None, packet_type=PacketType.BARE, sequence_number=None, source_serial_id=0, source_logical_id=0, lqi=None, supply_voltage=None, relay_count=0, di_state=<mwings.common.CrossSectional object>, di_valid=<mwings.common.CrossSectional object>, di_interrupt=<mwings.common.CrossSectional object>)
Arguments
Properties inherited from the base class are also listed.
| Name | Type | Description |
|---|---|---|
time_parsed | optional AwareDatetime | Parsing timestamp |
packet_type | common.PacketType | Packet type |
sequence_number | optional common.UInt16 | Sequence number |
source_serial_id | common.UInt32 | Source serial ID |
source_logical_id | common.UInt8 | Source logical device ID |
lqi | optional common.UInt8 | Radio communication quality |
supply_voltage | optional common.UInt16 | Always None because ADC is not used |
relay_count | common.UInt8 | Number of relays |
di_changed | common.CrossSectional[bool](12) | Digital interface change status |
di_state | common.CrossSectional[bool](12) | Digital interface state |
di_interrupt | common.CrossSectional[bool](12) | Digital interface interrupt status |
You can pass a dictionary unpacked with
**.
In addition to the constructor arguments, the following @computed_field are available.
| Name | Type | Description |
|---|---|---|
mwings_implementation | str | MWings implementation ("python" only) |
mwings_version | str | MWings version (PEP440 format) |
hostname | str | Host machine name |
system_type | str | System type (e.g., "Linux") |
Applicable field_validator
source_logical_id: Automatically validates the value rangecheck_source_logical_id(lid)
time_parsed: Automatically validates that timezone is attacheddatetime_must_be_clear(dt)
Applicable field_serializer
packet_type: Serializes the enum name as is for JSON, etc.serialize_packet_type(packet_type)
source_serial_id: Serializes as a hexadecimal string for JSON, etc.serialize_source_serial_id(source_serial_id)
time_parsed: Serializes as ISO 8601 string for JSON, etc.serialize_time_parsed(dt)
to_df()
to_df(include=None, exclude=None, verbose=True)
Converts the parsed data to a pandas DataFrame.
Arguments
| Name | Type | Description |
|---|---|---|
include | optional set[str] | Set of columns (members) to include in the DataFrame |
exclude | optional set[str] | Set of columns (members) to exclude from the DataFrame |
verbose | bool | Include system information (only when other args are None) |
Returns
| Type | Value | Description |
|---|---|---|
pd.DataFrame | - | Converted DataFrame |
Exceptions
EnvironmentErrorif pandas is not installed
to_dict()
to_dict(include=None, exclude=None, verbose=True, spread=False, sort_keys=False)
Converts the parsed data to a dictionary.
Arguments
| Name | Type | Description |
|---|---|---|
include | optional set[str] | Set of keys (members) to include in the dictionary |
exclude | optional set[str] | Set of keys (members) to exclude from the dictionary |
verbose | bool | Include system information (only when other args are None) |
spread | bool | Whether to split non-time-series list-like data |
sort_keys | bool | Whether to sort keys |
spread is set, data of type common.CrossSectional[T] is split.Returns
| Type | Value | Description |
|---|---|---|
dict[str, Any] | - | Converted dictionary |
to_json()
to_json(include=None, exclude=None, verbose=True, spread=False, indent=2, sort_keys=False)
Converts the parsed data to JSON format.
Arguments
| Name | Type | Description |
|---|---|---|
include | optional set[str] | Set of keys (members) to include in the JSON |
exclude | optional set[str] | Set of keys (members) to exclude from the JSON |
verbose | bool | Include system information (only when other args are None) |
spread | bool | Whether to split non-time-series list-like data |
indent | optional int | Number of spaces for indentation (None for no newline) |
sort_keys | bool | Whether to sort keys |
spread is set, data of type common.CrossSectional[T] is split.Returns
| Type | Value | Description |
|---|---|---|
str | - | Converted JSON string |
model_*()
pydantic.BaseModel.