modules: add vban midi support

src/modules/module-vban/midi.c

@@ -0,0 +1,327 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+static void vban_midi_process_playback(void *data)
+{
+	struct impl *impl = data;
+	struct pw_buffer *buf;
+	struct spa_data *d;
+	uint32_t timestamp, duration, maxsize, read;
+	struct spa_pod_builder b;
+	struct spa_pod_frame f[1];
+	void *ptr;
+	struct spa_pod *pod;
+	struct spa_pod_control *c;
+
+	if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
+		pw_log_debug("Out of stream buffers: %m");
+		return;
+	}
+	d = buf->buffer->datas;
+
+	maxsize = d[0].maxsize;
+
+	/* we always use the graph position to select events */
+	if (impl->io_position) {
+		duration = impl->io_position->clock.duration;
+		timestamp = impl->io_position->clock.position;
+	} else {
+		duration = 8192;
+		timestamp = 0;
+	}
+
+	/* we copy events into the buffer as they are available. */
+	spa_pod_builder_init(&b, d[0].data, maxsize);
+	spa_pod_builder_push_sequence(&b, &f[0], 0);
+
+	while (true) {
+		int32_t avail = spa_ringbuffer_get_read_index(&impl->ring, &read);
+		if (avail <= 0)
+			break;
+
+		ptr = SPA_PTROFF(impl->buffer, read & BUFFER_MASK2, void);
+
+		if ((pod = spa_pod_from_data(ptr, avail, 0, avail)) == NULL)
+			goto done;
+		if (!spa_pod_is_sequence(pod))
+			goto done;
+
+		/* the ringbuffer contains series of sequences, one for each
+		 * received packet */
+		SPA_POD_SEQUENCE_FOREACH((struct spa_pod_sequence*)pod, c) {
+#if 0
+			/* try to render with given delay */
+			uint32_t target = c->offset + impl->target_buffer;
+			target = (uint64_t)target * rate / impl->rate;
+#else
+			uint32_t target = timestamp;
+#endif
+			if (timestamp != 0) {
+				/* skip old packets */
+				if (target < timestamp)
+					continue;
+				/* event for next cycle */
+				if (target >= timestamp + duration)
+					goto complete;
+			} else {
+				timestamp = target;
+			}
+			spa_pod_builder_control(&b, target - timestamp, SPA_CONTROL_Midi);
+			spa_pod_builder_bytes(&b,
+					SPA_POD_BODY(&c->value),
+					SPA_POD_BODY_SIZE(&c->value));
+		}
+		/* we completed a sequence (one RTP packet), advance ringbuffer
+		 * and go to the next packet */
+		read += SPA_PTRDIFF(c, ptr);
+		spa_ringbuffer_read_update(&impl->ring, read);
+	}
+complete:
+	spa_pod_builder_pop(&b, &f[0]);
+
+	if (b.state.offset > maxsize) {
+		pw_log_warn("overflow buffer %u %u", b.state.offset, maxsize);
+		b.state.offset = 0;
+	}
+	d[0].chunk->size = b.state.offset;
+	d[0].chunk->stride = 1;
+	d[0].chunk->offset = 0;
+done:
+	pw_stream_queue_buffer(impl->stream, buf);
+}
+
+static int parse_varlen(uint8_t *p, uint32_t avail, uint32_t *result)
+{
+	uint32_t value = 0, offs = 0;
+	while (offs < avail) {
+		uint8_t b = p[offs++];
+		value = (value << 7) | (b & 0x7f);
+		if ((b & 0x80) == 0)
+			break;
+	}
+	*result = value;
+	return offs;
+}
+
+static int get_midi_size(uint8_t *p, uint32_t avail)
+{
+	int size;
+	uint32_t offs = 0, value;
+
+	switch (p[offs++]) {
+	case 0xc0 ... 0xdf:
+		size = 2;
+		break;
+	case 0x80 ... 0xbf:
+	case 0xe0 ... 0xef:
+		size = 3;
+		break;
+	case 0xff:
+	case 0xf0:
+	case 0xf7:
+		size = parse_varlen(&p[offs], avail - offs, &value);
+		size += value + 1;
+		break;
+	default:
+		return -EINVAL;
+	}
+	return size;
+}
+
+static int vban_midi_receive_midi(struct impl *impl, uint8_t *packet,
+		uint32_t payload_offset, uint32_t plen)
+{
+	uint32_t write;
+	int32_t filled;
+	struct spa_pod_builder b;
+	struct spa_pod_frame f[1];
+	void *ptr;
+	uint32_t offs = payload_offset;
+	uint32_t timestamp = 0;
+
+	/* no sync, resync */
+	if (!impl->have_sync) {
+		pw_log_info("sync to timestamp:%u", timestamp);
+		impl->have_sync = true;
+		impl->ring.readindex = impl->ring.writeindex;
+	}
+
+	filled = spa_ringbuffer_get_write_index(&impl->ring, &write);
+	if (filled > (int32_t)BUFFER_SIZE2) {
+		pw_log_warn("overflow");
+		return -ENOSPC;
+	}
+
+	ptr = SPA_PTROFF(impl->buffer, write & BUFFER_MASK2, void);
+
+	/* each packet is written as a sequence of events. The offset is
+	 * the receive timestamp */
+	spa_pod_builder_init(&b, ptr, BUFFER_SIZE2 - filled);
+	spa_pod_builder_push_sequence(&b, &f[0], 0);
+
+	while (offs < plen) {
+		int size;
+
+		spa_pod_builder_control(&b, timestamp, SPA_CONTROL_Midi);
+
+		size = get_midi_size(&packet[offs], plen - offs);
+
+		if (size <= 0 || offs + size > plen) {
+			pw_log_warn("invalid size (%08x) %d (%u %u)",
+					packet[offs], size, offs, plen);
+			break;
+		}
+
+		spa_pod_builder_bytes(&b, &packet[offs], size);
+
+		offs += size;
+	}
+	spa_pod_builder_pop(&b, &f[0]);
+
+	write += b.state.offset;
+	spa_ringbuffer_write_update(&impl->ring, write);
+
+	return 0;
+}
+
+static int vban_midi_receive(struct impl *impl, uint8_t *buffer, ssize_t len)
+{
+	struct vban_header *hdr;
+	ssize_t hlen;
+	uint32_t n_frames;
+
+	if (len < VBAN_HEADER_SIZE)
+		goto short_packet;
+
+	hdr = (struct vban_header*)buffer;
+	if (strncmp(hdr->vban, "VBAN", 3))
+		goto invalid_version;
+
+	hlen = VBAN_HEADER_SIZE;
+
+	n_frames = hdr->n_frames;
+	if (impl->have_sync && impl->n_frames != n_frames) {
+		pw_log_info("unexpected frame (%d != %d)",
+				n_frames, impl->n_frames);
+		impl->have_sync = false;
+	}
+	impl->n_frames = n_frames + 1;
+
+	impl->receiving = true;
+
+	return vban_midi_receive_midi(impl, buffer, hlen, len);
+
+short_packet:
+	pw_log_warn("short packet received");
+	return -EINVAL;
+invalid_version:
+	pw_log_warn("invalid RTP version");
+	spa_debug_mem(0, buffer, len);
+	return -EPROTO;
+}
+
+static void vban_midi_flush_packets(struct impl *impl,
+		struct spa_pod_sequence *sequence, uint32_t timestamp, uint32_t rate)
+{
+	struct spa_pod_control *c;
+	struct vban_header header;
+	struct iovec iov[2];
+	uint32_t len;
+
+	header = impl->header;
+
+	iov[0].iov_base = &header;
+	iov[0].iov_len = sizeof(header);
+	iov[1].iov_base = impl->buffer;
+	iov[1].iov_len = 0;
+
+	len = 0;
+
+	SPA_POD_SEQUENCE_FOREACH(sequence, c) {
+		void *ev;
+		uint32_t size;
+
+		if (c->type != SPA_CONTROL_Midi)
+			continue;
+
+		ev = SPA_POD_BODY(&c->value),
+                size = SPA_POD_BODY_SIZE(&c->value);
+		if (len == 0) {
+			/* start new packet */
+			header.n_frames++;
+		} else if (len + size > impl->mtu) {
+			/* flush packet when we have one and when it's too large */
+			iov[1].iov_len = len;
+
+			pw_log_debug("sending %d", len);
+			vban_stream_emit_send_packet(impl, iov, 2);
+			len = 0;
+		}
+		memcpy(&impl->buffer[len], ev, size);
+		len += size;
+	}
+	if (len > 0) {
+		/* flush last packet */
+		iov[1].iov_len = len;
+
+		pw_log_debug("sending %d", len);
+		vban_stream_emit_send_packet(impl, iov, 2);
+	}
+	impl->header.n_frames = header.n_frames;
+}
+
+static void vban_midi_process_capture(void *data)
+{
+	struct impl *impl = data;
+	struct pw_buffer *buf;
+	struct spa_data *d;
+	uint32_t offs, size, timestamp, rate;
+	struct spa_pod *pod;
+	void *ptr;
+
+	if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
+		pw_log_debug("Out of stream buffers: %m");
+		return;
+	}
+	d = buf->buffer->datas;
+
+	offs = SPA_MIN(d[0].chunk->offset, d[0].maxsize);
+	size = SPA_MIN(d[0].chunk->size, d[0].maxsize - offs);
+
+	if (SPA_LIKELY(impl->io_position)) {
+		rate = impl->io_position->clock.rate.denom;
+		timestamp = impl->io_position->clock.position * impl->rate / rate;
+	} else {
+		rate = 10000;
+		timestamp = 0;
+	}
+
+	ptr = SPA_PTROFF(d[0].data, offs, void);
+
+	if ((pod = spa_pod_from_data(ptr, size, 0, size)) == NULL)
+		goto done;
+	if (!spa_pod_is_sequence(pod))
+		goto done;
+
+	if (!impl->have_sync) {
+		pw_log_info("sync to timestamp:%u n_frames:%u",
+				timestamp, impl->n_frames);
+		impl->have_sync = true;
+	}
+
+	vban_midi_flush_packets(impl, (struct spa_pod_sequence*)pod, timestamp, rate);
+
+done:
+	pw_stream_queue_buffer(impl->stream, buf);
+}
+
+static int vban_midi_init(struct impl *impl, enum spa_direction direction)
+{
+	if (direction == SPA_DIRECTION_INPUT)
+		impl->stream_events.process = vban_midi_process_capture;
+	else
+		impl->stream_events.process = vban_midi_process_playback;
+	impl->receive_vban = vban_midi_receive;
+	return 0;
+}

