func (h *Hospital) MessagesLen() int {
	return h.messageQ.Len()
}
MessagesLen returns the number of messages waiting to be processed.
package hospital
import (
	"context"
	"time"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"google.golang.org/protobuf/encoding/prototext"
	"github.com/google/simhospital/pkg/clock"
	"github.com/google/simhospital/pkg/config"
	"github.com/google/simhospital/pkg/doctor"
	"github.com/google/simhospital/pkg/generator"
	"github.com/google/simhospital/pkg/generator/header"
	"github.com/google/simhospital/pkg/generator/id"
	"github.com/google/simhospital/pkg/generator/person"
	"github.com/google/simhospital/pkg/hardcoded"
	"github.com/google/simhospital/pkg/hl7"
	"github.com/google/simhospital/pkg/ir"
	"github.com/google/simhospital/pkg/location"
	"github.com/google/simhospital/pkg/logging"
	"github.com/google/simhospital/pkg/message"
	"github.com/google/simhospital/pkg/monitoring"
	"github.com/google/simhospital/pkg/orderprofile"
	"github.com/google/simhospital/pkg/pathway"
	"github.com/google/simhospital/pkg/processor"
	"github.com/google/simhospital/pkg/resource/cloud"
	"github.com/google/simhospital/pkg/resource"
	"github.com/google/simhospital/pkg/state/persist"
	"github.com/google/simhospital/pkg/state"
)
const (
	datetimeLayout = "2006-01-02 15:04:05"
	keyPathwayName           = "pathway_name"
	keyMessageName           = "message_name"
	keyEventType             = "event_type"
	keyEvent                 = "event"
	keyMessage               = "message"
	keyMessageType           = "message_type"
	keyLocation              = "location"
	keyNextEvent             = "next_event"
	keyNextEventType         = "next_event_type"
	keyIndex                 = "step_index"
	keyExpectedMessageTime   = "expected_message_time"
	keyExpectedEventTime     = "expected_event_time"
	keyExpectedNextEventTime = "expected_next_event_time"
	keyIsHistorical          = "is_historical"
	keyPatientID             = "mrn"
	inconsistentQueueError = "inconsistent event queue"
	unknown = "unknown"
)
var (
	log = logging.ForCallerPackage()
	defaultOrderAckDelay = &pathway.Delay{
		From: time.Second,
		To:   11 * time.Second,
	}
	counters struct {
		SimulatedHospital struct {
			PathwaysTotal            *prometheus.CounterVec   `help:"Number of pathways that were successfully started" labels:"pathway_name"`
			MessagesTotal            *prometheus.CounterVec   `help:"Number of messages sent" labels:"pathway_name,message_type,trigger_event"`
			ErrorsTotal              *prometheus.CounterVec   `help:"Number of errors" labels:"pathway_name,reason"`
			PathwayDurationMinutes   *prometheus.HistogramVec `help:"Duration (minutes) of the generated pathway, by pathway name" labels:"pathway_name" buckets:"1,5,10,30,60,180,720,1440,2880"`
			AdmissionDurationMinutes *prometheus.HistogramVec `help:"Duration (minutes) of the admissions in the generated pathways, by pathway name" labels:"pathway_name" buckets:"1,5,10,30,60,180,720,1440,2880"`
			MessageDelaySeconds      prometheus.Histogram     `help:"Difference, in seconds, between the time a message was expected to be sent, and the time when it was really sent" buckets:"1,5,10,30,60,180"`
		}
	}
)
type EventProcessor interface {
	Process(*state.Event, *ir.PatientInfo, *processor.Config) ([]*message.HL7Message, error)
	Matches(*state.Event) bool
}
type MessageProcessor interface {
	Process(*state.HL7Message) error
	Matches(*state.HL7Message) bool
}
type ResourceWriter interface {
	Generate(*ir.PatientInfo) error
	Close() error
}
type Arguments struct {
	LocationsFile *string
	HardcodedMessagesDir *string
	Hl7ConfigFile *string
	HeaderConfigFile *string
	DoctorsFile *string
	OrderProfilesFile *string
	ResourceArguments *ResourceArguments
	DeletePatientsFromMemory bool
	PathwayArguments *PathwayArguments
	SenderArguments *SenderArguments
	DataFiles *config.DataFiles
	MessageControlGenerator *header.MessageControlGenerator
	Clock clock.Clock
}
type PathwayArguments struct {
	Dir string
	Names []string
	ExcludeNames []string
	Type string
}
type SenderArguments struct {
	Output string
	OutputFile string
	MllpDestination string
	MllpKeepAlive bool
	MllpKeepAliveInterval *time.Duration
}
type ResourceArguments struct {
	Output    string
	OutputDir string
	Format    string
	CloudProjectID string
	CloudLocation  string
	CloudDataset   string
	CloudDatastore string
}
type Config struct {
	LocationManager *location.Manager
	MessageControlGenerator *header.MessageControlGenerator
	MessagesManager *hardcoded.Manager
	HL7Config *config.HL7Config
	Header *config.Header
	DataFiles config.DataFiles
	Doctors *doctor.Doctors
	OrderProfiles *orderprofile.OrderProfiles
	PathwayParser *pathway.Parser
	PathwayManager pathway.Manager
	Clock clock.Clock
	Sender hl7.Sender
	DeletePatientsFromMemory bool
	ResourceWriter ResourceWriter
	AdditionalConfig AdditionalConfig
}
type AdditionalConfig struct {
	ValidFn func(*pathway.Pathway) error
	Processors Processors
	ItemSyncers map[string]persist.ItemSyncer
	OrderAckDelay *pathway.Delay
	AddressGenerator person.AddressGenerator
	MRNGenerator id.Generator
	PlacerGenerator id.Generator
	FillerGenerator id.Generator
}
func DefaultConfig(ctx context.Context, arguments Arguments) (Config, error) {
	c := Config{
		MessageControlGenerator:  &header.MessageControlGenerator{},
		Clock:                    &clock.RealTimeClock{},
		DeletePatientsFromMemory: arguments.DeletePatientsFromMemory,
	}
	if arguments.MessageControlGenerator != nil {
		c.MessageControlGenerator = arguments.MessageControlGenerator
	}
	if arguments.Clock != nil {
		c.Clock = arguments.Clock
	}
	var err error
	if arguments.LocationsFile != nil {
		if c.LocationManager, err = location.NewManager(ctx, *arguments.LocationsFile); err != nil {
			return Config{}, errors.Wrap(err, "cannot create Location Manager")
		}
	}
	if arguments.HardcodedMessagesDir != nil {
		if c.MessagesManager, err = hardcoded.NewManager(ctx, *arguments.HardcodedMessagesDir, c.MessageControlGenerator); err != nil {
			return Config{}, errors.Wrap(err, "cannot create Hardcoded Messages Manager")
		}
	}
	if arguments.Hl7ConfigFile != nil {
		if c.HL7Config, err = config.LoadHL7Config(ctx, *arguments.Hl7ConfigFile); err != nil {
			return Config{}, errors.Wrap(err, "cannot load the message HL7 configuration")
		}
	}
	if arguments.HeaderConfigFile != nil {
		if c.Header, err = config.LoadHeaderConfig(ctx, *arguments.HeaderConfigFile); err != nil {
			return Config{}, errors.Wrap(err, "cannot load the header configuration")
		}
	}
	if arguments.DoctorsFile != nil {
		if c.Doctors, err = doctor.LoadDoctors(ctx, *arguments.DoctorsFile); err != nil {
			return Config{}, errors.Wrap(err, "cannot load the doctors configuration")
		}
	}
	if arguments.OrderProfilesFile != nil && c.HL7Config != nil {
		if c.OrderProfiles, err = orderprofile.Load(ctx, *arguments.OrderProfilesFile, c.HL7Config); err != nil {
			return Config{}, errors.Wrap(err, "cannot load the order profiles")
		}
	}
	if arguments.SenderArguments != nil {
		if c.Sender, err = hl7Sender(*arguments.SenderArguments); err != nil {
			return Config{}, errors.Wrap(err, "cannot create the sender")
		}
	}
	if arguments.ResourceArguments != nil && c.HL7Config != nil {
		if c.ResourceWriter, err = resourceWriter(ctx, *arguments.ResourceArguments, c.HL7Config); err != nil {
			return Config{}, errors.Wrap(err, "cannot create the resource writer")
		}
	}
	if c.OrderProfiles != nil && c.Doctors != nil && c.LocationManager != nil {
		c.PathwayParser = &pathway.Parser{Clock: c.Clock, OrderProfiles: c.OrderProfiles, Doctors: c.Doctors, LocationManager: c.LocationManager}
		if arguments.PathwayArguments != nil {
			if c.PathwayManager, err = pathwayManager(ctx, c.PathwayParser, *arguments.PathwayArguments); err != nil {
				return Config{}, errors.Wrap(err, "cannot create pathway manager")
			}
		}
	}
	if arguments.DataFiles != nil {
		c.DataFiles = *arguments.DataFiles
	}
	return c, nil
}
func resourceWriter(ctx context.Context, arguments ResourceArguments, hl7Config *config.HL7Config) (ResourceWriter, error) {
	output, err := resourceOutput(ctx, arguments)
	if err != nil {
		return nil, errors.Wrap(err, "cannot create resource output")
	}
	marshaller, err := resourceMarshaller(arguments)
	if err != nil {
		return nil, errors.Wrap(err, "cannot create resource marshaller")
	}
	cfg := resource.GeneratorConfig{
		HL7Config:   hl7Config,
		IDGenerator: &id.UUIDGenerator{},
		Output:      output,
		Marshaller:  marshaller,
	}
	return resource.NewFHIRWriter(cfg)
}
func resourceOutput(ctx context.Context, arguments ResourceArguments) (resource.Output, error) {
	switch arguments.Output {
	case "stdout":
		return &resource.StdOutput{}, nil
	case "file":
		return resource.NewDirectoryOutput(arguments.OutputDir)
	case "cloud":
		return cloud.NewOutput(ctx, arguments.CloudProjectID, arguments.CloudLocation, arguments.CloudDataset, arguments.CloudDatastore)
	default:
		return nil, errors.Errorf("unsupported output type %q", arguments.Output)
	}
}
func resourceMarshaller(arguments ResourceArguments) (resource.Marshaller, error) {
	switch arguments.Format {
	case "json":
		return resource.NewJSONMarshaller()
	case "proto":
		return prototext.MarshalOptions{Multiline: true, Indent: "  "}, nil
	default:
		return nil, errors.Errorf("unsupported output format %q", arguments.Format)
	}
}
func hl7Sender(arguments SenderArguments) (hl7.Sender, error) {
	switch arguments.Output {
	case "stdout":
		return hl7.NewStdoutSender(), nil
	case "mllp":
		return hl7.NewMLLPSender(arguments.MllpDestination, arguments.MllpKeepAlive, *arguments.MllpKeepAliveInterval)
	case "file":
		return hl7.NewFileSender(arguments.OutputFile)
	default:
		return nil, errors.Errorf("unsupported output type %q", arguments.Output)
	}
}
func pathwayManager(ctx context.Context, p *pathway.Parser, arguments PathwayArguments) (pathway.Manager, error) {
	pathways, err := p.ParsePathways(ctx, arguments.Dir)
	if err != nil {
		return nil, errors.Wrap(err, "failed to parse pathways for Pathway Manager")
	}
	switch arguments.Type {
	case "distribution":
		return pathway.NewDistributionManager(pathways, arguments.Names, arguments.ExcludeNames)
	case "deterministic":
		return pathway.NewDeterministicManager(pathways, arguments.Names)
	default:
		return nil, errors.Errorf("unsupported pathway manager type %q", arguments.Type)
	}
}
type Processors struct {
	EventOverride []EventProcessor
	EventPre []EventProcessor
	EventPost []EventProcessor
	MessageOverride []MessageProcessor
	MessagePre []MessageProcessor
	MessagePost []MessageProcessor
}
type Hospital struct {
	clock                   clock.Clock
	sender                  hl7.Sender
	generator               *generator.Generator
	locationManager         *location.Manager
	messageQ                *state.WrappedQueue
	eventQ                  *state.WrappedQueue
	pathwayManager          pathway.Manager
	hardcodedMessageManager *hardcoded.Manager
	patients                *state.PatientsMap
	processors              Processors
	resourceWriter          ResourceWriter
	messageConfig           *config.HL7Config
	orderAckDelay           *pathway.Delay
}
func init() {
	if err := monitoring.CreateAndRegisterMetricsFromStruct(&counters); err != nil {
		log.WithError(err).Fatal("Cannot register metrics from the 'hospital' package")
	}
}
func (h *Hospital) RunNextEventIfDue(ctx context.Context) (bool, error) {
	if !h.hasDueEvent() {
		return false, nil
	}
	err := h.runNextEvent(ctx)
	return err == nil, err
}
func (h *Hospital) ProcessNextMessageIfDue() (bool, error) {
	if !h.hasDueMessage() {
		return false, nil
	}
	err := h.processNextMessage()
	return err == nil, err
}
func (h Hospital) hasDueEvent() bool {
	i := h.eventQ.Peek()
	if i == nil {
		return false
	}
	e, ok := i.(state.Event)
	if !ok {
		log.Fatalf("Unknown item type %v, want state.Event", i)
	}
	return h.isTimeDue(e.EventTime)
}
func (h Hospital) hasDueMessage() bool {
	i := h.messageQ.Peek()
	if i == nil {
		return false
	}
	m, ok := i.(state.HL7Message)
	if !ok {
		log.Fatalf("Unknown item type %v, want state.HL7Message", i)
	}
	return h.isTimeDue(m.MessageTime)
}
func (h Hospital) isTimeDue(t time.Time) bool {
	return t.Unix() <= h.clock.Now().Unix()
}
func (h *Hospital) GetPathway(pathwayName string) (*pathway.Pathway, error) {
	return h.pathwayManager.GetPathway(pathwayName)
}
func (h *Hospital) StartNextPathway() error {
	p, err := h.pathwayManager.NextPathway()
	if err != nil {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": "unknown",
			"reason":       "get_pathway_failure",
		}).Inc()
		return errors.Wrap(err, "cannot get next pathway")
	}
	if _, err := h.StartPathway(p); err != nil {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": p.Name(),
			"reason":       "pathway_start_failure",
		}).Inc()
		return errors.Wrap(err, "cannot start pathway")
	}
	return nil
}
func (h *Hospital) StartPathway(p *pathway.Pathway) ([]*ir.Person, error) {
	logLocal := log.WithField(keyPathwayName, p.Name())
	if p.Persons == nil || len(*p.Persons) == 0 {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": p.Name(),
			"reason":       "invalid_persons_section",
		}).Inc()
		return nil, errors.New("invalid Persons section")
	}
	idsToMRN := map[pathway.PatientID]string{}
	var mbPersons []*ir.Person
	var patients []*state.Patient
	persons := *p.Persons
	i := 1
	for id, person := range persons {
		newPerson, p := h.newOrExistingPatient(&person, p.Consultant)
		patients = append(patients, p)
		mbPersons = append(mbPersons, newPerson)
		logLocal.Infof("Starting pathway, person %v: %s %s %s",
			id, newPerson.FirstName, newPerson.Surname, newPerson.MRN)
		idsToMRN[id] = newPerson.MRN
		i++
	}
	if err := h.queueFirstEvent(*p, idsToMRN, patients...); err != nil {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": p.Name(),
			"reason":       "queue_first_event",
		}).Inc()
		return nil, errors.Wrap(err, "Failed to queue first event")
	}
	counters.SimulatedHospital.PathwaysTotal.With(
		prometheus.Labels{"pathway_name": p.Name()},
	).Inc()
	return mbPersons, nil
}
func (h Hospital) newOrExistingPatient(person *pathway.Person, consultant *pathway.Consultant) (*ir.Person, *state.Patient) {
	if person.MRN != "" {
		if p := h.patients.Get(person.MRN); p != nil {
			h.generator.UpdateFromPathway(p.PatientInfo, &pathway.UpdatePerson{Person: person})
			return p.PatientInfo.Person, p
		}
	}
	return h.newPatient(person, consultant)
}
func (h Hospital) newPatient(person *pathway.Person, consultant *pathway.Consultant) (*ir.Person, *state.Patient) {
	newPerson := h.generator.NewPerson(person)
	newConsultant := h.generator.NewDoctor(consultant)
	return newPerson, h.generator.NewPatient(newPerson, newConsultant)
}
func calculateTimes(now time.Time, params *pathway.Parameters) (eventTime time.Time, msgTime time.Time) {
	eventTime = now
	msgTime = now
	if params != nil {
		if params.TimeFromNow != nil {
			eventTime = eventTime.Add(*params.TimeFromNow)
		}
		msgTime = eventTime.Add(params.DelayMessage.Random())
	}
	return
}
func NewHospital(ctx context.Context, c Config) (*Hospital, error) {
	if c.MessagesManager == nil {
		return nil, errors.New("Config.MessagesManager not provided; this is required")
	}
	if c.LocationManager == nil {
		return nil, errors.New("Config.LocationManager not provided; this is required")
	}
	if c.MessageControlGenerator == nil {
		return nil, errors.New("Config.MessageControlGenerator not provided; this is required")
	}
	if c.HL7Config == nil {
		return nil, errors.New("Config.HL7Config not provided; this is required")
	}
	if c.Header == nil {
		return nil, errors.New("Config.Header not provided; this is required")
	}
	if c.Doctors == nil {
		return nil, errors.New("Config.Doctors not provided; this is required")
	}
	if c.OrderProfiles == nil {
		return nil, errors.New("Config.OrderProfiles not provided; this is required")
	}
	if c.Sender == nil {
		return nil, errors.New("Config.Sender not provided; this is required")
	}
	if c.ResourceWriter == nil {
		return nil, errors.New("Config.ResourceWriter not provided; this is required")
	}
	if c.PathwayManager == nil {
		return nil, errors.New("Config.PathwayManager not provided; this is required")
	}
	if c.Clock == nil {
		return nil, errors.New("Config.Clock not provided; this is required")
	}
	ac := c.AdditionalConfig
	dataConfig, err := config.LoadData(ctx, c.DataFiles, c.HL7Config)
	if err != nil {
		return nil, errors.Wrap(err, "Failed to load the message data configuration")
	}
	genConfig := generator.Config{
		Clock:            c.Clock,
		HL7Config:        c.HL7Config,
		Header:           c.Header,
		Data:             dataConfig,
		Doctors:          c.Doctors,
		MsgCtrlGenerator: c.MessageControlGenerator,
		OrderProfiles:    c.OrderProfiles,
		AddressGenerator: ac.AddressGenerator,
		MRNGenerator:     ac.MRNGenerator,
		PlacerGenerator:  ac.PlacerGenerator,
		FillerGenerator:  ac.FillerGenerator,
	}
	messageQ := newMessageQueue(ac.ItemSyncers[state.MessageItemType])
	eventQ := newEventQueue(ac.ItemSyncers[state.EventItemType])
	patientsMap := state.NewPatientsMap(ac.ItemSyncers[state.PatientItemType], c.DeletePatientsFromMemory)
	if ac.OrderAckDelay == nil {
		ac.OrderAckDelay = defaultOrderAckDelay
	}
	return &Hospital{
		clock:                   c.Clock,
		sender:                  c.Sender,
		generator:               generator.NewGenerator(genConfig),
		locationManager:         c.LocationManager,
		messageQ:                messageQ,
		eventQ:                  eventQ,
		pathwayManager:          c.PathwayManager,
		hardcodedMessageManager: c.MessagesManager,
		patients:                patientsMap,
		processors:              c.AdditionalConfig.Processors,
		resourceWriter:          c.ResourceWriter,
		messageConfig:           c.HL7Config,
		orderAckDelay:           ac.OrderAckDelay,
	}, nil
}
func (h *Hospital) Close() error {
	if err := h.sender.Close(); err != nil {
		return errors.Wrap(err, "error closing sender")
	}
	if err := h.resourceWriter.Close(); err != nil {
		return errors.Wrap(err, "error closing resource writer")
	}
	return nil
}
func newMessageQueue(syncer persist.ItemSyncer) *state.WrappedQueue {
	messageQ, err := state.NewWrappedQueue(state.MessageItemType, syncer)
	if err != nil {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": unknown,
			"reason":       "cannot load messages",
		}).Inc()
		log.WithError(err).Error("Cannot load Messages from syncer")
	}
	return messageQ
}
func newEventQueue(syncer persist.ItemSyncer) *state.WrappedQueue {
	eventQ, err := state.NewWrappedQueue(state.EventItemType, syncer)
	if err != nil {
		counters.SimulatedHospital.ErrorsTotal.With(prometheus.Labels{
			"pathway_name": unknown,
			"reason":       "cannot load messages",
		}).Inc()
		log.WithError(err).Error("Cannot load Events from syncer")
	}
	return eventQ
}
func (h *Hospital) PatientsLen() int {
	return h.patients.Len()
}
func (h *Hospital) EventsLen() int {
	return h.eventQ.Len()
}