Add shaft-torque telemetry to the drive sample record
Python · Python · advanced · modification
Adds shaft-torque telemetry to the drive sample for the new HD-series drives: a signed int16 field appended after the winding-temperature field, decoder updated to match. Same compact fixed-point encoding the temperature field already uses, so a sample stays one contiguous record. Round-trip verified against a bench drive.
The stream feeds the plant's drive-analytics tier, where decoded torque drives the regen-braking dashboards and per-shift energy reports. The line is being upgraded from the pilot servo drives (rated to +/-300 Nm) to HD-series drives rated to +/-600 Nm.
Requirements
- Extend the telemetry sample to carry shaft torque: the new HD-series drives report integer torque from -60,000 to +60,000 centi-newton-metres (cNm), where negative is regenerative braking, and decode_sample(encode_sample(s)) must return every in-spec reading unchanged.
- Keep the existing fields (drive_id, uptime_ms, temp_cdeg) packed exactly as before and append torque after the winding-temperature field.
- Input contracts, enforced upstream before encoding, which encode_sample may assume: drive_id is an integer 0-65,535; uptime_ms is an integer 0-4,294,967,295; temp_cdeg (winding temperature) is an integer within -4,000..+18,000 (that is -40..+180 C); torque_cnm is an integer within the HD-series spec range.
- The record stays fixed-layout big-endian via struct, packed as one contiguous buffer with no per-sample object overhead, and encode_sample / decode_sample stay exact inverses.
Files touched
- app/telemetry/drive_sample.py
--- app/telemetry/drive_sample.py
them straight back into DriveSample objects.
-Sample layout (big-endian, 8 bytes):
+Sample layout (big-endian, 10 bytes):
offset 0 uint16 drive_id # per-drive id on the line
offset 2 uint32 uptime_ms # milliseconds since the drive powered on
offset 6 int16 temp_cdeg # winding temperature, centidegrees C
+ offset 8 int16 torque_cnm # shaft torque, centi-newton-metres; negative = regen braking
"""
uptime_ms: int # uint32, milliseconds since the drive powered on
temp_cdeg: int # winding temperature, centidegrees C (-40..+180 C)
+ torque_cnm: int # shaft torque, centi-newton-metres; negative = regen braking
-# Fields: drive_id (uint16), uptime_ms (uint32), temp_cdeg (int16).
+# Original fields: drive_id (uint16), uptime_ms (uint32), temp_cdeg (int16).
_SAMPLE = struct.Struct(">HIh")
def encode_sample(sample):
- """Pack one DriveSample into an 8-byte telemetry record.
+ """Pack one DriveSample into a 10-byte telemetry record.
Every field is validated upstream against the drive's published ranges
before it reaches this codec, so the packer may assume them.
"""
- return _SAMPLE.pack(sample.drive_id, sample.uptime_ms, sample.temp_cdeg)
+ head = _SAMPLE.pack(sample.drive_id, sample.uptime_ms, sample.temp_cdeg)