src/modules/module-vban/stream.c

@@ -81,7 +81,7 @@ struct impl {
 };
 
 #include "module-vban/audio.c"
-//#include "module-vban/midi.c"
+#include "module-vban/midi.c"
 
 struct format_info {
 	uint32_t media_subtype;
@@ -97,7 +97,7 @@ static const struct format_info audio_format_info[] = {
 	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S32_LE, 4, VBAN_DATATYPE_INT32, },
 	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F32_LE, 4, VBAN_DATATYPE_FLOAT32, },
 	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F64_LE, 8, VBAN_DATATYPE_FLOAT64, },
-	{ SPA_MEDIA_SUBTYPE_control, 0, 1, },
+	{ SPA_MEDIA_SUBTYPE_control, 0, 1, VBAN_SERIAL_MIDI | VBAN_DATATYPE_U8, },
 };
 
 static void stream_io_changed(void *data, uint32_t id, void *area, uint32_t size)
@@ -289,9 +289,6 @@ struct vban_stream *vban_stream_new(struct pw_core *core,
 		goto out;
 	}
 	memcpy(impl->header.vban, "VBAN", 4);
-	if ((str = pw_properties_get(props, "sess.name")) == NULL)
-		str = "Stream1";
-	strcpy(impl->header.stream_name, str);
 
 	switch (impl->info.media_subtype) {
 	case SPA_MEDIA_SUBTYPE_raw:
@@ -307,6 +304,7 @@ struct vban_stream *vban_stream_new(struct pw_core *core,
 		}
 		impl->stride = impl->format_info->size * impl->stream_info.info.raw.channels;
 		impl->rate = impl->stream_info.info.raw.rate;
+
 		impl->header.format_SR = vban_sr_index(impl->rate);
 		if (impl->header.format_SR == VBAN_SR_MAXNUMBER) {
 			pw_log_error("unsupported audio rate:%u", impl->rate);
@@ -314,6 +312,9 @@ struct vban_stream *vban_stream_new(struct pw_core *core,
 			goto out;
 		}
 		impl->header.format_bit = impl->format_info->format_bit;
+		if ((str = pw_properties_get(props, "sess.name")) == NULL)
+			str = "Stream1";
+		strcpy(impl->header.stream_name, str);
 		break;
 	case SPA_MEDIA_SUBTYPE_control:
 		impl->stream_info = impl->info;
@@ -327,6 +328,14 @@ struct vban_stream *vban_stream_new(struct pw_core *core,
 		impl->rate = pw_properties_get_uint32(props, "midi.rate", 10000);
 		if (impl->rate == 0)
 			impl->rate = 10000;
+
+		impl->header.format_SR = (0x1 << 5) | 14; /* 115200 */
+		impl->header.format_nbs = 0;
+		impl->header.format_nbc = 0;
+		impl->header.format_bit = impl->format_info->format_bit;
+		if ((str = pw_properties_get(props, "sess.name")) == NULL)
+			str = "Midi1";
+		strcpy(impl->header.stream_name, str);
 		break;
 	default:
 		spa_assert_not_reached();
@@ -413,7 +422,7 @@ struct vban_stream *vban_stream_new(struct pw_core *core,
                                 SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat,
                                 SPA_FORMAT_mediaType,           SPA_POD_Id(SPA_MEDIA_TYPE_application),
                                 SPA_FORMAT_mediaSubtype,        SPA_POD_Id(SPA_MEDIA_SUBTYPE_control));
-//		vban_midi_init(impl, direction);
+		vban_midi_init(impl, direction);
 		break;
 	default:
 		res = -EINVAL;

src/modules/module-vban/vban.h

@@ -53,6 +53,10 @@ static inline uint8_t vban_sr_index(uint32_t rate)
 #define VBAN_DATATYPE_12BITS	0x06
 #define VBAN_DATATYPE_10BITS	0x07
 
+#define VBAN_SERIAL_GENERIC	0x00
+#define VBAN_SERIAL_MIDI	0x10
+#define VBAN_SERIAL_USER	0xf0
+
 #ifdef __cplusplus
 }
 #endif