import serial
import time
import struct
from dataclasses import dataclass
from typing import Optional, Tuple, Dict, List, Union
from enum import Enum, IntEnum
class SonoflowCommands(IntEnum):
"""SONOFLOW command codes"""
CMD_GET_VALUES = 0x3F
CMD_PING = 0x29
CMD_RESTART = 0x16
CMD_RESET_VOLUME = 0x39
CMD_ZERO = 0x33
CMD_IDENT = 0x25
CMD_BOOT_START = 0x22
CMD_BOOT_IDENT = 0x15
CMD_READ_SRAM = 0x17
CMD_WRITE_SRAM = 0x12
CMD_READ_CODE = 0x11
CMD_WRITE_CODE = 0x13
CMD_ERASE_CODE = 0x14
@dataclass
class DeviceIdentification:
"""Device identification information"""
device_address: int
device_type: int
type_spec: int
sub_type: int
model: int
tubing: int
hardware_version: int
serial_number: int
software_version: int
microcontroller_type: int
boot_loader_type: int
tdc_type: int
@dataclass
class MeasurementResult:
"""Measurement data from sensor"""
flow_ml_min: float
temperature_c: float
volume_ul: float
state: int
error_code: int
bubble_size: float # in mm
bubble_detected: bool
class MeasurementError(IntEnum):
"""Error codes during flow measurement"""
CHANNELS_TOO_DIFFERENT = 1 << 0
FLOW_OUT_OF_RANGE = 1 << 1
AVERAGING_OVERFLOW = 1 << 2
NO_TDC_INTERRUPT = 1 << 5
FLOW_RATE_TOO_LOW = 1 << 6
PERMANENT_ERROR = 1 << 7
class SensorState(IntEnum):
"""Sensor state bits"""
BUBBLE_ALARM = 1 << 0
WARNING = 1 << 1
RESET_OCCURRED = 1 << 3
BUBBLE_ERROR = 1 << 4
FLOW_ERROR = 1 << 5
TEMP_ERROR = 1 << 6
DEVICE_ERROR = 1 << 7
class SonoflowSensor:
def __init__(self, port: str, address: int = 1):
"""Initialize SONOFLOW sensor communication.
Args:
port: Serial port (e.g. 'COM1')
address: RS485 address (1-12)
"""
if not 1 <= address <= 12:
raise ValueError("Address must be between 1 and 12")
[Link] = address
[Link] = [Link](
port=port,
baudrate=115200,
bytesize=[Link],
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1
def _calculate_crc(self, data: bytes) -> int:
"""Calculate CRC for command frames using polynomial 0xD4."""
b = data[0]
for byte in data[1:]:
a = b ^ byte
if (a & 0x80) != 0:
b = (a << 1) ^ 0xD4
else:
b = (a << 1)
return b & 0x3F
def _build_command(self, command: int, data: bytes = b'') -> bytes:
"""Build command frame with proper structure and CRC."""
start_byte = 0xF0 + [Link]
length = len(data) + 5
frame = bytes([start_byte]) + length.to_bytes(2, 'big') + bytes([command]) + data
crc = self._calculate_crc(frame)
return frame + bytes([crc])
def _read_response(self) -> Optional[bytes]:
"""Read and validate response from sensor."""
# Read until start byte found
while True:
start_byte = [Link](1)
if not start_byte:
return None
if start_byte[0] == (0xF0 + [Link]):
break
length_bytes = [Link](2)
if len(length_bytes) != 2:
return None
length = int.from_bytes(length_bytes, 'big')
frame = [Link](length - 3)
if len(frame) != length - 3:
return None
full_frame = start_byte + length_bytes + frame[:-1]
calc_crc = self._calculate_crc(full_frame)
if calc_crc != frame[-1]:
return None
return start_byte + length_bytes + frame
def _check_ack(self) -> bool:
"""Check for acknowledgment byte."""
ack = [Link](1)
return len(ack) == 1 and ack[0] == (0x10 + [Link])
def ping(self) -> bool:
"""Test communication with sensor."""
command = self._build_command(SonoflowCommands.CMD_PING)
[Link](command)
return self._check_ack()
def restart(self) -> bool:
"""Restart the sensor."""
command = self._build_command(SonoflowCommands.CMD_RESTART)
[Link](command)
return self._check_ack()
def reset_volume(self) -> bool:
"""Reset the volume counter to 0."""
command = self._build_command(SonoflowCommands.CMD_RESET_VOLUME)
[Link](command)
return self._check_ack()
def zero_adjust(self, mode: int = 1) -> Tuple[bool, float]:
"""Perform zero adjustment for flow measurement.
Args:
mode: 0=read, 1=auto, 2=absolute, 3=relative
Returns:
Tuple of (success, current_zero_value)
"""
data = bytes([mode & 0x03]) # Zero mode in first 2 bits
command = self._build_command(SonoflowCommands.CMD_ZERO, data)
[Link](command)
response = self._read_response()
if not response:
return False, 0.0
# Parse zero adjust value (float)
value_bytes = response[5:9]
value = [Link]('<f', value_bytes)[0]
return True, value
def get_identification(self) -> Optional[DeviceIdentification]:
"""Read device identification information."""
command = self._build_command(SonoflowCommands.CMD_IDENT)
[Link](command)
response = self._read_response()
if not response:
return None
return DeviceIdentification(
device_address=response[4],
device_type=response[5],
type_spec=response[6],
sub_type=response[7],
model=response[8],
tubing=response[9],
hardware_version=int.from_bytes(response[10:12], 'little'),
serial_number=int.from_bytes(response[12:16], 'little'),
software_version=int.from_bytes(response[16:18], 'little'),
microcontroller_type=int.from_bytes(response[46:48], 'little'),
boot_loader_type=response[48],
tdc_type=response[49]
def read_measurements(self) -> Optional[MeasurementResult]:
"""Read flow, temperature, and volume measurements.
Returns:
MeasurementResult object with current measurements, or None on error
"""
# Request flow, temperature, volume and bubble size
sequence = 0
data = bytes([
0x0B, # Protocol type 11
0x00, # Control register
sequence,
0x82, # Flow in ml/min (float)
0x88, # Temperature (float)
0x8B, # Volume (float)
0x2A # Bubble size (byte)
])
command = self._build_command(SonoflowCommands.CMD_GET_VALUES, data)
[Link](command)
response = self._read_response()
if not response:
return None
# Parse response
state = response[5]
error_code = 0
# Extract measurements
idx = 8
flow = [Link]('<f', response[idx:idx+4])[0]
idx += 5 # Skip type byte
temp = [Link]('<f', response[idx:idx+4])[0]
idx += 5
volume = [Link]('<f', response[idx:idx+4])[0]
idx += 5
bubble_size = response[idx] * 0.1 # Convert to mm
return MeasurementResult(
flow_ml_min=flow,
temperature_c=temp,
volume_ul=volume,
state=state,
error_code=error_code,
bubble_size=bubble_size,
bubble_detected=(state & SensorState.BUBBLE_ALARM) != 0
def close(self):
"""Close serial connection."""
[Link]()
# Example usage:
if __name__ == '__main__':
# Create sensor object
sensor = SonoflowSensor('COM6', address=1)
try:
# Test communication
if [Link]():
print("Communication OK")
# Get device info
ident = sensor.get_identification()
if ident:
print(f"\nDevice Information:")
print(f"Serial Number: {ident.serial_number}")
print(f"SW Version: {ident.software_version:04x}")
print(f"HW Version: {ident.hardware_version:04x}")
# Perform zero adjustment
print("\nPerforming zero adjustment...")
success, zero_value = sensor.zero_adjust(mode=1) # Auto mode
if success:
print(f"Zero adjustment value: {zero_value:.2f}")
# Reset volume counter
sensor.reset_volume()
print("\nVolume counter reset")
# Read measurements a few times
print("\nMeasurements:")
for i in range(5):
result = sensor.read_measurements()
if result:
print(f"\nReading {i+1}:")
print(f"Flow: {result.flow_ml_min:.1f} ml/min")
print(f"Temperature: {result.temperature_c:.1f}°C")
print(f"Volume: {result.volume_ul/1000:.3f} ml")
if result.bubble_detected:
print(f"Bubble detected! Size: {result.bubble_size:.1f} mm")
if [Link]:
print(f"State flags: 0x{[Link]:02x}")
else:
print("Error reading measurements")
[Link](1)
finally:
[Link]()